Merge pull request #77 from gisce/new_obcups_obagrecl_files

Añadir soporte para OBJAGRECL y OBCUPS

README.md

@@ -52,6 +52,8 @@
 - `MCIL345`
 - `MCIL345QH`
 - `MEDIDAS`
+- `OBAGRECL`
+- `OBCUPS`
 - `P1`
 - `P1D`
 - `P2`

mesures/headers.py

@@ -352,6 +352,35 @@
     'read_type'             # str(1) in ('E' for estimated, 'R' for real)
 ]
 
+OBAGRECL_HEADER = [
+    'distribuidora',        # str(4)
+    'comercialitzadora',    # str(4)
+    'agree_tensio',         # str(2)
+    'agree_tarifa',         # str(2)
+    'agree_dh',             # str(2)
+    'agree_tipo',           # str(2)
+    'provincia',            # str(2)
+    'tipus_demanda',        # str(3)
+    'periode',              # str aaaa/mm
+    'motiu_emissor',        # str(3)
+    'magnitud',             # str(2) AE/AS
+    'energia_publicada',    # str(10) kWh
+    'energia_proposada',    # str(10) kWh
+    'comentari_emissor',    # str(255)
+    'auto_obj',             # str(1) S/N
+]
+
+OBCUPS_HEADER = [
+    'cups',                 # str(22)
+    'periode',              # str aaaa/mm
+    'motiu_emissor',        # str(3)
+    'energia_publicada',    # str(10) kWh
+    'energia_proposada',    # str(10) kWh
+    'comentari_emissor',    # str(255)
+    'auto_obj',             # str(1) S/N
+    'magnitud'              # str(2) AS/AE
+]
+
 P1_HEADER = [
     'cups',          # str(22)
     'tipo_medida',   # str(2) (always '11')

mesures/obagrecl.py

@@ -0,0 +1,97 @@
+# -*- coding: utf-8 -*-
+from mesures.dates import *
+from mesures.headers import OBAGRECL_HEADER as COLUMNS
+from mesures.parsers.dummy_data import DummyCurve
+from mesures.utils import check_line_terminator_param
+import os
+import pandas as pd
+
+
+class OBAGRECL(object):
+    def __init__(self, data, emissor=None, distribuidora=None, periode=None, compression='bz2', columns=COLUMNS,
+                 version=0):
+        """
+        :param data: list of dicts or absolute file_path
+        :param emissor: str emissor REE code
+        :param distribuidora: str distribuïdora REE code
+        :param periode: str e.g. '202401'
+        :param compression: 'bz2', 'gz'... OR False otherwise
+        :param columns: list
+        :param version: int
+        """
+        if isinstance(data, list):
+            data = DummyCurve(data).curve_data
+        self.file = self.reader(data)
+        self.generation_date = datetime.now()
+        self.prefix = 'OBAGRECL'
+        self.version = version
+        self.emissor = emissor
+        self.distribuidora = distribuidora
+        self.default_compression = compression
+        self.periode = periode
+        self.columns = columns
+
+    def __len__(self):
+        return len(self.file)
+
+    @property
+    def filename(self):
+        filename = "{prefix}_{emissor}_{distribuidora}_{periode}_{timestamp}.{version}".format(
+            prefix=self.prefix,
+            emissor=self.emissor,
+            distribuidora=self.distribuidora,
+            periode=self.periode,
+            timestamp=self.generation_date.strftime('%Y%m%d'),
+            version=self.version
+            )
+        if self.default_compression:
+            filename += '.{compression}'.format(compression=self.default_compression)
+
+        return filename
+
+    def reader(self, filepath):
+        if isinstance(filepath, str):
+            df = pd.read_csv(filepath, sep=';', names=COLUMNS)
+        elif isinstance(filepath, list):
+            df = pd.DataFrame(data=filepath)
+        else:
+            raise Exception("Filepath must be an str or a list")
+
+        df['comentari_emissor'] = df.apply(lambda row: row['comentari_emissor'] or '', axis=1)
+        df['energia_publicada'] = df.apply(lambda row: row['energia_publicada'] or '', axis=1)
+        df['energia_proposada'] = df.apply(lambda row: row['energia_proposada'] or '', axis=1)
+
+        return df
+
+    def writer(self):
+        """
+        OBAGRECL TMP file generattion
+        :return: file path
+        """
+        # Check and change value version of file
+        existing_files = os.listdir('/tmp')
+        if existing_files:
+            versions = [int(f.split('.')[1]) for f in existing_files if self.filename.split('.')[0] in f]
+            if versions:
+                self.version = max(versions) + 1
+
+        # Generate file path
+        file_path = os.path.join('/tmp', self.filename)
+
+        kwargs = {'sep': ';',
+                  'header': False,
+                  'columns': self.columns,
+                  'index': False,
+                  'float_format': '%.2f',
+                  check_line_terminator_param(): ';\n',
+                  'encoding': 'iso-8859-15',
+                  }
+
+        if self.default_compression:
+            kwargs.update({'compression': self.default_compression})
+
+        # Convert dict data to csv file and save it to the file path generated
+        self.file.to_csv(file_path, **kwargs)
+
+        # Return location tmp file
+        return file_path

mesures/obcups.py

@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+from mesures.headers import OBCUPS_HEADER as COLUMNS
+from mesures.obagrecl import OBAGRECL
+
+
+class OBCUPS(OBAGRECL):
+    def __init__(self, data, emissor=None, distribuidora=None, periode=None, compression='bz2', columns=COLUMNS,
+                 version=0):
+        """
+        :param data: list of dicts or absolute file_path
+        :param emissor: str emissor REE code
+        :param distribuidora: str distribuidora REE code
+        :param periode: str e.g. '202401'
+        :param compression: 'bz2', 'gz'... OR False otherwise
+        :param columns: list
+        :param version: int
+        """
+        super(OBCUPS, self).__init__(data, emissor=emissor, distribuidora=distribuidora, periode=periode,
+                                      compression=compression, columns=COLUMNS, version=version)
+        self.prefix = 'OBCUPS'

spec/generation_files_spec.py

@@ -21,6 +21,8 @@
 from mesures.mcil345 import MCIL345
 from mesures.mcil345qh import MCIL345QH
 from mesures.medidas import MEDIDAS
+from mesures.obagrecl import OBAGRECL
+from mesures.obcups import OBCUPS
 from mesures.p1 import P1
 from mesures.p1d import P1D
 from mesures.p2d import P2D
@@ -829,6 +831,66 @@ def get_sample_reobjecil_data():
             'acceptacio': 'N',
         }]
 
