<commit_before>import csv
import math
from nose.plugins.attrib import attr

import numpy.testing

from openquake.baselib.general import AccumDict
from openquake.hazardlib.site import FilteredSiteCollection
from openquake.commonlib.util import max_rel_diff_index
from openquake.commonlib.tests.calculators import CalculatorTestCase
from openquake.qa_tests_data.event_based import (
    blocksize, case_1, case_2, case_4, case_5, case_6, case_7, case_12,
    case_13, case_17, case_18)
from openquake.qa_tests_data.event_based.spatial_correlation import (
    case_1 as sc1, case_2 as sc2, case_3 as sc3)


def joint_prob_of_occurrence(gmvs_site_1, gmvs_site_2, gmv, time_span,
                             num_ses, delta_gmv=0.1):
    """
    Compute the Poissonian probability of a ground shaking value to be in the
    range [``gmv`` - ``delta_gmv`` / 2, ``gmv`` + ``delta_gmv`` / 2] at two
    different locations within a given ``time_span``.

    :param gmvs_site_1, gmvs_site_2:
        Lists of ground motion values (as floats) for two different sites.
    :param gmv:
        Reference value for computing joint probability.
    :param time_span:
        `investigation_time` parameter from the calculation which produced
        these ground motion values.
    :param num_ses:
        `ses_per_logic_tree_path` parameter from the calculation which produced
        these ground motion values. In other words, the total number of
        stochastic event sets.
    :param delta_gmv:
        the interval to consider
    """
    assert len(gmvs_site_1) == len(gmvs_site_2)

    half_delta = float(delta_gmv) / 2
    gmv_close = lambda v: (gmv - half_delta <= v <= gmv + half_delta)
    count = 0
    for gmv_site_1, gmv_site_2 in zip(gmvs_site_1, gmvs_site_2):
        if gmv_close(gmv_site_1) and gmv_close(gmv_site_2):
            count += 1

    prob = 1 - math.exp(- (float(count) / (time_span * num_ses)) * time_span)
    return prob


def get_gmfs_by_imt(fname, sitecol, imts):
    """
    Return a list of dictionaries with a ground motion field per IMT,
    one dictionary per rupture.

    :param fname: path to the CSV file
    :param sitecol: the underlying site collection
    :param imts: the IMTs corresponding to the columns in the CSV file
    """
    dicts = []
    with open(fname) as f:
        for row in csv.reader(f):
            indices = map(int, row[1].split())
            sc = FilteredSiteCollection(indices, sitecol)
            dic = AccumDict()
            for imt, col in zip(imts, row[2:]):
                gmf = numpy.array(map(float, col.split()))
                dic[imt] = sc.expand(gmf, 0)
            dic.tag = row[0]
            dicts.append(dic)
    return sorted(dicts, key=lambda dic: dic.tag)


class EventBasedTestCase(CalculatorTestCase):

    @attr('qa', 'hazard', 'event_based')
    def test_spatial_correlation(self):
        expected = {sc1: [0.99, 0.41],
                    sc2: [0.99, 0.64],
                    sc3: [0.99, 0.22]}

        for case in expected:
            out = self.run_calc(case.__file__, 'job.ini')
            oq = self.calc.oqparam
            self.assertEqual(list(oq.imtls), ['PGA'])
            [fname] = out['gmf_by_trt_gsim', 'csv']
            gmfs = get_gmfs_by_imt(fname, self.calc.sitecol, oq.imtls)
            gmvs_site_1 = [gmf['PGA'][0] for gmf in gmfs]
            gmvs_site_2 = [gmf['PGA'][1] for gmf in gmfs]

            joint_prob_0_5 = joint_prob_of_occurrence(
                gmvs_site_1, gmvs_site_2, 0.5, oq.investigation_time,
                oq.ses_per_logic_tree_path)
            joint_prob_1_0 = joint_prob_of_occurrence(
                gmvs_site_1, gmvs_site_2, 1.0, oq.investigation_time,
                oq.ses_per_logic_tree_path)

            p05, p10 = expected[case]
            numpy.testing.assert_almost_equal(joint_prob_0_5, p05, decimal=1)
            numpy.testing.assert_almost_equal(joint_prob_1_0, p10, decimal=1)

    @attr('qa', 'hazard', 'event_based')
    def test_blocksize(self):
        out = self.run_calc(blocksize.__file__, 'job.ini', concurrent_tasks=4)
        [fname] = out['gmf_by_trt_gsim', 'csv']
        self.assertEqualFiles('expected/0-ChiouYoungs2008.csv',
                              fname, sorted)

        out = self.run_calc(blocksize.__file__, 'job.ini', concurrent_tasks=8)
        [fname] = out['gmf_by_trt_gsim', 'csv']<commit_msg>Fixed the block_size test<commit_after>import csv
import math
from nose.plugins.attrib import attr

import numpy.testing

