<commit_before># The Hazard Library
# Copyright (C) 2012-2016 GEM Foundation
#
# This program 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.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
import mock
import unittest
import warnings

import numpy

from openquake.hazardlib.calc import disagg
from openquake.hazardlib.calc import filters
from openquake.hazardlib.tom import PoissonTOM
from openquake.hazardlib.geo import Point, Mesh
from openquake.hazardlib.site import Site
from openquake.hazardlib.gsim.base import ContextMaker


class _BaseDisaggTestCase(unittest.TestCase):
    class FakeSurface(object):
        def __init__(self, distance, lon, lat):
            self.distance = distance
            self.lon = lon
            self.lat = lat

        def get_joyner_boore_distance(self, sites):
            assert len(sites) == 1
            return numpy.array([self.distance], float)

        def get_closest_points(self, sites):
            assert len(sites) == 1
            return Mesh(numpy.array([self.lon], float),
                        numpy.array([self.lat], float),
                        depths=None)

    class FakeRupture(object):
        def __init__(self, mag, probability, distance, lon, lat):
            self.mag = mag
            self.probability = probability
            self.surface = _BaseDisaggTestCase.FakeSurface(distance, lon, lat)

        def get_probability_no_exceedance(self, poe):
            return (1 - self.probability) ** poe

    class FakeSource(object):
        def __init__(self, source_id, ruptures, tom, tectonic_region_type):
            self.source_id = source_id
            self.ruptures = ruptures
            self.tom = tom
            self.tectonic_region_type = tectonic_region_type

        def iter_ruptures(self):
            return iter(self.ruptures)

    class FailSource(FakeSource):
        def iter_ruptures(self):
            raise ValueError('Something bad happened')

    class FakeGSIM(object):
        REQUIRES_DISTANCES = set()
        REQUIRES_RUPTURE_PARAMETERS = set()
        REQUIRES_SITES_PARAMETERS = set()

        def __init__(self, iml, imt, truncation_level, n_epsilons,
                     disaggregated_poes):
            self.disaggregated_poes = disaggregated_poes
            self.n_epsilons = n_epsilons
            self.iml = iml
            self.imt = imt
            self.truncation_level = truncation_level
            self.dists = object()

        def disaggregate_poe(self, sctx, rctx, dctx, imt, iml,
                             truncation_level, n_epsilons):
            assert truncation_level is self.truncation_level
            assert imt is self.imt
            assert iml is self.iml
            assert n_epsilons is self.n_epsilons
            assert len(sctx) == 1
            return numpy.array([self.disaggregated_poes[rctx]])

    def setUp(self):
        self.ruptures_and_poes1 = [
            ([0, 0, 0], self.FakeRupture(5, 0.1, 3, 22, 44)),
            ([0.1, 0.2, 0.1], self.FakeRupture(5, 0.2, 11, 22, 44)),
            ([0, 0, 0.3], self.FakeRupture(5, 0.01, 12, 22, 45)),
            ([0, 0.05, 0.001], self.FakeRupture(5, 0.33, 13, 22, 45)),
            ([0, 0, 0], self.FakeRupture(9, 0.4, 14, 21, 44)),
            ([0, 0, 0.02], self.FakeRupture(5, 0.05, 11, 21, 44)),
            ([0.04, 0.1, 0.04], self.FakeRupture(5, 0.53, 11, 21, 45)),
            ([0.2, 0.3, 0.2], self.FakeRupture(5, 0.066, 10, 21, 45)),
            ([0.3, 0.4, 0.3], self.FakeRupture(6, 0.1, 12, 22, 44)),
            ([0, 0, 0.1], self.FakeRupture(6, 0.1, 12, 21, 44)),
            ([0, 0, 0], self.FakeRupture(6, 0.1, 11, 22, 45)),
        ]
        self.ruptures_and_poes2 = [
            ([0, 0.1, 0.04], self.FakeRupture(8, 0.04, 5, 11, 45)),
            ([0.1, 0.5, 0.1], self.FakeRupture(7, 0.03, 5, 11, 46))
        ]
        self.time_span = 10
        self.tom = PoissonTOM(time_span=10)
        self.source1 = self.FakeSource(
            1, [rupture for poes, rupture in self.ruptures_and_poes1],
            self.tom, 'trt1'
        )
        self.source2 = self.FakeSource(
            2, [rupture for poes, rupture in self.ruptures_and_poes2],
            self.tom, 'trt2'
        )
        self.disagreggated_poes = dict(
            (rupture, poes) for (poes, rupture) in self.ruptures_and_poes1 +
            self.ruptures_and_poes2
        )
        self.site = Site(Point(0, 0), 2, False, 4, 5)

        self.iml, self.imt, self.truncation_level = (
            object(), object(), object())
        gsim = self.FakeGSIM(self.iml, self.imt, self.truncation_level,
                             n_epsilons=3,
                             disaggregated_poes=self.disagreggated_poes)
        self.gsim = gsim
        self.gsims = {'trt1': gsim, 'trt2': gsim}
        self.sources = [self.source1, self.source2]

        self.orig_make_contexts = ContextMaker.make_contexts
        ContextMaker.make_contexts = lambda self, sites, rupture: (
            sites, rupture, None)

    def tearDown(self):
        ContextMaker.make_contexts = self.orig_make_contexts


