Implement BO Lastgang + BO LastgangKompakt

src/bo4e/bo/lastgang.py

@@ -0,0 +1,131 @@
+"""
+Contains Lastgang and LastgangKompakt class
+and corresponding marshmallow schema for de-/serialization
+"""
+from typing import List, Optional
+
+import attr
+from marshmallow import fields
+from marshmallow_enum import EnumField  # type:ignore[import]
+
+from bo4e.bo.geschaeftsobjekt import Geschaeftsobjekt, GeschaeftsobjektSchema
+from bo4e.com.tagesvektor import Tagesvektor, TagesvektorSchema
+from bo4e.com.zeitintervall import Zeitintervall, ZeitintervallSchema
+from bo4e.com.zeitreihenwert import Zeitreihenwert, ZeitreihenwertSchema
+from bo4e.enum.botyp import BoTyp
+from bo4e.enum.lokationstyp import Lokationstyp
+from bo4e.enum.mengeneinheit import Mengeneinheit
+from bo4e.enum.sparte import Sparte
+from bo4e.validators import check_list_length_at_least_one, obis_validator
+
+
+# pylint: disable=too-few-public-methods
+@attr.s(auto_attribs=True, kw_only=True)
+class _LastgangBody:
+    """
+    The LastgangBody is a mixin that contains the "body" of a Lastgang that is used in both the :class:`Lastgang` as
+    well as :class:`LastgangKompakt`.
+    """
+
+    #: Angabe, ob es sich um einen Gas- oder Stromlastgang handelt
+    sparte: Sparte = attr.ib(validator=attr.validators.instance_of(Sparte))
+
+    #: Eindeutige Nummer der Marktlokation bzw der Messlokation, zu der der Lastgang gehört
+    lokations_id: str = attr.ib(validator=attr.validators.instance_of(str))
+
+    #: Marktlokation oder Messlokation
+    lokationstyp: str = attr.ib(validator=attr.validators.instance_of(Lokationstyp))
+    # todo: implement a lokations-id + lokationstyp cross check (such that lokationstyp malo checks for valid malo id)
+    # https://github.com/Hochfrequenz/BO4E-python/issues/321
+
+    #: Definition der gemessenen Größe anhand ihrer Einheit
+    messgroesse: Mengeneinheit = attr.ib(validator=attr.validators.instance_of(Mengeneinheit))
+
+    # optional attributes
+    #: Versionsnummer des Lastgangs
+    version: Optional[str] = attr.ib(default=None, validator=attr.validators.optional(attr.validators.instance_of(str)))
+    #: Die OBIS-Kennzahl für den Wert, die festlegt, welche Größe mit dem Stand gemeldet wird, z.B. '1-0:1.8.1'
+    obis_kennzahl: Optional[str] = attr.ib(default=None, validator=attr.validators.optional(obis_validator))
+
+
+# pylint: disable=too-many-instance-attributes, too-few-public-methods
+@attr.s(auto_attribs=True, kw_only=True)
+class LastgangKompakt(Geschaeftsobjekt, _LastgangBody):
+    """
+    Modell zur Abbildung eines kompakten Lastganges.
+    In diesem Modell werden die Messwerte in Form von Tagesvektoren mit fester Anzahl von Werten übertragen.
+    Daher ist dieses BO nur zur Übertragung von äquidistanten Messwertverläufen geeignet.
+    """
+
+    # required attributes
+    bo_typ: BoTyp = attr.ib(default=BoTyp.LASTGANG_KOMPAKT)
+
+    #: Angabe des Rasters innerhalb aller Tagesvektoren dieses Lastgangs
+    zeitintervall: Zeitintervall = attr.ib(validator=attr.validators.instance_of(Zeitintervall))
+    # todo: implement a cross check that this zeitintervall is actually the one used in tagesvektoren
+    # https://github.com/Hochfrequenz/BO4E-python/issues/322
+
+    #: Die im Lastgang enthaltenen Messwerte in Form von Tagesvektoren
+    tagesvektoren: List[Tagesvektor] = attr.ib(
+        validator=attr.validators.deep_iterable(
+            member_validator=attr.validators.instance_of(Tagesvektor),
+            iterable_validator=attr.validators.instance_of(list),
+        )
+    )
+
+
+# pylint: disable=too-many-instance-attributes, too-few-public-methods
+@attr.s(auto_attribs=True, kw_only=True)
+class Lastgang(Geschaeftsobjekt, _LastgangBody):
+    """
+    Modell zur Abbildung eines Lastganges;
+    In diesem Modell werden die Messwerte mit einem vollständigen Zeitintervall angegeben und es bietet daher eine hohe
+    Flexibilität in der Übertragung jeglicher zeitlich veränderlicher Messgrössen.
+    """
+
+    # required attributes
+    bo_typ: BoTyp = attr.ib(default=BoTyp.LASTGANG)
+
+    #: Die im Lastgang enthaltenen Messwerte
+    werte: List[Zeitreihenwert] = attr.ib(
+        validator=attr.validators.deep_iterable(
+            member_validator=attr.validators.instance_of(Zeitreihenwert),
+            iterable_validator=check_list_length_at_least_one,
+        )
+    )
+
+
+class _LastgangBodySchemaMixin:
+    """
+    A mixin for schemas to deserialize Lastgang and LastgangKompakt objects.
+    """
+
+    sparte = EnumField(Sparte)
+    lokations_id = fields.Str()
+    lokationstyp = EnumField(Lokationstyp)
+    messgroesse = EnumField(Mengeneinheit)
+
+    # optional attributes
+    obis_kennzahl = fields.Str(load_default=None)
+    version = fields.Str(allow_none=True)
+
+
+class LastgangKompaktSchema(GeschaeftsobjektSchema, _LastgangBodySchemaMixin):
+    """
+    Schema for de-/serialization of LastgangKompakt
+    """
+
+    class_name = LastgangKompakt
+    # required attributes
+    zeitintervall = fields.Nested(ZeitintervallSchema)
+    tagesvektoren = fields.List(fields.Nested(TagesvektorSchema))
+
+
+class LastgangSchema(GeschaeftsobjektSchema, _LastgangBodySchemaMixin):
+    """
+    Schema for de-/serialization of Lastgang
+    """
+
+    class_name = Lastgang
+    # required attributes
+    werte = fields.List(fields.Nested(ZeitreihenwertSchema))

