<commit_before># -*- encoding: utf-8 -*-
# Copyright (c) 2010-2014, GEM Foundation.
#
# OpenQuake is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# OpenQuake is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with OpenQuake.  If not, see <http://www.gnu.org/licenses/>.

import getpass
import unittest
import mock

import numpy

from nose.plugins.attrib import attr

from django.contrib.gis.db import models as djm

from openquake.hazardlib.geo.mesh import Mesh
from openquake.hazardlib.geo.point import Point
from openquake.hazardlib.geo.surface.planar import PlanarSurface
from openquake.hazardlib.geo.surface.simple_fault import SimpleFaultSurface

from openquake.engine.calculators.hazard.classical import core as cls_core
from openquake.engine.calculators.hazard.scenario import core as scen_core
from openquake.engine.db import models

from openquake.engine.tests.utils import helpers


class ProbabilisticRuptureTestCase(unittest.TestCase):

    @classmethod
    def setUpClass(self):
        cfg = helpers.get_data_path('simple_fault_demo_hazard/job.ini')
        job = helpers.get_job(cfg)

        lt_model = models.LtSourceModel.objects.create(
            hazard_calculation=job, ordinal=0,
            sm_lt_path='foo')
        lt_rlz = models.LtRealization.objects.create(
            lt_model=lt_model, ordinal=0, gsim_lt_path='bar', weight=1)
        output = models.Output.objects.create(
            oq_job=job, display_name='test', output_type='ses')
        ses_coll = models.SESCollection.objects.create(
            output=output, lt_model=lt_rlz.lt_model, ordinal=0)

        self.mesh_lons = numpy.array(
            [0.1 * x for x in range(16)]).reshape((4, 4))
        self.mesh_lats = numpy.array(
            [0.2 * x for x in range(16)]).reshape((4, 4))
        self.mesh_depths = numpy.array(
            [0.3 * x for x in range(16)]).reshape((4, 4))

        sfs = SimpleFaultSurface(
            Mesh(self.mesh_lons, self.mesh_lats, self.mesh_depths))

        ps = PlanarSurface(
            10, 20, 30,
            Point(3.9, 2.2, 10), Point(4.90402718, 3.19634248, 10),
            Point(5.9, 2.2, 90), Point(4.89746275, 1.20365263, 90))

        trt = 'Active Shallow Crust'
        trt_model = models.TrtModel.objects.create(
            lt_model=lt_model,
            tectonic_region_type=trt,
            num_sources=0,
            num_ruptures=1,
            min_mag=5,
            max_mag=5,
            gsims=['testGSIM'])
        self.fault_rupture = models.ProbabilisticRupture.objects.create(
            ses_collection=ses_coll, magnitude=5, rake=0, surface=sfs,
            trt_model=trt_model, is_from_fault_source=True,
            is_multi_surface=False)
        self.source_rupture = models.ProbabilisticRupture.objects.create(
            ses_collection=ses_coll, magnitude=5, rake=0, surface=ps,
            trt_model=trt_model, is_from_fault_source=False,
            is_multi_surface=False)

    def test_fault_rupture(self):
        # Test loading a fault rupture from the DB, just to illustrate a use
        # case.
        # Also, we should that planar surface corner points are not valid and
        # are more or less disregarded for this type of rupture.
        fault_rupture = models.ProbabilisticRupture.objects.get(
            id=self.fault_rupture.id)
        self.assertIs(None, fault_rupture.top_left_corner)
        self.assertIs(None, fault_rupture.top_right_corner)
        self.assertIs(None, fault_rupture.bottom_right_corner)
        self.assertIs(None, fault_rupture.bottom_left_corner)

    def test_source_rupture(self):
        source_rupture = models.ProbabilisticRupture.objects.get(
            id=self.source_rupture.id)
        self.assertEqual((3.9, 2.2, 10.), source_rupture.top_left_corner)
        self.assertEqual((4.90402718, 3.19634248, 10.0),
                         source_rupture.top_right_corner)
        self.assertEqual((4.89746275, 1.20365263, 90.0),
                         source_rupture.bottom_left_corner)
        self.assertEqual((5.9, 2.2, 90.0), source_rupture.bottom_right_corner)


