Path management

Errors management
Unique version management

configuration/environment/notebook_environment.yaml

@@ -0,0 +1,11 @@
+data:
+ externals: ../data/externals
+ raw: ../data/raw
+ processed: ../data/processed
+ final: ../data/final
+ results: ../results
+
+
+reload: true
+extract: true
+save_to_disk: false

configuration/main.yaml

@@ -1,22 +1,15 @@
 # @package _global_
 version_base: 0.01
-data:
- externals: ../data/externals
- raw: ../data/raw
- processed: ../data/processed
- final: ../data/final
- results: ../results
+defaults:
+ - environment/notebook_environment
 
 export_path: .
 
 output_file: unit_risk_score.csv
 
 
 surveys: []
-survey_version: all
-reload: false
-extract: true
-save_to_disk: false
+survey_version: null
 
 features:
  answer_hour_set:

main.py

@@ -1,49 +1,43 @@
 import os
 from omegaconf import DictConfig, OmegaConf
+from hydra.core.hydra_config import HydraConfig
 from src.unit_proccessing import *
 import hydra
-#from memory_profiler import memory_usage
+# from memory_profiler import memory_usage
 import warnings
-warnings.simplefilter(action='ignore', category=Warning)
-
-
-def manage_relative_path(config, abosulute_path):
- for name, relative_path in config.data.items():
- if relative_path.startswith('../'):
- config['data'][name] = os.path.join(abosulute_path, relative_path.replace('../', ''))
- return config
 
-
-def manage_survey_definition(config):
- if config['surveys'] != 'all' and type(config['surveys']) == str:
- config['surveys'] = [config['surveys']]
- if config['survey_version'] != 'all' and type(config['survey_version']) == str:
- config['survey_version'] = [config['survey_version']]
- return config
+warnings.simplefilter(action='ignore', category=Warning)
 
 
-def manage_export_path(config):
+def manage_path(config):
  if config['export_path'] is not None:
- config['data']['externals'] = os.path.dirname(config['export_path'])
+ if os.path.isabs(config['export_path']) is False:
+ root_path = HydraConfig.get().runtime.cwd
+ config['export_path'] = os.path.join(root_path, config['export_path'])
+ config['environment']['data']['externals'] = os.path.dirname(config['export_path'])
  config['surveys'] = [os.path.basename(config['export_path'])]
+ if os.path.isabs(config['output_file']) is False:
+ root_path = HydraConfig.get().runtime.cwd
+ config['output_file'] = os.path.join(root_path, config['output_file'])
  return config
 
 
 @hydra.main(config_path='configuration', version_base='1.1', config_name='main.yaml')
 def unit_risk_score(config: DictConfig) -> None:
- #print(OmegaConf.to_yaml(config))
+ # print(OmegaConf.to_yaml(config))
  print("*" * 12)
- config = manage_export_path(config)
- config = manage_relative_path(config, hydra.utils.get_original_cwd())
- config = manage_survey_definition(config)
- features_class = UnitDataProcessing(config)
- df_item = features_class.df_item
- df_unit = features_class.df_unit
- features_class.make_global_score()
- features_class.save()
+ config = manage_path(config)
+ try:
+ survey_class = UnitDataProcessing(config)
+ df_item = survey_class.df_item
+ df_unit = survey_class.df_unit
+ survey_class.make_global_score()
+ survey_class.save()
+ except ValueError as e:
+ print(f"An error occurred: {e}")
 
 
 if __name__ == "__main__":
  unit_risk_score()
- #mem_usage = memory_usage(unit_risk_score)
- #print(f"Memory usage (in MB): {max(mem_usage)}")
+ # mem_usage = memory_usage(unit_risk_score)
+ # print(f"Memory usage (in MB): {max(mem_usage)}")

src/feature_processing.py

@@ -7,8 +7,8 @@ def __init__(self, config):
  super().__init__(config)
 
  self.extract()
- paradata, questionaire, microdata = self.get_dataframes(reload=self.config['reload'],
- save_to_disk=self.config['save_to_disk'])
+ paradata, questionaire, microdata = self.get_dataframes(reload=self.config['environment']['reload'],
+ save_to_disk=self.config['environment']['save_to_disk'])
  print('Data Loaded')
  self._allowed_features = ['f__' + k for k, v in config['features'].items() if v['use']]
  self.item_level_columns = ['interview__id', 'variable_name', 'roster_level']
@@ -87,14 +87,14 @@ def df_paradata(self):
 
  @property
  def df_microdata(self):
- paradata, questionaire, microdata = self.get_dataframes(reload=self.config['reload'],
- save_to_disk=self.config['save_to_disk'])
+ paradata, questionaire, microdata = self.get_dataframes(reload=self.config['environment']['reload'],
+ save_to_disk=self.config['environment']['save_to_disk'])
  return microdata
 
  @property
  def df_questionaire(self):
- paradata, questionaire, microdata = self.get_dataframes(reload=self.config['reload'],
- save_to_disk=self.config['save_to_disk'])
+ paradata, questionaire, microdata = self.get_dataframes(reload=self.config['environment']['reload'],
+ save_to_disk=self.config['environment']['save_to_disk'])
  return questionaire
 
  def make_index_col(self, df):
@@ -274,7 +274,7 @@ def make_df_responsible(self):
 
  def save_data(self, df, file_name):
 