src/bo4e/enum/botyp.py

@@ -12,11 +12,13 @@ class BoTyp(StrEnum):
     AUSSCHREIBUNG = "AUSSCHREIUNG"
     ENERGIEMENGE = "ENERGIEMENGE"
     # It is practical to use the BoTyp Enum as discriminator in the database.
-    # Therefore we added one additional entry for GESCHAEFTSOBJEKT
+    # Therefore, we added one additional entry for GESCHAEFTSOBJEKT
     # This is not defined by the documentation!
     GESCHAEFTSOBJEKT = "GESCHAEFTSOBJEKT"
     GESCHAEFTSPARTNER = "GESCHAEFTSPARTNER"
     KOSTEN = "KOSTEN"
+    LASTGANG = "LASTGANG"
+    LASTGANG_KOMPAKT = "LASTGANG_KOMPAKT"
     MARKTLOKATION = "MARKTLOKATION"
     MESSLOKATION = "MESSLOKATION"
     MARKTTEILNEHMER = "MARKTTEILNEHMER"

tests/test_lastgang.py

@@ -0,0 +1,38 @@
+import pytest  # type:ignore[import]
+
+from bo4e.bo.lastgang import Lastgang, LastgangSchema
+from bo4e.enum.lokationstyp import Lokationstyp
+from bo4e.enum.mengeneinheit import Mengeneinheit
+from bo4e.enum.sparte import Sparte
+from tests.serialization_helper import assert_serialization_roundtrip  # type:ignore[import]
+from tests.test_zeitreihenwert import example_zeitreihenwert  # type:ignore[import]
+
+
+class TestLastgang:
+    @pytest.mark.parametrize(
+        "lastgang_kompakt",
+        [
+            pytest.param(
+                Lastgang(
+                    version="1.1",
+                    sparte=Sparte.STROM,
+                    lokations_id="DE0000011111222223333344444555556",
+                    obis_kennzahl="1-0:1.8.1",
+                    lokationstyp=Lokationstyp.MELO,
+                    messgroesse=Mengeneinheit.KWH,
+                    werte=[example_zeitreihenwert],
+                ),
+            ),
+        ],
+    )
+    def test_serialization_roundtrip(self, lastgang_kompakt: Lastgang):
+        """
+        Test de-/serialisation
+        """
+        assert_serialization_roundtrip(lastgang_kompakt, LastgangSchema())
+
+    def test_missing_required_attribute(self):
+        with pytest.raises(TypeError) as excinfo:
+            _ = Lastgang()
+
+        assert "missing 5 required" in str(excinfo.value)

tests/test_lastgangkompakt.py