class FloatFieldTestCase(unittest.TestCase):
    def test_truncate_small_numbers(self):
        # workaround a postgres error "out of range for type double precision"
        self.assertEqual(djm.FloatField().get_prep_value(1e-301), 0)


class GetSiteCollectionTestCase(unittest.TestCase):

    @attr('slow')
    def test_get_site_collection_with_site_model(self):
        cfg = helpers.get_data_path(
            'simple_fault_demo_hazard/job_with_site_model.ini')
        job = helpers.get_job(cfg)
        models.JobStats.objects.create(oq_job=job)
        calc = cls_core.ClassicalHazardCalculator(job)

        # Bootstrap the `hazard_site` table:
        calc.initialize_site_collection()
        calc.initialize_sources()

        site_coll = job.hazard_calculation.site_collection
        # Since we're using a pretty big site model, it's a bit excessive to
        # check each and every value.
        # Instead, we'll just test that the lenth of each site collection attr
        # is equal to the number of points of interest in the calculation.
        expected_len = len(site_coll)

        self.assertEqual(expected_len, len(site_coll.vs30))
        self.assertEqual(expected_len, len(site_coll.vs30measured))
        self.assertEqual(expected_len, len(site_coll.z1pt0))
        self.assertEqual(expected_len, len(site_coll.z2pt5))

    def test_site_collection_and_ses_collection(self):
        cfg = helpers.get_data_path('scenario_hazard/job.ini')
        job = helpers.get_job(cfg, username=getpass.getuser())
        models.JobStats.objects.create(oq_job=job)

        calc = scen_core.ScenarioHazardCalculator(job)
        calc.initialize_site_collection()
        site_coll = job.hazard_calculation.site_collection

        # all of the parameters should be the same:
        self.assertTrue((site_coll.vs30 == 760).all())
        self.assertTrue((site_coll.vs30measured).all())
        self.assertTrue((site_coll.z1pt0 == 100).all())
        self.assertTrue((site_coll.z2pt5 == 5).all())

        # test SESCollection
        calc.initialize_sources()<commit_msg>Fixed models_test.py<commit_after># -*- encoding: utf-8 -*-
# Copyright (c) 2010-2014, GEM Foundation.
#
# OpenQuake is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# OpenQuake is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with OpenQuake.  If not, see <http://www.gnu.org/licenses/>.

import getpass
import unittest
import mock

import numpy

from nose.plugins.attrib import attr

from django.contrib.gis.db import models as djm

from openquake.hazardlib.geo.mesh import Mesh
from openquake.hazardlib.geo.point import Point
from openquake.hazardlib.geo.surface.planar import PlanarSurface
from openquake.hazardlib.geo.surface.simple_fault import SimpleFaultSurface

from openquake.engine.calculators.hazard.classical import core as cls_core
from openquake.engine.calculators.hazard.scenario import core as scen_core
from openquake.engine.db import models

from openquake.engine.tests.utils import helpers