class CollectBinsDataTestCase(_BaseDisaggTestCase):
    def test_no_filters(self):
        (mags, dists, lons, lats, trts, trt_bins, probs_no_exceed) = \
            disagg._collect_bins_data_old(
                self.sources, self.site, self.imt, self.iml, self.gsims,
                self.truncation_level, n_epsilons=3)

        aae = numpy.testing.assert_array_equal

        aae(mags, [5, 5, 5, 5, 9, 5, 5, 5, 6, 6, 6, 8, 7])
        aae(dists, [3, 11, 12, 13, 14, 11, 11, 10, 12, 12, 11, 5, 5])
        aae(lons, [22, 22, 22, 22, 21, 21, 21, 21, 22, 21, 22, 11, 11])
        aae(lats, [44, 44, 45, 45, 44, 44, 45, 45, 44, 44, 45, 45, 46])<commit_msg>Restored a test how it was<commit_after># The Hazard Library
# Copyright (C) 2012-2016 GEM Foundation
#
# This program 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.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
import mock
import unittest
import warnings

import numpy

from openquake.hazardlib.calc import disagg
from openquake.hazardlib.calc import filters
from openquake.hazardlib.tom import PoissonTOM
from openquake.hazardlib.geo import Point, Mesh
from openquake.hazardlib.site import Site
from openquake.hazardlib.gsim.base import ContextMaker