+    @staticmethod
+    def get_sample_obagrecl_data():
+        return [{
+            'distribuidora': '4444',
+            'comercialitzadora': '5555',
+            'agree_tensio': 'E0',
+            'agree_tarifa': '2T',
+            'agree_dh': 'E3',
+            'agree_tipo': '05',
+            'provincia': 'HU',
+            'tipus_demanda': '41',
+            'periode': '2024/10',
+            'motiu_emissor': '100',
+            'magnitud': 'AE',
+            'energia_publicada': '100',
+            'energia_proposada': '110',
+            'auto_obj': 'N'
+        },
+        {
+            'distribuidora': '4444',
+            'comercialitzadora': '6666',
+            'agree_tensio': 'E0',
+            'agree_tarifa': '2T',
+            'agree_dh': 'E3',
+            'agree_tipo': '05',
+            'provincia': 'HU',
+            'tipus_demanda': '00',
+            'periode': '2024/10',
+            'motiu_emissor': '100',
+            'magnitud': 'AE',
+            'energia_publicada': 40,
+            'energia_proposada': 50,
+            'comentari_emissor': 'Paga la energia, primer aviso.',
+            'auto_obj': 'N'
+        }
+        ]
+
+    @staticmethod
+    def get_sample_obcups_data():
+        return [{
+            'cups': 'ES0291000000004444QR1F',
+            'periode': '2024/10',
+            'motiu_emissor': '100',
+            'energia_publicada': '100',
+            'energia_proposada': '110',
+            'comentari_emissor': 'Paga la energia, primer aviso.',
+            'auto_obj': 'N',
+            'magnitud': 'AE'
+        },
+        {
+            'cups': 'ES0291000000004444QR1F',
+            'periode': '2024/10',
+            'motiu_emissor': '100',
+            'energia_publicada': '100',
+            'comentari_emissor': 'Paga la energia, primer aviso.',
+            'auto_obj': 'N',
+            'magnitud': 'AE'
+        }
+        ]
+
 
 with description('A P5D'):
     with it('is instance of P5D Class'):
@@ -1472,3 +1534,34 @@ def get_sample_reobjecil_data():
                     "Paga la energia, primer aviso.;N;N;99;La energia está correcta. A llorar a la llorería.\n"
                     "ES0291000000004444QR1F;2024/01/01 01;2024/02/01 00;100;;;;;;;;N;N;;\n")
         assert f.file[f.columns].to_csv(sep=';', header=None, index=False) == expected
+
+with description('An OBAGRECL'):
+    with it('is instance of OBAGRECL Class'):
+        data = SampleData().get_sample_obagrecl_data()
+        f = OBAGRECL(data)
+        assert isinstance(f, OBAGRECL)
+
+    with it('gets expected content'):
+        data = SampleData().get_sample_obagrecl_data()
+        f = OBAGRECL(data)
+        res = f.writer()
+        expected = "4444;5555;E0;2T;E3;05;HU;41;2024/10;100;AE;100;110;;N\n" \
+                   "4444;6666;E0;2T;E3;05;HU;00;2024/10;100;AE;40;50;Paga la energia, primer aviso.;N\n"
+        assert f.file[f.columns].to_csv(sep=';', header=None, index=False) == expected
+
+with description('An OBCUPS'):
+    with it('is instance of OBCUPS Class'):
+        data = SampleData().get_sample_obcups_data()
+        f = OBCUPS(data)
+        assert isinstance(f, OBCUPS)
+
+    with it('gets expected content'):
+        data = SampleData().get_sample_obcups_data()
+        f = OBCUPS(data)
+        res = f.writer()
+        expected = ("ES0291000000004444QR1F;2024/10;100;100;110;Paga la energia, "
+                    "primer aviso.;N;AE\n"
+                    "ES0291000000004444QR1F;2024/10;100;100;;Paga la energia, "
+                    "primer aviso.;N;AE\n"
+                    )
+        assert f.file[f.columns].to_csv(sep=';', header=None, index=False) == expected