predictops/source/source.py

   1 from csv import DictReader
   2 from logging import getLogger
   3 from logging.config import fileConfig
   4 from os import listdir
   5 from pathlib import Path
   6
   7 fileConfig((Path.cwd() / 'config') / 'logging.cfg')
   8 logger = getLogger()
   9
  10
  11 class Source:
  12     def __init__(self):
  13         '''
  14         Check if the same feature name is used in two different feature sources
  15         '''
  16         logger.info('Check for redondant feature names')
  17         csv_files = Path.cwd() / 'config' / 'features'
  18         list_of_names = []
  19         for csv_file in listdir(csv_files):
  20             with open(csv_file, "r") as f:
  21                 reader = DictReader(f, delimiter=',')
  22                 list_of_names.extend([row['name'] for row in reader])
  23         if len(list_of_names) != len(set(list_of_names)):
  24             raise ValueError("At least two features have the same name")