diff --git a/emod_api/demographics/demographics_base.py b/emod_api/demographics/demographics_base.py index ff72738..e47c846 100644 --- a/emod_api/demographics/demographics_base.py +++ b/emod_api/demographics/demographics_base.py @@ -1,9 +1,7 @@ import json import math import os -import pathlib import sys -import tempfile import warnings from collections import Counter from functools import partial @@ -21,7 +19,6 @@ from emod_api.demographics.age_distribution_old import AgeDistributionOld as AgeDistribution from emod_api.demographics.demographic_exceptions import InvalidNodeIdException from emod_api.demographics.mortality_distribution_old import MortalityDistributionOld as MortalityDistribution -from emod_api.migration import migration class DemographicsBase(BaseInputFile): @@ -366,43 +363,7 @@ def get_nodes_by_name(self, node_names: List[str]) -> Dict[str, Node]: if node_name in node_names} return requested_nodes - def SetMigrationPattern(self, pattern: str = "rwd"): - """ - Set migration pattern. Migration is enabled implicitly. - It's unusual for the user to need to set this directly; normally used by emodpy. - - Args: - pattern: Possible values are "rwd" for Random Walk Diffusion and "srt" for Single Round Trips. - """ - if self.implicits is not None: - if pattern.lower() == "srt": - self.implicits.append(DT._set_migration_pattern_srt) - elif pattern.lower() == "rwd": - self.implicits.append(DT._set_migration_pattern_rwd) - else: - raise ValueError('Unknown migration pattern: %s. Possible values are "rwd" and "srt".', pattern) - - def _SetRegionalMigrationFileName(self, file_name): - """ - Set path to migration file. - - Args: - file_name: Path to migration file. - """ - if self.implicits is not None: - self.implicits.append(partial(DT._set_regional_migration_filenames, file_name=file_name)) - - def _SetLocalMigrationFileName(self, file_name): - """ - Set path to migration file. - - Args: - file_name: Path to migration file. - """ - if self.implicits is not None: - self.implicits.append(partial(DT._set_local_migration_filename, file_name=file_name)) - - def _SetDemographicFileNames(self, file_names): + def set_demographics_filenames(self, file_names: List[str]): """ Set paths to demographic file. @@ -412,127 +373,6 @@ def _SetDemographicFileNames(self, file_names): if self.implicits is not None: self.implicits.append(partial(DT._set_demographic_filenames, file_names=file_names)) - def SetRoundTripMigration(self, - gravity_factor: float, - probability_of_return: float = 1.0, - id_ref: str = 'short term commuting migration'): - """ - Set commuter/seasonal/temporary/round-trip migration rates. You can use the x_Local_Migration configuration - parameter to tune/calibrate. - - Args: - gravity_factor: 'Big G' in gravity equation. Combines with 1, 1, and -2 as the other exponents. - probability_of_return: Likelihood that an individual who 'commuter migrates' will return to the node - of origin during the next migration (not timestep). Defaults to 1.0. Aka, travel, - shed, return." - id_ref: Text string that appears in the migration file itself; needs to match corresponding demographics - file. - """ - if gravity_factor < 0: - raise ValueError("gravity factor can't be negative.") - - gravity_params = [gravity_factor, 1.0, 1.0, -2.0] - if probability_of_return < 0 or probability_of_return > 1.0: - raise ValueError(f"probability_of_return parameter passed by not a probability: {probability_of_return}") - - mig = migration._from_demog_and_param_gravity(self, gravity_params=gravity_params, - id_ref=id_ref, - migration_type=migration.Migration.LOCAL) - migration_file_path = tempfile.NamedTemporaryFile().name + ".bin" - mig.to_file(migration_file_path) - self.migration_files.append(migration_file_path) - - if self.implicits is not None: - self.implicits.append(partial(DT._set_local_migration_roundtrip_probability, - probability_of_return=probability_of_return)) - self.implicits.append(partial(DT._set_local_migration_filename, - file_name=pathlib.PurePath(migration_file_path).name)) - self.SetMigrationPattern("srt") - - def SetOneWayMigration(self, - rates_path: Union[str, os.PathLike], - id_ref: str = 'long term migration'): - """ - Set one way migration. You can use the x_Regional_Migration configuration parameter to tune/calibrate. - - Args: - rates_path: Path to csv file with node-to-node migration rates. Format is: source (node id),destination - (node id),rate. - id_ref: Text string that appears in the migration file itself; needs to match corresponding demographics - file. - """ - - mig = migration.from_csv(pathlib.Path(rates_path), id_ref=id_ref, mig_type=migration.Migration.REGIONAL) - migration_file_path = tempfile.NamedTemporaryFile().name + ".bin" - mig.to_file(migration_file_path) - self.migration_files.append(migration_file_path) - - if self.implicits is not None: - self.implicits.append(partial(DT._set_regional_migration_roundtrip_probability, probability_of_return=0.0)) - self.implicits.append(partial(DT._set_regional_migration_filenames, - file_name=pathlib.PurePath(migration_file_path).name)) - self.SetMigrationPattern("srt") - - def SetSimpleVitalDynamics(self, - crude_birth_rate: CrudeRate = CrudeRate(40), - crude_death_rate: CrudeRate = CrudeRate(20), - node_ids: List = None): - """ - Set fertility, mortality, and initial age with single birth rate and single mortality rate. - - Args: - crude_birth_rate: Birth rate, per year per kiloperson. - crude_death_rate: Mortality rate, per year per kiloperson. - node_ids: Optional list of nodes to limit these settings to. - - """ - - self.SetBirthRate(crude_birth_rate, node_ids) - self.SetMortalityRate(crude_death_rate, node_ids) - self.SetEquilibriumAgeDistFromBirthAndMortRates(crude_birth_rate, crude_death_rate, node_ids) - - # TODO: is this useful in a way that warrants a special-case function in emodpy? - # https://github.com/InstituteforDiseaseModeling/emod-api-old/issues/790 - def SetEquilibriumVitalDynamics(self, - crude_birth_rate: CrudeRate = CrudeRate(40), - node_ids: List = None): - """ - Set fertility, mortality, and initial age with single rate and mortality to achieve steady state population. - - Args: - crude_birth_rate: Birth rate. And mortality rate. - node_ids: Optional list of nodes to limit these settings to. - - """ - - self.SetSimpleVitalDynamics(crude_birth_rate, crude_birth_rate, node_ids) - - # TODO: is this useful in a way that warrants a special-case function in emodpy? - # https://github.com/InstituteforDiseaseModeling/emod-api-old/issues/791 - def SetEquilibriumVitalDynamicsFromWorldBank(self, - wb_births_df: pd.DataFrame, - country: str, - year: int, - node_ids: List = None): - """ - Set steady-state fertility, mortality, and initial age with rates from world bank, for given country and year. - - Args: - wb_births_df: Pandas dataframe with World Bank birth rate by country and year. - country: Country to pick from World Bank dataset. - year: Year to pick from World Bank dataset. - node_ids: Optional list of nodes to limit these settings to. - - """ - - try: - birth_rate = CrudeRate(wb_births_df[wb_births_df['Country Name'] == country][str(year)].tolist()[0]) - # result_scale_factor = 2.74e-06 # assuming world bank units for input - # birth_rate *= result_scale_factor # from births per 1000 pop per year to per person per day - except Exception as ex: - raise ValueError(f"Exception trying to find {year} and {country} in dataframe.\n{ex}") - self.SetEquilibriumVitalDynamics(birth_rate, node_ids) - def SetDefaultIndividualAttributes(self): """ NOTE: This is very Measles-ish. We might want to move into MeaslesDemographics @@ -888,22 +728,6 @@ def SetInitialAgeLikeSubSaharanAfrica(self, description=""): self.SetInitialAgeExponential(description=description) # use default rate - def SetOverdispersion(self, new_overdispersion_value, nodes: List = None): - """ - Set the overdispersion value for the specified nodes (all if empty). - """ - if nodes is None: - nodes = [] - - def enable_overdispersion(config): - print("DEBUG: Setting 'Enable_Infection_Rate_Overdispersion' to 1.") - config.parameters.Enable_Infection_Rate_Overdispersion = 1 - return config - - if self.implicits is not None: - self.implicits.append(enable_overdispersion) - self.raw['Defaults']['NodeAttributes']["InfectivityOverdispersion"] = new_overdispersion_value - def SetInitPrevFromUniformDraw(self, min_init_prev, max_init_prev, description=""): """ Set Initial Prevalence (one value per node) drawn from an uniform distribution. diff --git a/tests/test_config_demog.py b/tests/test_config_demog.py index 22f2c1b..daf6b49 100644 --- a/tests/test_config_demog.py +++ b/tests/test_config_demog.py @@ -19,20 +19,6 @@ def get_config_as_object(self): def reset_config(self): self.config = self.get_config_as_object() - # Tests that if overdispersion is set, Enable_Infection_Rate_Overdispersion is True - def test_age_dependent_transmission_config(self): - for index in range(2): - demog = Demographics.from_template_node() - demog.SetDefaultProperties() - if index: - demog.SetOverdispersion(0.75) - self.assertEqual(len(demog.implicits), 5 + index) - demog.implicits[-1](self.config) - if not index: - self.assertEqual(self.config.parameters.Enable_Infection_Rate_Overdispersion, 0) - else: - self.assertEqual(self.config.parameters.Enable_Infection_Rate_Overdispersion, 1) - def test_set_birth_rate_config(self): demog = Demographics.from_template_node() self.config.parameters.Enable_Birth = 0 # since it is 1 by default diff --git a/tests/test_demog.py b/tests/test_demog.py index 1a0f886..7db1f76 100644 --- a/tests/test_demog.py +++ b/tests/test_demog.py @@ -201,16 +201,6 @@ def test_set_default_properties(self): self.assertIn('IndividualProperties', demog.raw['Defaults']) self.assertEqual(len(demog.implicits), 5) - def test_set_overdispersion(self): - demog = Demographics.from_template_node() - new_overdispersion_value = 0.3 - demog.SetOverdispersion(new_overdispersion_value=new_overdispersion_value) - - self.assertEqual(new_overdispersion_value, demog.raw['Defaults']['NodeAttributes']['InfectivityOverdispersion']) - self.assertEqual(len(demog.implicits), 2) - # TODO: add a test to set different InfectivityOverdispersion values for specific nodes when feature is - # implemented. - @staticmethod def demog_template_test(template, **kwargs): demog = Demographics.from_template_node() @@ -1885,38 +1875,6 @@ def test_create_overlay_file_2(self): overlay_dict = overlay.to_dict() self.assertDictEqual(reference["Defaults"], overlay_dict["Defaults"]) - def test_SetCommuter_default(self): - demog = Demographics.from_template_node() - demog.SetRoundTripMigration(1e-4) - - # get names of functions that will be called by emodpy to set configuration parameters - implicit_funcs_names = [] - for implicit_funcs_name in demog.implicits: - try: - implicit_funcs_names.append(implicit_funcs_name.__name__) - except Exception: - implicit_funcs_names.append(implicit_funcs_name.func.__name__) # partial function - - self.assertTrue(DT._set_local_migration_roundtrip_probability.__name__ in implicit_funcs_names) - self.assertTrue(DT._set_local_migration_filename.__name__ in implicit_funcs_names) - for mig_file in demog.migration_files: - self.assertTrue(os.path.isfile(mig_file), msg='commuter_migration.bin was not generated.') - - def test_SetLongTermMigration_default(self): - demog = Demographics.from_template_node() - demog.SetOneWayMigration(rates_path=manifest.ltm_csv_path) - implicit_funcs_names = [] - for implicit_funcs_name in demog.implicits: - try: - implicit_funcs_names.append(implicit_funcs_name.__name__) - except Exception: - implicit_funcs_names.append(implicit_funcs_name.func.__name__) # partial function - - self.assertTrue(DT._set_regional_migration_roundtrip_probability.__name__ in implicit_funcs_names) - self.assertTrue(DT._set_regional_migration_filenames.__name__ in implicit_funcs_names) - for mig_file in demog.migration_files: - self.assertTrue(os.path.isfile(mig_file), msg='commuter_migration.bin was not generated.') - def test_create_overlay_for_Kurt(self): # ***** Write vital dynamics and susceptibility initialization overlays ***** vd_over_dict = dict() @@ -1957,556 +1915,3 @@ def test_create_overlay_for_Kurt(self): overlay_dict = demog.to_dict() self.assertDictEqual(vd_over_dict["Defaults"], overlay_dict["Defaults"]) - - -class DemographicsComprehensiveTests_Migration(unittest.TestCase): - def setUp(self) -> None: - print(f"\n{self._testMethodName} started...") - self.out_folder = manifest.output_folder - - def test_01_RoundTripMigration_func(self): - """ - fn: Demographics.SetCommuterMigration - Functional test - Defaults - Args: (self, - gravity_factor=1e-4, - probability_of_return=1.0, - id_ref='short term commuting migration'): - """ - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_csv.json") - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - manifest.delete_existing_file(out_filename) - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref="from_csv_test") - demog.generate_file(out_filename) - - list_of_all_node_ids = [] - for i in range(0, len(demog.to_dict()['Nodes'])): - list_of_all_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - demog.SetRoundTripMigration(gravity_factor=.0001) - - # Test 01 - using defaults for basic case. - self.assertEqual(1, len(demog.migration_files)) - bin_file_contents = os.path.join(self.out_folder, self._testMethodName + "_generated_bin_file_unpacked.csv") - with open(demog.migration_files[0], mode='rb') as file: # reading the file as binary. - fileContent = file.read() - unpf = list(struct.unpack("i" * (len(fileContent) // 4), fileContent)) - pd.DataFrame.from_dict(unpf).to_string(bin_file_contents.replace('_file', f'_from_{i}_nodes')) # .to_csv(bin_file_contents) - - def test_02_RoundTripMigration_func(self): - """ - fn: Demographics.SetCommuterMigration - Functional test - calling the function twice: - - Call #1 - Using Defaults: - - Call #2 - Use a gravity_factor=9e-4 - """ - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_csv.json") - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - manifest.delete_existing_file(out_filename) - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref="from_csv_test") - demog.generate_file(out_filename) - - list_of_all_node_ids = [] - for i in range(0, len(demog.to_dict()['Nodes'])): - list_of_all_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - - # Call #1: With Defaults: - demog.SetRoundTripMigration(1e-4) - - # Call 21: With Defaults: - demog.SetRoundTripMigration(9e-4) - - # Validation - self.assertEqual(len(demog.migration_files), 2) - - def test_OneWayMigration_eh(self): - """ - fn: Demographics.SetOneWayMigration - Functional test - EH - rates_path: Path to csv file with node-to-node migration rates. Format is: source (node id),destination (node id),rate. - id_ref: Text string that appears in the migration file itself; needs to match corresponding demographics file. - """ - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_csv.json") - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - manifest.delete_existing_file(out_filename) - - # Case 1 - no arguments - with self.assertRaises(Exception): - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref="from_csv_test") - demog.SetOneWayMigration() - - # Case 2: valid rates_path file name, invalid contents - with self.assertRaises(Exception): - fname = os.path.join(self.out_folder, self._testMethodName + "_ratesfile.csv") - if os.path.isfile(fname): - os.remove(fname) - f = open(fname, 'w') - writer = csv.writer(f) - writer.writerow("x") # Invalid data will be passed to the SetOneWayMigration function therefore it should raise an exception. - f.close() - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref="from_csv_test") - demog.SetOneWayMigration(rates_path=fname) - - def test_01_OneWayMigration_onenode(self): - """ - fn: Demographics.SetOneWayMigration - Functional test - Functional - One row file - node_ids = numeric. - rates_path: Path to csv file with node-to-node migration rates: source (node id),destination (node id),rate. - """ - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - rates_file = os.path.join(self.out_folder, self._testMethodName + "_ratesfile.txt") - if os.path.isfile(rates_file): - os.remove(rates_file) - with open(rates_file, 'w') as rf: - lines = ['source,destination,rate\n', '1881811163,1876306490,0.8\n'] - rf.writelines(lines) - rf.close() - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref="from_csv_test") - demog.SetOneWayMigration(rates_path=rates_file) - - def test_02_OneWayMigration_manynodes(self): - """ - fn: Demographics.SetOneWayMigration - Functional test - Functional - Multiple rows - node_ids = numeric. - rates_path: Path to csv file with node-to-node migration rates: source (node id),destination (node id),rate. - """ - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - rates_file = os.path.join(self.out_folder, self._testMethodName + "_rates_file.csv") - bin_file_contents = os.path.join(self.out_folder, self._testMethodName + "_generated_bin_file_unpacked.csv") - - # Demographics object - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref="from_csv_test") - list_of_all_node_ids = [] - for i in range(0, len(demog.to_dict()['Nodes'])): - list_of_all_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - if os.path.isfile(rates_file): - os.remove(rates_file) - - source = list_of_all_node_ids[0:88] - destination = list_of_all_node_ids[1:89] - rate = [] - for i in range(1, len(source) + 1): - rate.append(1 / i) - - data = {'source': source, 'destination': destination, 'rate': rate} - pd.DataFrame.from_dict(data).to_csv(pathlib.Path(rates_file), index=False) - - # SetOneWayMigration call - demog.SetOneWayMigration(rates_path=rates_file) - - print(",".join(demog.migration_files)) # btw, there should be only one file for this test. - with open(demog.migration_files[0], mode='rb') as file: # reading the file as binary. - fileContent = file.read() - unpf = list(struct.unpack("i" * (len(fileContent) // 4), fileContent)) - pd.DataFrame.from_dict(unpf).to_string(bin_file_contents.replace('_file', f'_file_with_{len(source)}_migrations')) # .to_csv(bin_file_contents) - - -class DemographicsComprehensiveTests_VitalDynamics(unittest.TestCase): - - def setUp(self) -> None: - print(f"\n{self._testMethodName} started...") - self.out_folder = manifest.output_folder - - def validate_node(self, test_node): - self.assertGreater(len(test_node['AgeDistribution']['DistributionValues']), 0) - self.assertEqual(test_node['MortalityDistribution']['AxisNames'], ["gender", "age"]) - self.assertEqual(test_node['MortalityDistribution']['AxisScaleFactors'], [1, 365]) - self.assertEqual(test_node['MortalityDistribution']['AxisUnits'], ["male=0,female=1", "years"]) - self.assertEqual(test_node['MortalityDistribution']['PopulationGroups'], [[0, 1], [0]]) - self.assertEqual(len(test_node['MortalityDistribution']['ResultValues']), 2) - - def verify_demographics_Json(self, demog, test_case, with_nodeids=False, nodes_count=0): - if not with_nodeids: - demogdict = demog.to_dict()['Defaults']['IndividualAttributes'] - self.validate_node(demogdict) - - if with_nodeids: - demogdict = demog.to_dict()['Nodes'] - if nodes_count == 0: - nodes_count = len(demogdict) - for n in range(0, nodes_count): - self.validate_node(demogdict[n]['IndividualAttributes']) - - def test_setsimplevitaldynamics_bvt_01(self): - # SetSimpleVitalDynamics - # Test Type: Core Functional Test - Created FROM_TEMPLATE_NODE - # Arg: No arguments - Using defaults - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_csv.json") - out_updated_filename = out_filename.replace('_demographics_from_csv', '_updated_demographics_from_csv') - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - - demog = Demographics.from_template_node() - demog.generate_file(out_filename) - - demog.SetSimpleVitalDynamics() # function invoke - demog.generate_file(out_updated_filename) - - self.verify_demographics_Json(demog, self._testMethodName) - - def test_setsimplevitaldynamics_bvt_02(self): - # SetSimpleVitalDynamics - # Test Type: Core Functional Test - Created FROM_CSV - # Arg: No arguments - Using defaults - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_demographics_from_template_node') - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - demog.generate_file(out_filename) - - demog.SetSimpleVitalDynamics() # function invoke - demog.generate_file(out_updated_filename) - self.verify_demographics_Json(demog, self._testMethodName) - - def test_SetSimpleVitalDynamics_multiple_nodeids(self): - # SetSimpleVitalDynamics with node ids - # Test Type: Core Functional Test - Created FROM_CSV - # Arg: Using same values as 'docstrings' and a subset of 3 node_ids - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_demographics_from_template_node') - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - list_of_node_ids = [] - for i in range(0, 3): - list_of_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - - demog.generate_file(out_filename) - demog.SetSimpleVitalDynamics(crude_birth_rate=DT.CrudeRate(40), crude_death_rate=DT.CrudeRate(20), node_ids=list_of_node_ids) - - demog.generate_file(out_updated_filename) - self.verify_demographics_Json(demog, self._testMethodName, True, len(list_of_node_ids)) - - def test_SetSimpleVitalDynamics_repeated_call(self): - # SetSimpleVitalDynamics - # Test Type: Core Functional Test - calling the function more than once with diffent args - # Arg: same as docstrings plus subset of node_ids, and, second call with different values and no Node_ids argument) - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_01_demographics_from_template_node') - out_updated_filename_02 = out_updated_filename.replace("01", "02") - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - manifest.delete_existing_file(out_updated_filename_02) - - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - list_of_node_ids = [] - for i in range(0, 3): - list_of_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - - demog.generate_file(out_filename) - demog.SetSimpleVitalDynamics(crude_birth_rate=DT.CrudeRate(40), crude_death_rate=DT.CrudeRate(20), node_ids=list_of_node_ids) # Call #1 - demog.generate_file(out_updated_filename) - self.verify_demographics_Json(demog, self._testMethodName, True, len(list_of_node_ids)) # Call #2 - demog.SetSimpleVitalDynamics(crude_birth_rate=DT.CrudeRate(10), crude_death_rate=DT.CrudeRate(5)) - demog.generate_file(out_updated_filename_02) - - self.verify_demographics_Json(demog, self._testMethodName, True, len(list_of_node_ids)) - - def test_SetSimpleVitalDynamics_combined_functions(self): - # SetSimpleVitalDynamics - # Test Type: Combining the use of this method with other methods - (i.e. SetAgeDistribution) - # Arg: Using same values as 'docstrings' and a subset of 3 node_ids - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_01_demographics_from_template_node') - out_updated_filename_02 = out_updated_filename.replace("01", "02") - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - manifest.delete_existing_file(out_updated_filename_02) - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - list_of_node_ids = [] - for i in range(0, 3): - list_of_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - demog.generate_file(out_filename) - demog.SetSimpleVitalDynamics(crude_birth_rate=DT.CrudeRate(40), crude_death_rate=DT.CrudeRate(20), node_ids=list_of_node_ids) - demog.generate_file(out_updated_filename) - self.verify_demographics_Json(demog, self._testMethodName, True, len(list_of_node_ids)) - demog.SetAgeDistribution(distribution=Distributions.SEAsia_Diag) - demog.generate_file(out_updated_filename_02) - self.verify_demographics_Json(demog, self._testMethodName, True, len(list_of_node_ids)) - - # TODO: Do we need verification for the contents? - - def validate_node_02(self, test_node): - # SetEquilibriumVitalDynamics validations - self.assertGreater(len(test_node['AgeDistribution']['DistributionValues']), 0) - self.assertEqual(test_node['MortalityDistribution']['AxisNames'], ["gender", "age"]) - self.assertEqual(test_node['MortalityDistribution']['AxisScaleFactors'], [1, 365]) - self.assertEqual(test_node['MortalityDistribution']['AxisUnits'], ["male=0,female=1", "years"]) - self.assertEqual(test_node['MortalityDistribution']['PopulationGroups'], [[0, 1], [0]]) - self.assertEqual(len(test_node['MortalityDistribution']['ResultValues']), 2) - - def verify_demographics_Json_02(self, demog, test_case, with_nodeids=False, nodes_count=0): - # validate SetEquilibriumVitalDynamics initial age, fertility, mortality to achieve steady state population. - if not with_nodeids: - demogdict = demog.to_dict()['Defaults'] - individual_attributes = demogdict['IndividualAttributes'] - self.validate_node_02(individual_attributes) # validate initial age, mortality - self.assertGreater(demogdict['NodeAttributes']['BirthRate'], 0) # validate fertility - - if with_nodeids: - demogdict = demog.to_dict()['Nodes'] - if nodes_count == 0: - nodes_count = len(demogdict) - for n in range(0, nodes_count): # For each node - self.validate_node(demogdict[n]['IndividualAttributes']) # validate initial age, mortality - self.assertGreater(demogdict[n]['NodeAttributes']['BirthRate'], 0) # validate fertility - - def test_SetEquilibriumVitalDynamics_bvt_01(self): - # SetEquilibriumVitalDynamics - # Test Type: Core Functional Test - Created FROM_TEMPLATE_NODE - # Arg: No arguments - Using defaults (crude_birth_rate=40/1000, All) - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_csv.json") - out_updated_filename = out_filename.replace('_demographics_from_csv', '_updated_demographics_from_csv') - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - - demog = Demographics.from_template_node() - demog.generate_file(out_filename) - - demog.SetEquilibriumVitalDynamics() # function under test - demog.generate_file(out_updated_filename) - self.verify_demographics_Json_02(demog, self._testMethodName) - - def test_SetEquilibriumVitalDynamics_bvt_02(self): - # SetEquilibriumVitalDynamics - # Test Type: Core Functional Test - Created FROM_CSV - # Arg: No arguments - Using defaults (crude_birth_rate=40/1000, All) - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_demographics_from_template_node') - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - demog.generate_file(out_filename) - - demog.SetEquilibriumVitalDynamics() # function under test - demog.generate_file(out_updated_filename) - self.verify_demographics_Json_02(demog, self._testMethodName) - - def test_SetEquilibriumVitalDynamics_multiple_nodeids(self): - # SetEquilibriumVitalDynamics with 3 node ids - # Test Type: Core Functional Test - Created FROM_CSV - # Arg: Using same values as 'docstrings' and a subset of 3 node_ids - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_demographics_from_template_node') - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - list_of_node_ids = [] - for i in range(0, 3): - list_of_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - - demog.generate_file(out_filename) - demog.SetEquilibriumVitalDynamics(crude_birth_rate=DT.CrudeRate(40), node_ids=list_of_node_ids) - - demog.generate_file(out_updated_filename) - self.verify_demographics_Json_02(demog, self._testMethodName, True, len(list_of_node_ids)) - - def test_SetEquilibriumVitalDynamics_repeated_call(self): - # SetEquilibriumVitalDynamics - # Test Type: Core Functional Test - calling the function more than once with diffent args - # Arg: same as docstrings plus subset of node_ids, and, second call with different values and no Node_ids argument) - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_01_demographics_from_template_node') - out_updated_filename_02 = out_updated_filename.replace("01", "02") - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - manifest.delete_existing_file(out_updated_filename_02) - - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - list_of_node_ids = [] - for i in range(0, 3): - list_of_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - - demog.generate_file(out_filename) - - demog.SetEquilibriumVitalDynamics(crude_birth_rate=DT.CrudeRate(50), node_ids=list_of_node_ids) # Call #1 - demog.generate_file(out_updated_filename) - self.verify_demographics_Json_02(demog, self._testMethodName, True, len(list_of_node_ids)) # verify data for some nodes - - demog.SetEquilibriumVitalDynamics(crude_birth_rate=DT.CrudeRate(30)) - demog.generate_file(out_updated_filename_02) - self.verify_demographics_Json_02(demog, self._testMethodName) # verify default data - - def test_SetEquilibriumVitalDynamics_twice_All(self): - # SetEquilibriumVitalDynamics - # Test Type: Setting Equilibrium Vital Dynamics twice for all nodes - # Arg: crude_death only - different value from first and second calls. - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_01_demographics_from_template_node') - out_updated_filename_02 = out_updated_filename.replace("01", "02") - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - manifest.delete_existing_file(out_updated_filename_02) - - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - demog.generate_file(out_filename) - - demog.SetEquilibriumVitalDynamics(crude_birth_rate=DT.CrudeRate(50)) # First call - demog.generate_file(out_updated_filename) - self.verify_demographics_Json_02(demog, self._testMethodName) # verify default data - birth_rate_01 = demog.to_dict()['Defaults']['NodeAttributes']['BirthRate'] - - demog.SetEquilibriumVitalDynamics(crude_birth_rate=DT.CrudeRate(30)) # Second call - demog.generate_file(out_updated_filename_02) - self.verify_demographics_Json_02(demog, self._testMethodName) # verify default data again - birth_rate_02 = demog.to_dict()['Defaults']['NodeAttributes']['BirthRate'] - self.assertNotEqual(birth_rate_01, birth_rate_02) - - def test_SetEquilibriumVitalDynamics_combined_functions(self): - # SetEquilibriumVitalDynamics - # Test Type: Combining the use of SetEquilibriumVitalDynamics method with other methods - (i.e. SetAgeDistribution) - # Args: Using same values as 'docstrings' and a subset of 3 node_ids - - out_filename = os.path.join(self.out_folder, self._testMethodName + "_demographics_from_template_node.json") - out_updated_filename = out_filename.replace('_demographics_from_template_node', '_updated_01_demographics_from_template_node') - out_updated_filename_02 = out_updated_filename.replace("01", "02") - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - manifest.delete_existing_file(out_updated_filename_02) - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_in_subset.csv') - - demog = Demographics.from_csv(input_file, res=25 / 3600, id_ref=id_ref) - list_of_node_ids = [] - for i in range(0, 3): - list_of_node_ids.append(demog.to_dict()['Nodes'][i]['NodeID']) - demog.generate_file(out_filename) - demog.SetEquilibriumVitalDynamics(crude_birth_rate=DT.CrudeRate(40), node_ids=list_of_node_ids) # FUT - aka - feature under test :) - demog.generate_file(out_updated_filename) - self.verify_demographics_Json(demog, self._testMethodName, True, len(list_of_node_ids)) - demog.SetAgeDistribution(distribution=Distributions.SEAsia_Diag) # Second funcion - demog.generate_file(out_updated_filename_02) - self.verify_demographics_Json_02(demog, self._testMethodName, True, len(list_of_node_ids)) - - # region From World Bank - def from_template_node_with_country(self, _country, _year): - out_filename = os.path.join(self.out_folder, self._testMethodName + f"_demographics_{_country}_FromTNode.json") - out_updated_filename = out_filename.replace('_demographics_', '_demographics_updated_') - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - # World Bank Data: - input_WB_file = os.path.join(manifest.demo_folder, 'wb_data.csv') - _wb_births_df = pd.read_csv(input_WB_file) - - demog = Demographics.from_template_node() - demog.generate_file(out_filename) - - demog.SetEquilibriumVitalDynamicsFromWorldBank(wb_births_df=_wb_births_df, country=_country, year=_year) # FUT - demog.generate_file(out_updated_filename) - - def from_csv_with_country(self, _country, _year): - out_filename = os.path.join(self.out_folder, self._testMethodName + f"_demographics_{_country}_FromCSV.json") - out_updated_filename = out_filename.replace('_demographics_', '_demographics_updated_') - manifest.delete_existing_file(out_filename) - manifest.delete_existing_file(out_updated_filename) - # World Bank Data: - input_WB_file = os.path.join(manifest.demo_folder, 'wb_data.csv') - _wb_births_df = pd.read_csv(input_WB_file) - - id_ref = "from_csv_test" - input_file = os.path.join(manifest.demo_folder, 'demog_some_nodes.csv') - demog = Demographics.from_csv(input_file, res=35 / 3600, id_ref=id_ref) - demog.generate_file(out_filename) - - demog.SetEquilibriumVitalDynamicsFromWorldBank(wb_births_df=_wb_births_df, country=_country, year=_year) # FUT - demog.generate_file(out_updated_filename) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_bvt_01(self): - # SetEquilibriumVitalDynamicsFromWorldBank - # Test Type: Core Functional Test - Created FROM_TEMPLATE_NODE - # Arg: All valid values - # wb_births_df: Pandas dataframe with World Bank birth rate by country and year. - # country: Country to pick from World Bank dataset. - # year: Year to pick from World Bank dataset. - # node_ids: Optional list of nodes to limit these settings to. - self.from_template_node_with_country("Aruba", 1980) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_bvt_02(self): - # SetEquilibriumVitalDynamicsFromWorldBank - # Test Type: Core Functional Test - Created FROM_csv - # Arg: All valid values - # wb_births_df: Pandas dataframe with World Bank birth rate by country and year. - # country: Country to pick from World Bank dataset. - # year: Year to pick from World Bank dataset. - # node_ids: Optional list of nodes to limit these settings to. - self.from_csv_with_country("Mozambique", 1980) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_EH_01(self): - self.from_csv_with_country("Cote d'Ivoire", 1980) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_EH_02(self): - self.from_csv_with_country("Congo, Rep.", 1981) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_EH_03(self): - self.from_csv_with_country("East Asia & Pacific (excluding high income)", 1980) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_EH_04(self): - self.from_csv_with_country("Egypt, Arab Rep.", 1980) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_EH_05(self): - self.from_csv_with_country("Fragile and conflict affected situations", 1980) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_EH_06(self): - self.from_csv_with_country("Least developed countries: UN classification", 1980) - - def test_SetEquilibriumVitalDynamicsFromWorldBank_EH_07(self): - self.from_csv_with_country("Turks and Caicos Islands", 1980) - - # moved from now-defunct test_demog_Parser.py, because the file it used to test has been deleted. - def test_node_id_from_lat_lon_res(self): - node_id = Demographics._node_id_from_lat_lon_res(lat=1000, lon=1000, res=30 / 3600) - node_id_2 = Demographics._node_id_from_lat_lon_res(lat=1000, lon=1000, res=30 / 3600) - self.assertEqual(node_id, node_id_2) - - node_id_3 = Demographics._node_id_from_lat_lon_res(lat=1000, lon=1000, res=30 / 360) - self.assertNotEqual(node_id, node_id_3) - - node_id_4 = Demographics._node_id_from_lat_lon_res(lat=999, lon=1000, res=30 / 3600) - self.assertNotEqual(node_id, node_id_4) - - node_id_5 = Demographics._node_id_from_lat_lon_res(lat=1000, lon=1001, res=30 / 3600) - self.assertNotEqual(node_id, node_id_5)