@@ -0,0 +1,57 @@
+import pytest  # type:ignore[import]
+
+from bo4e.bo.lastgang import LastgangKompakt, LastgangKompaktSchema
+from bo4e.com.zeitintervall import Zeitintervall
+from bo4e.enum.lokationstyp import Lokationstyp
+from bo4e.enum.mengeneinheit import Mengeneinheit
+from bo4e.enum.sparte import Sparte
+from bo4e.enum.zeiteinheit import Zeiteinheit
+from tests.serialization_helper import assert_serialization_roundtrip  # type:ignore[import]
+from tests.test_tagesvektor import example_tagesvektor, example_tagesvektor_json  # type:ignore[import]
+
+
+class TestLastgangKompakt:
+    @pytest.mark.parametrize(
+        "lastgang_kompakt, expected_json_dict",
+        [
+            pytest.param(
+                LastgangKompakt(
+                    version="1.1",
+                    sparte=Sparte.STROM,
+                    lokations_id="DE0000011111222223333344444555556",
+                    obis_kennzahl="1-0:1.8.1",
+                    lokationstyp=Lokationstyp.MELO,
+                    messgroesse=Mengeneinheit.KWH,
+                    zeitintervall=Zeitintervall(
+                        wert=1,
+                        zeiteinheit=Zeiteinheit.VIERTEL_STUNDE,
+                    ),
+                    tagesvektoren=[example_tagesvektor],
+                ),
+                {
+                    "version": "1.1",
+                    "sparte": "STROM",
+                    "lokationstyp": "MELO",
+                    "messgroesse": "KWH",
+                    "zeitintervall": {"zeiteinheit": "VIERTEL_STUNDE", "wert": 1},
+                    "tagesvektoren": [example_tagesvektor_json],
+                    "versionstruktur": "2",
+                    "externeReferenzen": [],
+                    "lokationsId": "DE0000011111222223333344444555556",
+                    "boTyp": "LASTGANG_KOMPAKT",
+                    "obisKennzahl": "1-0:1.8.1",
+                },
+            ),
+        ],
+    )
+    def test_serialization_roundtrip(self, lastgang_kompakt: LastgangKompakt, expected_json_dict: dict):
+        """
+        Test de-/serialisation of LastgangKompakt.
+        """
+        assert_serialization_roundtrip(lastgang_kompakt, LastgangKompaktSchema(), expected_json_dict)
+
+    def test_missing_required_attribute(self):
+        with pytest.raises(TypeError) as excinfo:
+            _ = LastgangKompakt()
+
+        assert "missing 6 required" in str(excinfo.value)

tests/test_tagesvektor.py

@@ -8,30 +8,33 @@
 from tests.serialization_helper import assert_serialization_roundtrip  # type:ignore[import]
 from tests.test_sigmoidparameter import example_sigmoidparameter  # type:ignore[import]
 
+example_tagesvektor: Tagesvektor = Tagesvektor(
+    tag=datetime(2021, 12, 15, 5, 0, tzinfo=timezone.utc),
+    werte=[
+        Zeitreihenwertkompakt(
+            wert=Decimal(40),
+        ),
+        Zeitreihenwertkompakt(
+            wert=Decimal(50),
+        ),
+    ],
+)
+example_tagesvektor_json = {
+    "tag": "2021-12-15T05:00:00+00:00",
+    "werte": [
+        {"wert": "40", "statuszusatz": None, "status": None},
+        {"wert": "50", "statuszusatz": None, "status": None},
+    ],
+}
+
 
 class TestTagesvektor:
     @pytest.mark.parametrize(
         "tagesvektor, expected_json_dict",
         [
             pytest.param(
-                Tagesvektor(
-                    tag=datetime(2021, 12, 15, 5, 0, tzinfo=timezone.utc),
-                    werte=[
-                        Zeitreihenwertkompakt(
-                            wert=Decimal(40),
-                        ),
-                        Zeitreihenwertkompakt(
-                            wert=Decimal(50),
-                        ),
-                    ],
-                ),
-                {
-                    "tag": "2021-12-15T05:00:00+00:00",
-                    "werte": [
-                        {"wert": "40", "statuszusatz": None, "status": None},
-                        {"wert": "50", "statuszusatz": None, "status": None},
-                    ],
-                },
+                example_tagesvektor,
+                example_tagesvektor_json,
             ),
         ],
     )

tests/test_zeitreihenwert.py

@@ -7,17 +7,19 @@
 from bo4e.enum.messwertstatus import Messwertstatus
 from bo4e.enum.messwertstatuszusatz import Messwertstatuszusatz
 
+example_zeitreihenwert = Zeitreihenwert(
+    wert=Decimal(2.5),
+    datum_uhrzeit_von=datetime(2001, 3, 15, tzinfo=timezone.utc),
+    datum_uhrzeit_bis=datetime(2007, 11, 27, tzinfo=timezone.utc),
+)
+
 
 class TestZeitreihenwert:
     def test_zeitreihenwert_only_required_attributes(self):
         """
         Test de-/serialisation of Zeitreihenwert with minimal attributes.
         """
-        zeitreihenwert = Zeitreihenwert(
-            wert=Decimal(2.5),
-            datum_uhrzeit_von=datetime(2001, 3, 15, tzinfo=timezone.utc),
-            datum_uhrzeit_bis=datetime(2007, 11, 27, tzinfo=timezone.utc),
-        )
+        zeitreihenwert = example_zeitreihenwert
 
         schema = ZeitreihenwertSchema()
         json_string = schema.dumps(zeitreihenwert, ensure_ascii=False)