from openquake.baselib.general import AccumDict
from openquake.hazardlib.site import FilteredSiteCollection
from openquake.commonlib.util import max_rel_diff_index
from openquake.commonlib.tests.calculators import CalculatorTestCase
from openquake.qa_tests_data.event_based import (
    blocksize, case_1, case_2, case_4, case_5, case_6, case_7, case_12,
    case_13, case_17, case_18)
from openquake.qa_tests_data.event_based.spatial_correlation import (
    case_1 as sc1, case_2 as sc2, case_3 as sc3)


def joint_prob_of_occurrence(gmvs_site_1, gmvs_site_2, gmv, time_span,
                             num_ses, delta_gmv=0.1):
    """
    Compute the Poissonian probability of a ground shaking value to be in the
    range [``gmv`` - ``delta_gmv`` / 2, ``gmv`` + ``delta_gmv`` / 2] at two
    different locations within a given ``time_span``.

    :param gmvs_site_1, gmvs_site_2:
        Lists of ground motion values (as floats) for two different sites.
    :param gmv:
        Reference value for computing joint probability.
    :param time_span:
        `investigation_time` parameter from the calculation which produced
        these ground motion values.
    :param num_ses:
        `ses_per_logic_tree_path` parameter from the calculation which produced
        these ground motion values. In other words, the total number of
        stochastic event sets.
    :param delta_gmv:
        the interval to consider
    """
    assert len(gmvs_site_1) == len(gmvs_site_2)

    half_delta = float(delta_gmv) / 2
    gmv_close = lambda v: (gmv - half_delta <= v <= gmv + half_delta)
    count = 0
    for gmv_site_1, gmv_site_2 in zip(gmvs_site_1, gmvs_site_2):
        if gmv_close(gmv_site_1) and gmv_close(gmv_site_2):
            count += 1

    prob = 1 - math.exp(- (float(count) / (time_span * num_ses)) * time_span)
    return prob


def get_gmfs_by_imt(fname, sitecol, imts):
    """
    Return a list of dictionaries with a ground motion field per IMT,
    one dictionary per rupture.

    :param fname: path to the CSV file
    :param sitecol: the underlying site collection
    :param imts: the IMTs corresponding to the columns in the CSV file
    """
    dicts = []
    with open(fname) as f:
        for row in csv.reader(f):
            indices = map(int, row[1].split())
            sc = FilteredSiteCollection(indices, sitecol)
            dic = AccumDict()
            for imt, col in zip(imts, row[2:]):
                gmf = numpy.array(map(float, col.split()))
                dic[imt] = sc.expand(gmf, 0)
            dic.tag = row[0]
            dicts.append(dic)
    return sorted(dicts, key=lambda dic: dic.tag)


class EventBasedTestCase(CalculatorTestCase):

    @attr('qa', 'hazard', 'event_based')
    def test_spatial_correlation(self):
        expected = {sc1: [0.99, 0.41],
                    sc2: [0.99, 0.64],
                    sc3: [0.99, 0.22]}

        for case in expected:
            out = self.run_calc(case.__file__, 'job.ini')
            oq = self.calc.oqparam
            self.assertEqual(list(oq.imtls), ['PGA'])
            [fname] = out['gmf_by_trt_gsim', 'csv']
            gmfs = get_gmfs_by_imt(fname, self.calc.sitecol, oq.imtls)
            gmvs_site_1 = [gmf['PGA'][0] for gmf in gmfs]
            gmvs_site_2 = [gmf['PGA'][1] for gmf in gmfs]

            joint_prob_0_5 = joint_prob_of_occurrence(
                gmvs_site_1, gmvs_site_2, 0.5, oq.investigation_time,
                oq.ses_per_logic_tree_path)
            joint_prob_1_0 = joint_prob_of_occurrence(
                gmvs_site_1, gmvs_site_2, 1.0, oq.investigation_time,
                oq.ses_per_logic_tree_path)

            p05, p10 = expected[case]
            numpy.testing.assert_almost_equal(joint_prob_0_5, p05, decimal=1)
            numpy.testing.assert_almost_equal(joint_prob_1_0, p10, decimal=1)

    @attr('qa', 'hazard', 'event_based')
    def test_blocksize(self):
        out = self.run_calc(blocksize.__file__, 'job.ini', concurrent_tasks=4,
                            exports='csv')
        [fname] = out['gmf_by_trt_gsim', 'csv']
        self.assertEqualFiles('expected/0-ChiouYoungs2008.csv',
                              fname, sorted)

        out = self.run_calc(blocksize.__file__, 'job.ini', concurrent_tasks=8,
                            exports='csv')
        [fname] = out['gmf_by_trt_gsim', 'csv']
{'fraction_non_alphanumeric': 0.09963931469792606, 'fraction_numerical': 0.022993688007213707, 'mean_word_length': 2.606910569105691, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6886181', 'n_tokens_mistral': 3168, 'n_tokens_neox': 2989, 'n_words': 728}