class _BaseDisaggTestCase(unittest.TestCase):
    class FakeSurface(object):
        def __init__(self, distance, lon, lat):
            self.distance = distance
            self.lon = lon
            self.lat = lat

        def get_joyner_boore_distance(self, sites):
            assert len(sites) == 1
            return numpy.array([self.distance], float)

        def get_closest_points(self, sites):
            assert len(sites) == 1
            return Mesh(numpy.array([self.lon], float),
                        numpy.array([self.lat], float),
                        depths=None)

    class FakeRupture(object):
        def __init__(self, mag, probability, distance, lon, lat):
            self.mag = mag
            self.probability = probability
            self.surface = _BaseDisaggTestCase.FakeSurface(distance, lon, lat)

        def get_probability_no_exceedance(self, poe):
            return (1 - self.probability) ** poe

    class FakeSource(object):
        def __init__(self, source_id, ruptures, tom, tectonic_region_type):
            self.source_id = source_id
            self.ruptures = ruptures
            self.tom = tom
            self.tectonic_region_type = tectonic_region_type

        def iter_ruptures(self):
            return iter(self.ruptures)

    class FailSource(FakeSource):
        def iter_ruptures(self):
            raise ValueError('Something bad happened')

    class FakeGSIM(object):
        REQUIRES_DISTANCES = set()
        REQUIRES_RUPTURE_PARAMETERS = set()
        REQUIRES_SITES_PARAMETERS = set()

        def __init__(self, iml, imt, truncation_level, n_epsilons,
                     disaggregated_poes):
            self.disaggregated_poes = disaggregated_poes
            self.n_epsilons = n_epsilons
            self.iml = iml
            self.imt = imt
            self.truncation_level = truncation_level
            self.dists = object()

        def disaggregate_poe(self, sctx, rctx, dctx, imt, iml,
                             truncation_level, n_epsilons):
            assert truncation_level is self.truncation_level
            assert imt is self.imt
            assert iml is self.iml
            assert n_epsilons is self.n_epsilons
            assert len(sctx) == 1
            return numpy.array([self.disaggregated_poes[rctx]])

    def setUp(self):
        self.ruptures_and_poes1 = [
            ([0, 0, 0], self.FakeRupture(5, 0.1, 3, 22, 44)),
            ([0.1, 0.2, 0.1], self.FakeRupture(5, 0.2, 11, 22, 44)),
            ([0, 0, 0.3], self.FakeRupture(5, 0.01, 12, 22, 45)),
            ([0, 0.05, 0.001], self.FakeRupture(5, 0.33, 13, 22, 45)),
            ([0, 0, 0], self.FakeRupture(9, 0.4, 14, 21, 44)),
            ([0, 0, 0.02], self.FakeRupture(5, 0.05, 11, 21, 44)),
            ([0.04, 0.1, 0.04], self.FakeRupture(5, 0.53, 11, 21, 45)),
            ([0.2, 0.3, 0.2], self.FakeRupture(5, 0.066, 10, 21, 45)),
            ([0.3, 0.4, 0.3], self.FakeRupture(6, 0.1, 12, 22, 44)),
            ([0, 0, 0.1], self.FakeRupture(6, 0.1, 12, 21, 44)),
            ([0, 0, 0], self.FakeRupture(6, 0.1, 11, 22, 45)),
        ]
        self.ruptures_and_poes2 = [
            ([0, 0.1, 0.04], self.FakeRupture(8, 0.04, 5, 11, 45)),
            ([0.1, 0.5, 0.1], self.FakeRupture(7, 0.03, 5, 11, 46))
        ]
        self.time_span = 10
        self.tom = PoissonTOM(time_span=10)
        self.source1 = self.FakeSource(
            1, [rupture for poes, rupture in self.ruptures_and_poes1],
            self.tom, 'trt1'
        )
        self.source2 = self.FakeSource(
            2, [rupture for poes, rupture in self.ruptures_and_poes2],
            self.tom, 'trt2'
        )
        self.disagreggated_poes = dict(
            (rupture, poes) for (poes, rupture) in self.ruptures_and_poes1 +
            self.ruptures_and_poes2
        )
        self.site = Site(Point(0, 0), 2, False, 4, 5)

        self.iml, self.imt, self.truncation_level = (
            object(), object(), object())
        gsim = self.FakeGSIM(self.iml, self.imt, self.truncation_level,
                             n_epsilons=3,
                             disaggregated_poes=self.disagreggated_poes)
        self.gsim = gsim
        self.gsims = {'trt1': gsim, 'trt2': gsim}
        self.sources = [self.source1, self.source2]

        self.orig_make_contexts = ContextMaker.make_contexts
        ContextMaker.make_contexts = lambda self, sites, rupture: (
            sites, rupture, None)

    def tearDown(self):
        ContextMaker.make_contexts = self.orig_make_contexts


class CollectBinsDataTestCase(_BaseDisaggTestCase):
    def test_no_filters(self):
        (mags, dists, lons, lats, trts, trt_bins, probs_no_exceed) = \
            disagg._collect_bins_data_old(
                self.sources, self.site, self.imt, self.iml, self.gsims,
                self.truncation_level, n_epsilons=3,
                source_site_filter=filters.source_site_noop_filter)

        aae = numpy.testing.assert_array_equal

        aae(mags, [5, 5, 5, 5, 9, 5, 5, 5, 6, 6, 6, 8, 7])
        aae(dists, [3, 11, 12, 13, 14, 11, 11, 10, 12, 12, 11, 5, 5])
        aae(lons, [22, 22, 22, 22, 21, 21, 21, 21, 22, 21, 22, 11, 11])
        aae(lats, [44, 44, 45, 45, 44, 44, 45, 45, 44, 44, 45, 45, 46])
{'fraction_non_alphanumeric': 0.11365973265838804, 'fraction_numerical': 0.05030451633314878, 'mean_word_length': 2.228804902962206, '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': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7050037', 'n_tokens_mistral': 4801, 'n_tokens_neox': 4194, 'n_words': 1111}