Merge pull request #39 from gisce/new_medidas_format

Soporte para cargar ficheros MEDIDAS

cchloader/adapters/medidas.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import
+
+from cchloader.adapters import CchAdapter
+from cchloader.models.medidas import MedidasSchema
+from marshmallow import Schema, fields, pre_load
+
+
+class MedidasBaseAdapter(Schema):
+    """ MEDIDAS Adapter
+    """
+
+    @pre_load
+    def fix_numbers(self, data):
+        for attr, field in self.fields.iteritems():
+            if isinstance(field, (fields.Integer, fields.Float)):
+                if not data.get(attr):
+                    data[attr] = None
+        return data
+
+    @pre_load
+    def fix_ae(self, data):
+        data['ae'] = data.get('ae', 0)
+
+    @pre_load
+    def fix_r2(self, data):
+        data['r2'] = data.get('r2', 0)
+
+    @pre_load
+    def fix_r3(self, data):
+        data['r3'] = data.get('r3', 0)
+
+    @pre_load
+    def fix_factor_potencia(self, data):
+        factor_potencia = data.get('factor_potencia', 0.0)
+        if ',' in factor_potencia:
+            factor_potencia = factor_potencia.replace(',', '.')
+        data['factor_potencia'] = factor_potencia
+
+    @pre_load
+    def fix_tipo_factor_potencia(self, data):
+        data['tipo_factor_potencia'] = data.get('tipo_factor_potencia', 0)
+
+    @pre_load
+    def fix_season(self, data):
+        valid_values = [0, 1]
+        season = data.get('season')
+        if season and season.isdigit() and season in map(str, valid_values):
+            data['season'] = int(season)
+        else:
+            data['season'] = None
+
+
+class MedidasAdapter(MedidasBaseAdapter, CchAdapter, MedidasSchema):
+    pass

cchloader/models/medidas.py

@@ -0,0 +1,21 @@
+# -*- coding: utf-8 -*-
+# -*- encoding: utf-8 -*-
+from __future__ import absolute_import
+
+from marshmallow import Schema, fields
+from marshmallow.validate import OneOf
+
+
+class MedidasSchema(Schema):
+    cil = fields.String(position=0, required=True)
+    datetime = fields.DateTime(position=1, format='%Y/%m/%d %H:%M:%S')
+    season = fields.Integer(position=2, validate=OneOf([0, 1]))
+    ae = fields.Integer(position=3, allow_none=True)
+    r2 = fields.Integer(position=4, allow_none=True)
+    r3 = fields.Integer(position=5, allow_none=True)
+    factor_potencia = fields.Float(position=6, allow_none=True)
+    tipo_factor_potencia = fields.Integer(position=7, allow_none=True)
+    tipo_lectura = fields.String(position=8, validate=OneOf(['R', 'E']))
+
+
+MedidasSchema()

cchloader/parsers/__init__.py

@@ -13,5 +13,6 @@
 from cchloader.parsers.p2d import P2D
 from cchloader.parsers.rf5d import Rf5d
 from cchloader.parsers.mhcil import Mhcil
+from cchloader.parsers.medidas import Medidas
 from cchloader.parsers.reganecu import Reganecu
 from cchloader.parsers.corbagen import CorbaGen

cchloader/parsers/medidas.py

@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import
+
+from cchloader import logger
+from cchloader.utils import build_dict
+from cchloader.adapters.medidas import MedidasAdapter
+from cchloader.models.medidas import MedidasSchema
+from cchloader.parsers.parser import Parser, register
+
+
+class Medidas(Parser):
+
+    patterns = ['^medidas_(\d{4})_(\d{4})(\d{2})_(\d{1})_(\d{4})(\d{2})(\d{2})',
+                '^medidas_(\d{4})_(\d{4})_(\d{4})(\d{2})_(\d{1})_(\d{4})(\d{2})(\d{2})']
+    encoding = "iso-8859-15"
+    delimiter = ';'
+
+    def __init__(self, strict=False):
+        self.adapter = MedidasAdapter(strict=strict)
+        self.schema = MedidasSchema(strict=strict)
+        self.fields = []
+        self.headers = []
+        for f in sorted(self.schema.fields, key=lambda f: self.schema.fields[f].metadata['position']):
+            field = self.schema.fields[f]
+            self.fields.append((f, field.metadata))
+            self.headers.append(f)
+
+    def parse_line(self, line):
+        slinia = tuple(unicode(line.decode(self.encoding)).split(self.delimiter))
+        slinia = map(lambda s: s.strip(), slinia)
+        parsed = {'medidas': {}, 'orig': line}
+        data = build_dict(self.headers, slinia)
+        result, errors = self.adapter.load(data)
+        if errors:
+            logger.error(errors)
+        parsed['medidas'] = result
+        return parsed, errors
+
+
+register(Medidas)

spec/test_parsers_spec.py

@@ -9,6 +9,7 @@
 from cchloader.parsers.b5d import B5d
 from cchloader.parsers.rf5d import Rf5d
 from cchloader.parsers.mhcil import Mhcil
+from cchloader.parsers.medidas import Medidas
 from cchloader.parsers.corbagen import CorbaGen
 from cchloader.exceptions import ParserNotFoundException
 from cchloader.file import PackedCchFile, CchFile
@@ -47,6 +48,10 @@
         self.mhcil_filenames = [
             'MHCIL_H2_4444_A1_20211201.0'  # Documented
         ]
+        self.medidas_filenames = [
+            'medidas_1234_202402_2_20240301.zip',  # Documented
+            'medidas_1234_5678_202402_2_20240301.zip'  # Documented
+        ]
         self.corbagen_filenames = [
             'CORBAGEN_202403.0'  # Documented
         ]
@@ -139,6 +144,25 @@
                 assert result_mhcil == expected_mhcil
                 break
 
+    with it('test to get MEDIDAS parser'):
+        for filename in self.medidas_filenames:
+            expect(get_parser(filename)).to(equal(Medidas))
+    with it('MEDIDAS parser fits file format'):
+        with PackedCchFile('spec/curve_files/medidas_1234_202402_2_20240301.zip') as packed:
+            for cch_file in packed:
+                for line in cch_file:
+                    expected_medidas = 'ES1234000000001234561F001;2024/02/01 01:00:00;0;-24;0;0;;;R;\n'
+                    result_medidas = line['orig']
+                    assert result_medidas == expected_medidas
+                    break
+        with PackedCchFile('spec/curve_files/medidas_1234_5678_202402_2_20240301.zip') as packed:
+            for cch_file in packed:
+                for line in cch_file:
+                    expected_medidas = 'ES1234000000001234561F001;2024/02/01 01:00:00;0;-24;0;0;;;R;\n'
+                    result_medidas = line['orig']
+                    assert result_medidas == expected_medidas
+                    break
+
     with it('test to get CORBAGEN parser'):
         for filename in self.corbagen_filenames:
             expect(get_parser(filename)).to(equal(CorbaGen))