Merge pull request #40 from gisce/new_infpa_format

Soporte para cargar ficheros INFPA

README.md

@@ -17,6 +17,7 @@ Eina d'importació de CCH
 - `CORBAGEN`
 - `F1`
 - `F5D`
+- `INFPA`
 - `MEDIDAS`
 - `MHCIL`
 - `P1`

cchloader/adapters/infpa.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import
+
+from cchloader.adapters import CchAdapter
+from cchloader.models.infpa import InfpaSchema
+from marshmallow import Schema, fields, pre_load
+
+
+class InfpaBaseAdapter(Schema):
+    """ INFPA 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
+
+
+class InfpaAdapter(InfpaBaseAdapter, CchAdapter, InfpaSchema):
+    pass

cchloader/models/infpa.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+# -*- encoding: utf-8 -*-
+from __future__ import absolute_import
+
+from marshmallow import Schema, fields
+
+
+class InfpaSchema(Schema):
+    cil = fields.String(position=0, required=True)
+    valor = fields.Integer(position=1, required=True)
+    horas = fields.Integer(position=2, required=True)
+
+InfpaSchema()

cchloader/parsers/infpa.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.infpa import InfpaAdapter
+from cchloader.models.infpa import InfpaSchema
+from cchloader.parsers.parser import Parser, register
+
+
+class Infpa(Parser):
+
+    patterns = ['^INFPA_([H][23CPD])_(\d{4})_([PA][12])_(\d{4})(\d{2}).(\d)',
+                '^INFPA_([H][23CPD])_(\d{4})_([PA][12])_(\d{4})(\d{2})']
+    encoding = "iso-8859-15"
+    delimiter = ';'
+
+    def __init__(self, strict=False):
+        self.adapter = InfpaAdapter(strict=strict)
+        self.schema = InfpaSchema(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 = {'infpa': {}, 'orig': line}
+        data = build_dict(self.headers, slinia)
+        result, errors = self.adapter.load(data)
+        if errors:
+            logger.error(errors)
+        parsed['infpa'] = result
+        return parsed, errors
+
+
+register(Infpa)

spec/test_parsers_spec.py

@@ -11,6 +11,7 @@
 from cchloader.parsers.mhcil import Mhcil
 from cchloader.parsers.medidas import Medidas
 from cchloader.parsers.corbagen import CorbaGen
+from cchloader.parsers.infpa import Infpa
 from cchloader.exceptions import ParserNotFoundException
 from cchloader.file import PackedCchFile, CchFile
 
@@ -56,6 +57,9 @@
             'CORBAGEN_202403.0'  # Documented
         ]
         self.wrong_filename = 'P1_20170507_20170706.6'
+        self.infpa_filenames = [
+            'INFPA_H3_1234_P2_202401.0.bz2'  # Documented
+        ]
 
     with it('test to get F1 parser'):
         for filename in self.f1_filenames:
@@ -174,6 +178,18 @@
                 assert result_corbagen == expected_corbagen
                 break
 
+    with it('test to get INFPA parser'):
+        for filename in self.infpa_filenames:
+            expect(get_parser(filename)).to(equal(Infpa))
+    with it('INFPA parser fits file format'):
+        with PackedCchFile('spec/curve_files/INFPA_H3_1234_P2_202401.0.bz2') as packed:
+            for cch_file in packed:
+                for line in cch_file:
+                    expected_infpa = 'ES1234000000002267QR1F001;0;20;\n'
+                    result_infpa = line['orig']
+                    assert result_infpa == expected_infpa
+                    break
+
     with it('test error to get exception'):
         def test_raise_error():
             get_parser(self.wrong_filename)