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