- target_dir = os.path.join(self.config.data.raw, self.config.surveys)
+ target_dir = os.path.join(self.config['environment']['data']['raw'], self.config.surveys)
  survey_path = os.path.join(target_dir, self.config.survey_version)
  processed_data_path = os.path.join(survey_path, 'processed_data')
  df.to_pickle(os.path.join(processed_data_path, f'{file_name}.pkl'))

src/import_manager.py

@@ -478,23 +478,24 @@ def __init__(self, config):
  self.file_dict = {}
  self.get_survey_version()
 
+
  def get_files(self):
  """
  Get a dictionary with all zip files from the surveys defined in the config.
  """
- # code omitted for brevity
  # Get a dictionary with all zip files from the surveys defined in config
  if self.config.surveys == 'all':
- import_path = os.listdir(self.config.data.externals)
+ import_path = os.listdir(self.config['environment']['data']['externals'])
  else:
  # Get surveys defined in the config file that are present in the path
- import_path = [survey for survey in self.config.surveys if survey in os.listdir(self.config.data.externals)]
-
+ import_path = [survey for survey in self.config.surveys if survey in os.listdir(self.config['environment']['data']['externals'])]
+ if len(import_path) == 0:
+ raise ValueError(f"ERROR: survey path {self.config['export_path']} does not exists")
  for survey_name in import_path:
- if os.path.isdir(os.path.join(self.config.data.externals, survey_name)):
+ if os.path.isdir(os.path.join(self.config['environment']['data']['externals'], survey_name)):
  self.file_dict[survey_name] = self.file_dict.get(survey_name, {})
 
- survey_path = os.path.join(self.config.data.externals, survey_name)
+ survey_path = os.path.join(self.config['environment']['data']['externals'], survey_name)
  for filename in os.listdir(survey_path):
  if filename.endswith('.zip'):
 
@@ -515,12 +516,18 @@ def get_survey_version(self):
  """
  self.get_files()
  if self.config.surveys != 'all':
- if self.config.survey_version != 'all':
- self.file_dict = {k: {nk: v for nk, v in nested_dict.items() if nk in self.config.survey_version} for
- k, nested_dict in self.file_dict.items() if k in self.config.surveys}
- else:
+ if self.config.survey_version is None:
+ if len(self.file_dict[self.config.surveys[0]]) > 1:
+ raise ValueError(f"There are multiple versions in {self.config['export_path']}. "
+ f"Either specify survey_version=all in python main.py i.e. \n"
+ f"python main.py export_path={self.config['export_path']} output_file={self.config['output_file']} survey_version=all "
+ f"\n OR provide a path with only one version.")
+ elif self.config.survey_version == 'all':
  self.file_dict = {survey: survey_data for survey, survey_data in self.file_dict.items() if
  survey in self.config.surveys}
+ else:
+ self.file_dict = {k: {nk: v for nk, v in nested_dict.items() if nk in self.config.survey_version} for
+ k, nested_dict in self.file_dict.items() if k in self.config.surveys}
 
  def extract(self, overwrite_dir=False):
  """
@@ -529,9 +536,9 @@ def extract(self, overwrite_dir=False):
  Parameters:
  overwrite_dir: A boolean indicating whether to overwrite the existing directory.
  """
- if self.config['extract']:
+ if self.config['environment']['extract']:
  for survey_name, survey in self.file_dict.items():
- target_dir = os.path.join(self.config.data.raw, survey_name)
+ target_dir = os.path.join(self.config['environment']['data']['raw'], survey_name)
  if overwrite_dir and os.path.exists(target_dir):
  shutil.rmtree(target_dir)
  # Create a new target directory if it does not yet exist
@@ -589,7 +596,7 @@ def get_dataframes(self, save_to_disk=True, reload=False):
  dfs_questionnaires = []
  dfs_microdata = []
  for survey_name, survey in self.file_dict.items():
- target_dir = os.path.join(self.config.data.raw, survey_name)
+ target_dir = os.path.join(self.config['environment']['data']['raw'], survey_name)
 
  for survey_version, files in survey.items():
  print(f"IMPORTING: {survey_name} with version {survey_version}. ")

src/unit_proccessing.py

@@ -36,14 +36,14 @@ def __init__(self, config):
  def df_unit_score(self):
  for method_name in self.get_make_methods(method_type='score', level='unit'):
  feature_name = method_name.replace('make_score_unit__', 'f__')
-
+ score_name = self.rename_feature(feature_name)
  if feature_name in self._allowed_features and self._score_columns is None:
  try:
- print('Processing Score {}...'.format(feature_name))
+ print('Processing Score {}...'.format(score_name))
  getattr(self, method_name)(feature_name)
  # print('Score{} Processed'.format(feature_name))
  except Exception as e:
- print("WARNING: SCORE: {} won't be used in further calculation".format(feature_name))
+ print("WARNING: SCORE: {} won't be used in further calculation".format(score_name))
 
  score_columns = [col for col in self._df_unit if
  col.startswith('s__')] # and col.replace('s__','f__') in self._allowed_features]
@@ -100,7 +100,7 @@ def save(self):
  df['unit_risk_score'] = df['unit_risk_score'].round(2)
  df.sort_values('unit_risk_score', inplace=True)
  file_name = "_".join([self.config.surveys[0], self.config.survey_version[0], 'unit_risk_score']) + ".csv"
- output_path = self.config.output_file.split('.')[0] + '.csv'
+ output_path = self.config['output_file'].split('.')[0] + '.csv'
  df.to_csv(output_path, index=False)
  print(f'SUCCESS! you can find the unit_risk_score output file in {output_path}')