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
  37
  38     @property
  39     def start(self):
  40         return self._start
  41
  42     @start.setter
  43     def start(self, x):
  44         self._start = x
  45
  46
  47     @property
  48     def end(self):
  49         return self._end
  50
  51     @end.setter
  52     def end(self, x):
  53         self._end = x
  54
  55
  56     @property
  57     def y(self):
  58         return self._y
  59
  60     @y.setter
  61     def end(self, y):
  62         self._y = y
  63
  64
  65
  66     def _get_located_interventions(self):
  67         if not self._config['SPECIFICATION'].getboolean('origin')\
  68            and not self._config['SPECIFICATION'].getboolean('destination'):
  69             logger.info('Integrating interventions for the whole area')
  70             with open(self._stream_file) as f:
  71                 reader = DictReader(f, delimiter=',')
  72                 if self._cumulative:
  73                     for row in reader:
  74                         if row['start'] != '':
  75                             start_interv = datetime.strptime(row['start'], '%d/%m/%Y %H:%M:%S')
  76                             start_interv = start_interv.replace(minute=0)
  77                             end_interv = datetime.strptime(row['end'], '%d/%m/%Y %H:%M:%S')
  78                             end_interv = end_interv.replace(minute=0)
  79                             if not (start_interv > self._end or end_interv < self._start):
  80                                 if start_interv < self._start and end_interv <= self._end:
  81                                     current = self._start
  82                                     while current <= end_interv:
  83                                         self._y[current] += 1
  84                                         current += self._timestep
  85                                 elif start_interv >= self._start and end_interv > self._end:
  86                                     current = start_interv
  87                                     while current not in self._y:
  88                                         current -= timedelta(hours=1)
  89                                     while current <= self._end:
  90                                         self._y[current] += 1
  91                                         current += self._timestep
  92                                 elif start_interv >= self._start and end_interv <= self._end:
  93                                     current = start_interv
  94                                     while current not in self._y:
  95                                         current -= timedelta(hours=1)
  96                                     while current <= end_interv:
  97                                         self._y[current] += 1
  98                                         current += self._timestep
  99                 else:
 100                     for row in reader:
 101                         if row['start'] != '':
 102                             start_interv = datetime.strptime(row['start'], '%d/%m/%Y %H:%M:%S')
 103                             start_interv = start_interv.replace(minute=0)
 104                             if start_interv in self._y:
 105                                 self._y[start_interv] += 1
 106
 107