class ProbabilisticRuptureTestCase(unittest.TestCase):

    @classmethod
    def setUpClass(self):
        cfg = helpers.get_data_path('simple_fault_demo_hazard/job.ini')
        job = helpers.get_job(cfg)

        lt_model = models.LtSourceModel.objects.create(
            hazard_calculation=job, ordinal=0,
            sm_lt_path='foo')
        lt_rlz = models.LtRealization.objects.create(
            lt_model=lt_model, ordinal=0, gsim_lt_path='bar', weight=1)
        output = models.Output.objects.create(
            oq_job=job, display_name='test', output_type='ses')
        ses_coll = models.SESCollection.objects.create(
            output=output, lt_model=lt_rlz.lt_model, ordinal=0)

        self.mesh_lons = numpy.array(
            [0.1 * x for x in range(16)]).reshape((4, 4))
        self.mesh_lats = numpy.array(
            [0.2 * x for x in range(16)]).reshape((4, 4))
        self.mesh_depths = numpy.array(
            [0.3 * x for x in range(16)]).reshape((4, 4))

        sfs = SimpleFaultSurface(
            Mesh(self.mesh_lons, self.mesh_lats, self.mesh_depths))

        ps = PlanarSurface(
            10, 20, 30,
            Point(3.9, 2.2, 10), Point(4.90402718, 3.19634248, 10),
            Point(5.9, 2.2, 90), Point(4.89746275, 1.20365263, 90))

        trt = 'Active Shallow Crust'
        trt_model = models.TrtModel.objects.create(
            lt_model=lt_model,
            tectonic_region_type=trt,
            num_sources=0,
            num_ruptures=1,
            min_mag=5,
            max_mag=5,
            gsims=['testGSIM'])
        self.fault_rupture = models.ProbabilisticRupture.objects.create(
            ses_collection=ses_coll, magnitude=5, rake=0, surface=sfs,
            trt_model=trt_model, is_from_fault_source=True,
            is_multi_surface=False)
        self.source_rupture = models.ProbabilisticRupture.objects.create(
            ses_collection=ses_coll, magnitude=5, rake=0, surface=ps,
            trt_model=trt_model, is_from_fault_source=False,
            is_multi_surface=False)

    def test_fault_rupture(self):
        # Test loading a fault rupture from the DB, just to illustrate a use
        # case.
        # Also, we should that planar surface corner points are not valid and
        # are more or less disregarded for this type of rupture.
        fault_rupture = models.ProbabilisticRupture.objects.get(
            id=self.fault_rupture.id)
        self.assertIs(None, fault_rupture.top_left_corner)
        self.assertIs(None, fault_rupture.top_right_corner)
        self.assertIs(None, fault_rupture.bottom_right_corner)
        self.assertIs(None, fault_rupture.bottom_left_corner)

    def test_source_rupture(self):
        source_rupture = models.ProbabilisticRupture.objects.get(
            id=self.source_rupture.id)
        self.assertEqual((3.9, 2.2, 10.), source_rupture.top_left_corner)
        self.assertEqual((4.90402718, 3.19634248, 10.0),
                         source_rupture.top_right_corner)
        self.assertEqual((4.89746275, 1.20365263, 90.0),
                         source_rupture.bottom_left_corner)
        self.assertEqual((5.9, 2.2, 90.0), source_rupture.bottom_right_corner)


class FloatFieldTestCase(unittest.TestCase):
    def test_truncate_small_numbers(self):
        # workaround a postgres error "out of range for type double precision"
        self.assertEqual(djm.FloatField().get_prep_value(1e-301), 0)


class GetSiteCollectionTestCase(unittest.TestCase):

    @attr('slow')
    def test_get_site_collection_with_site_model(self):
        cfg = helpers.get_data_path(
            'simple_fault_demo_hazard/job_with_site_model.ini')
        job = helpers.get_job(cfg)
        models.JobStats.objects.create(oq_job=job)
        calc = cls_core.ClassicalHazardCalculator(job)

        # Bootstrap the `hazard_site` table:
        calc.initialize_site_collection()
        calc.initialize_sources()

        site_coll = calc.site_collection
        # Since we're using a pretty big site model, it's a bit excessive to
        # check each and every value.
        # Instead, we'll just test that the lenth of each site collection attr
        # is equal to the number of points of interest in the calculation.
        expected_len = len(site_coll)

        self.assertEqual(expected_len, len(site_coll.vs30))
        self.assertEqual(expected_len, len(site_coll.vs30measured))
        self.assertEqual(expected_len, len(site_coll.z1pt0))
        self.assertEqual(expected_len, len(site_coll.z2pt5))

    def test_site_collection_and_ses_collection(self):
        cfg = helpers.get_data_path('scenario_hazard/job.ini')
        job = helpers.get_job(cfg, username=getpass.getuser())
        models.JobStats.objects.create(oq_job=job)

        calc = scen_core.ScenarioHazardCalculator(job)
        calc.initialize_site_collection()
        site_coll = calc.site_collection

        # all of the parameters should be the same:
        self.assertTrue((site_coll.vs30 == 760).all())
        self.assertTrue((site_coll.vs30measured).all())
        self.assertTrue((site_coll.z1pt0 == 100).all())
        self.assertTrue((site_coll.z2pt5 == 5).all())

        # test SESCollection
        calc.initialize_sources()
{'fraction_non_alphanumeric': 0.09284581098436201, 'fraction_numerical': 0.027347586311097682, 'mean_word_length': 3.003023888720895, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2965552', 'n_tokens_mistral': 4448, 'n_tokens_neox': 4083, 'n_words': 922}