predictops/target/target.py

   1 from configparser import ConfigParser
   2 from csv import DictReader
   3 from datetime import datetime, timedelta
   4 from logging import getLogger
   5 from logging.config import fileConfig
   6 from pathlib import Path
   7
   8 fileConfig((Path.cwd() / 'config') / 'logging.cfg')
   9 logger = getLogger()
  10
  11
  12 class Target:
  13
  14     def __init__(self, config_file=None,
  15                  start=None, end=None, timestep=None, cumulative=None):
  16
  17         self._config = ConfigParser()
  18         self._config.read(config_file)
  19
  20         self._start = start
  21         self._end = end
  22         self._timestep = timestep
  23         self._cumulative = cumulative
  24
  25         logger.info('Initialization of target variable')
  26         self._y = {}
  27         current = self._start
  28         while current <= self._end:
  29             self._y[current] = 0
  30             current += self._timestep
  31
  32         self._timestep = timestep
  33         self._stream_file = eval(self._config['DATA']['csv_file'])
  34         self._get_located_interventions()
  35
  36     @property
  37     def start(self):
  38         return self._start
  39
  40     @start.setter
  41     def start(self, x):
  42         self._start = x
  43
  44     @property
  45     def end(self):
  46         return self._end
  47
  48     @end.setter
  49     def end(self, x):
  50         self._end = x
  51
  52     @property
  53     def y(self):
  54         return self._y
  55
  56     @y.setter
  57     def end(self, y):
  58         self._y = y
  59
  60     def _get_located_interventions(self):
  61         if not self._config['SPECIFICATION'].getboolean('origin')\
  62            and not self._config['SPECIFICATION'].getboolean('destination'):
  63             logger.info('Integrating interventions for the whole area')
  64             with open(self._stream_file) as f:
  65                 reader = DictReader(f, delimiter=',')
  66                 if self._cumulative:
  67                     for row in reader:
  68                         if row['start'] != '':
  69                             start_interv = datetime.strptime(row['start'], '%d/%m/%Y %H:%M:%S')
  70                             start_interv = start_interv.replace(minute=0)
  71                             end_interv = datetime.strptime(row['end'], '%d/%m/%Y %H:%M:%S')
  72                             end_interv = end_interv.replace(minute=0)
  73                             if not (start_interv > self._end or end_interv < self._start):
  74                                 if start_interv < self._start and end_interv <= self._end:
  75                                     current = self._start
  76                                     while current <= end_interv:
  77                                         self._y[current] += 1
  78                                         current += self._timestep
  79                                 elif start_interv >= self._start and end_interv > self._end:
  80                                     current = start_interv
  81                                     while current not in self._y:
  82                                         current -= timedelta(hours=1)
  83                                     while current <= self._end:
  84                                         self._y[current] += 1
  85                                         current += self._timestep
  86                                 elif start_interv >= self._start and end_interv <= self._end:
  87                                     current = start_interv
  88                                     while current not in self._y:
  89                                         current -= timedelta(hours=1)
  90                                     while current <= end_interv:
  91                                         self._y[current] += 1
  92                                         current += self._timestep
  93                 else:
  94                     for row in reader:
  95                         if row['start'] != '':
  96                             start_interv = datetime.strptime(row['start'], '%d/%m/%Y %H:%M:%S')
  97                             start_interv = start_interv.replace(minute=0)
  98                             if start_interv in self._y:
  99                                 self._y[start_interv] += 1
 100
 101