Implement COM Rechnungsposition

src/bo4e/bo/marktlokation.py

@@ -2,7 +2,6 @@
 Contains Marktlokation class
 and corresponding marshmallow schema for de-/serialization
 """
-import re
 
 import attr
 from marshmallow import fields
@@ -22,8 +21,7 @@
 from bo4e.enum.netzebene import Netzebene
 from bo4e.enum.sparte import Sparte
 from bo4e.enum.verbrauchsart import Verbrauchsart
-
-_malo_id_pattern = re.compile(r"^[1-9][\d]{10}$")
+from bo4e.validators import validate_marktlokations_id
 
 
 # pylint: disable=too-many-instance-attributes, too-few-public-methods
@@ -33,24 +31,10 @@ class Marktlokation(Geschaeftsobjekt):
     Object containing information about a Marktlokation
     """
 
-    # pylint: disable=unused-argument, no-self-use
-    def _validate_marktlokations_id(self, marktlokations_id_attribute, value):
-        if not value:
-            raise ValueError("The marktlokations_id must not be empty.")
-        if not _malo_id_pattern.match(value):
-            raise ValueError(f"The marktlokations_id '{value}' does not match {_malo_id_pattern.pattern}")
-        expected_checksum = Marktlokation._get_checksum(value)
-        actual_checksum = value[10:11]
-        if expected_checksum != actual_checksum:
-            # pylint: disable=line-too-long
-            raise ValueError(
-                f"The marktlokations_id '{value}' has checksum '{actual_checksum}' but '{expected_checksum}' was expected."
-            )
-
     # required attributes
     bo_typ: BoTyp = attr.ib(default=BoTyp.MARKTLOKATION)
     #: Identifikationsnummer einer Marktlokation, an der Energie entweder verbraucht, oder erzeugt wird.
-    marktlokations_id: str = attr.ib(validator=_validate_marktlokations_id)
+    marktlokations_id: str = attr.ib(validator=validate_marktlokations_id)
     #: Sparte der Marktlokation, z.B. Gas oder Strom
     sparte: Sparte
     #: Kennzeichnung, ob Energie eingespeist oder entnommen (ausgespeist) wird
@@ -122,6 +106,7 @@ def _validate_marktlokations_id(self, marktlokations_id_attribute, value):
 
     # todo: add kundengruppe
 
+    # pylint:disable=unused-argument
     @lokationsadresse.validator
     @geoadresse.validator
     @katasterinformation.validator
@@ -136,31 +121,6 @@ def validate_address_info(self, address_attribute, value):
         if amount_of_given_address_infos != 1:
             raise ValueError("No or more than one address information is given.")
 
-    @staticmethod
-    def _get_checksum(malo_id: str) -> str:
-        """
-        Get the checksum of a marktlokations id.
-        a) Quersumme aller Ziffern in ungerader Position
-        b) Quersumme aller Ziffern auf gerader Position multipliziert mit 2
-        c) Summe von a) und b) d) Differenz von c) zum nächsten Vielfachen von 10 (ergibt sich hier 10, wird die
-           Prüfziffer 0 genommen
-        https://bdew-codes.de/Content/Files/MaLo/2017-04-28-BDEW-Anwendungshilfe-MaLo-ID_Version1.0_FINAL.PDF
-        :param self:
-        :return: the checksum as string
-        """
-        odd_checksum: int = 0
-        even_checksum: int = 0
-        # start counting at 1 to be consistent with the above description
-        # of "even" and "odd" but stop at tenth digit.
-        for i in range(1, 11):
-            digit = malo_id[i - 1 : i]
-            if i % 2 - 1 == 0:
-                odd_checksum += int(digit)
-            else:
-                even_checksum += 2 * int(digit)
-        result: int = (10 - ((even_checksum + odd_checksum) % 10)) % 10
-        return str(result)
-
 
 class MarktlokationSchema(GeschaeftsobjektSchema):
     """

src/bo4e/com/rechnungsposition.py

@@ -0,0 +1,122 @@
+"""
+Contains Rechnungsposition class and corresponding marshmallow schema for de-/serialization
+"""
+from datetime import datetime
+from typing import Optional
+
+import attr
+from marshmallow import fields
+from marshmallow_enum import EnumField  # type:ignore[import]
+
+from bo4e.com.betrag import Betrag, BetragSchema
+from bo4e.com.com import COM, COMSchema
+from bo4e.com.menge import Menge, MengeSchema
+from bo4e.com.preis import Preis, PreisSchema
+from bo4e.com.steuerbetrag import Steuerbetrag, SteuerbetragSchema
+from bo4e.enum.artikelid import ArtikelId
+from bo4e.enum.bdewartikelnummer import BDEWArtikelnummer
+from bo4e.enum.zeiteinheit import Zeiteinheit
+from bo4e.validators import check_bis_is_later_than_von, validate_marktlokations_id
+
+
+# pylint: disable=too-few-public-methods, too-many-instance-attributes
+@attr.s(auto_attribs=True, kw_only=True)
+class Rechnungsposition(COM):
+    """
+    Über Rechnungspositionen werden Rechnungen strukturiert.
+    In einem Rechnungsteil wird jeweils eine in sich geschlossene Leistung abgerechnet.
+    """
+
+    # required attributes
+    #: Fortlaufende Nummer für die Rechnungsposition
+    positionsnummer: int = attr.ib(validator=attr.validators.instance_of(int))
+
+    lieferung_von: datetime = attr.ib(
+        validator=[attr.validators.instance_of(datetime), check_bis_is_later_than_von]
+    )  #: Start der Lieferung für die abgerechnete Leistung (inklusiv)
+    lieferung_bis: datetime = attr.ib(
+        validator=[attr.validators.instance_of(datetime), check_bis_is_later_than_von]
+    )  #: Ende der Lieferung für die abgerechnete Leistung (exklusiv)
+
+    #: Bezeichung für die abgerechnete Position
+    positionstext: str = attr.ib(validator=attr.validators.instance_of(str))
+
+    #: Die abgerechnete Menge mit Einheit
+    positions_menge: Menge = attr.ib(validator=attr.validators.instance_of(Menge))
+    #: Der Preis für eine Einheit der energetischen Menge
+    einzelpreis: Preis = attr.ib(validator=attr.validators.instance_of(Preis))
+
+    teilsumme_netto: Betrag = attr.ib(validator=attr.validators.instance_of(Betrag))
+    """
+    Das Ergebnis der Multiplikation aus einzelpreis * positionsMenge * (Faktor aus zeitbezogeneMenge).
+    Z.B. 12,60€ * 120 kW * 3/12 (für 3 Monate).
+    """
+    # the cross check in general doesn't work because Betrag and Preis use different enums to describe the currency
+    # see https://github.com/Hochfrequenz/BO4E-python/issues/126
+
+    #: Auf die Position entfallende Steuer, bestehend aus Steuersatz und Betrag
+    teilsumme_steuer: Steuerbetrag = attr.ib(validator=attr.validators.instance_of(Steuerbetrag))
+
+    # optional attributes
+    #: Falls sich der Preis auf eine Zeit bezieht, steht hier die Einheit
+    zeiteinheit: Optional[Zeiteinheit] = attr.ib(
+        default=None, validator=attr.validators.optional(attr.validators.instance_of(Zeiteinheit))
+    )
+
+    #: Kennzeichnung der Rechnungsposition mit der Standard-Artikelnummer des BDEW
+    artikelnummer: Optional[BDEWArtikelnummer] = attr.ib(
+        default=None, validator=attr.validators.optional(attr.validators.instance_of(BDEWArtikelnummer))
+    )
+    #: Marktlokation, die zu dieser Position gehört
+    lokations_id: Optional[str] = attr.ib(default=None, validator=attr.validators.optional(validate_marktlokations_id))
+
+    zeitbezogene_menge: Optional[Menge] = attr.ib(
+        default=None, validator=attr.validators.optional(attr.validators.instance_of(Menge))
+    )
+    """
+    Eine auf die Zeiteinheit bezogene Untermenge.
+    Z.B. bei einem Jahrespreis, 3 Monate oder 146 Tage.
+    Basierend darauf wird der Preis aufgeteilt.
+    """
+    #: Nettobetrag für den Rabatt dieser Position
+    teilrabatt_netto: Optional[Betrag] = attr.ib(
+        default=None, validator=attr.validators.optional(attr.validators.instance_of(Betrag))
+    )
+
+    artikel_id: Optional[ArtikelId] = attr.ib(
+        validator=attr.validators.optional(attr.validators.instance_of(ArtikelId)), default=None
+    )  #: Standardisierte vom BDEW herausgegebene Liste, welche im Strommarkt die BDEW-Artikelnummer ablöst
+
+    def _get_inclusive_start(self) -> datetime:
+        """return the inclusive start (used in the validator)"""
+        return self.lieferung_von
+
+    def _get_exclusive_end(self) -> datetime:
+        """return the exclusive end (used in the validator)"""
+        return self.lieferung_bis
+
+
+class RechnungspositionSchema(COMSchema):
+    """
+    Schema for de-/serialization of RechnungspositionSchema
+    """
+
+    class_name = Rechnungsposition
+
+    # required attributes
+    positionsnummer = fields.Integer()
+    lieferung_von = fields.DateTime()
+    lieferung_bis = fields.DateTime()
+    positionstext = fields.String()
+    positions_menge = fields.Nested(MengeSchema)
+    einzelpreis = fields.Nested(PreisSchema)
+    teilsumme_netto = fields.Nested(BetragSchema)
+    teilsumme_steuer = fields.Nested(SteuerbetragSchema)
+
+    # optional attributes
+    zeiteinheit = EnumField(Zeiteinheit, load_default=None)
+    artikelnummer = EnumField(BDEWArtikelnummer, load_default=None)
+    lokations_id = fields.String(load_default=None)
+    zeitbezogene_menge = fields.Nested(MengeSchema, load_default=None)
+    teilrabatt_netto = fields.Nested(BetragSchema, load_default=None)
+    artikel_id = EnumField(ArtikelId, load_default=None)

src/bo4e/validators.py

@@ -1,6 +1,7 @@
 """
 Contains validators for BO s and COM s classes.
 """
+import re
 from datetime import datetime
 from typing import Optional, Protocol
 
@@ -72,3 +73,48 @@ def check_bis_is_later_than_von(instance: _VonBisType, attribute, value):
 OBIS_PATTERN = r"((1)-((?:[0-5]?[0-9])|(?:6[0-5])):((?:[1-8]|99))\.((?:6|8|9|29))\.([0-9]{1,2})|(7)-((?:[0-5]?[0-9])|(?:6[0-5])):(.{1,2})\.(.{1,2})\.([0-9]{1,2}))"
 #: an attr validator
 obis_validator = attr.validators.matches_re(OBIS_PATTERN)
+
+_malo_id_pattern = re.compile(r"^[1-9][\d]{10}$")
+
+
+# pylint: disable=unused-argument, no-self-use
+def validate_marktlokations_id(self, marktlokations_id_attribute, value):
+    """
+    A validator for marktlokations IDs
+    """
+    if not value:
+        raise ValueError("The marktlokations_id must not be empty.")
+    if not _malo_id_pattern.match(value):
+        raise ValueError(f"The {marktlokations_id_attribute.name} '{value}' does not match {_malo_id_pattern.pattern}")
+    expected_checksum = _get_malo_id_checksum(value)
+    actual_checksum = value[10:11]
+    if expected_checksum != actual_checksum:
+        # pylint: disable=line-too-long
+        raise ValueError(
+            f"The {marktlokations_id_attribute.name} '{value}' has checksum '{actual_checksum}' but '{expected_checksum}' was expected."
+        )
+
+
+def _get_malo_id_checksum(malo_id: str) -> str:
+    """
+    Get the checksum of a marktlokations id.
+    a) Quersumme aller Ziffern in ungerader Position
+    b) Quersumme aller Ziffern auf gerader Position multipliziert mit 2
+    c) Summe von a) und b) d) Differenz von c) zum nächsten Vielfachen von 10 (ergibt sich hier 10, wird die
+       Prüfziffer 0 genommen
+    https://bdew-codes.de/Content/Files/MaLo/2017-04-28-BDEW-Anwendungshilfe-MaLo-ID_Version1.0_FINAL.PDF
+    :param self:
+    :return: the checksum as string
+    """
+    odd_checksum: int = 0
+    even_checksum: int = 0
+    # start counting at 1 to be consistent with the above description
+    # of "even" and "odd" but stop at tenth digit.
+    for i in range(1, 11):
+        digit = malo_id[i - 1 : i]
+        if i % 2 - 1 == 0:
+            odd_checksum += int(digit)
+        else:
+            even_checksum += 2 * int(digit)
+    result: int = (10 - ((even_checksum + odd_checksum) % 10)) % 10
+    return str(result)

tests/test_preis.py

@@ -7,13 +7,15 @@
 from bo4e.enum.preisstatus import Preisstatus
 from bo4e.enum.waehrungseinheit import Waehrungseinheit
 
+example_preis = Preis(wert=Decimal(2.53), einheit=Waehrungseinheit.EUR, bezugswert=Mengeneinheit.KWH)
+
 
 class TestPreis:
     def test_preis_only_required(self):
         """
         Test de-/serialisation of Preis (only has required attributes).
         """
-        preis = Preis(wert=Decimal(2.53), einheit=Waehrungseinheit.EUR, bezugswert=Mengeneinheit.KWH)
+        preis = example_preis
 
         schema = PreisSchema()
         json_string = schema.dumps(preis, ensure_ascii=False)

tests/test_rechnungsposition.py

@@ -0,0 +1,50 @@
+from datetime import datetime, timezone
+
+import pytest  # type:ignore[import]
+
+from bo4e.com.rechnungsposition import Rechnungsposition, RechnungspositionSchema
+from bo4e.enum.artikelid import ArtikelId
+from bo4e.enum.bdewartikelnummer import BDEWArtikelnummer
+from bo4e.enum.zeiteinheit import Zeiteinheit
+from tests.serialization_helper import assert_serialization_roundtrip  # type:ignore[import]
+from tests.test_betrag import example_betrag  # type:ignore[import]
+from tests.test_menge import example_menge  # type:ignore[import]
+from tests.test_preis import example_preis  # type:ignore[import]
+from tests.test_steuerbetrag import example_steuerbetrag  # type:ignore[import]
+
+
+class TestRechnungsposition:
+    @pytest.mark.parametrize(
+        "rechnungsposition",
+        [
+            pytest.param(
+                Rechnungsposition(
+                    positionsnummer=1,
+                    lieferung_von=datetime(2021, 3, 15, tzinfo=timezone.utc),
+                    lieferung_bis=datetime(2022, 3, 15, tzinfo=timezone.utc),
+                    positionstext="Besonders wertvolle Rechnungsposition",
+                    zeiteinheit=Zeiteinheit.JAHR,
+                    artikelnummer=BDEWArtikelnummer.AUSGLEICHSENERGIE_UNTERDECKUNG,
+                    lokations_id="51238696781",
+                    positions_menge=example_menge,
+                    zeitbezogene_menge=example_menge,
+                    einzelpreis=example_preis,
+                    teilsumme_netto=example_betrag,
+                    teilrabatt_netto=example_betrag,
+                    teilsumme_steuer=example_steuerbetrag,
+                    artikel_id=ArtikelId.ARTIKEL_2017004,
+                ),
+                id="maximal attributes",
+            )
+        ],
+    )
+    def test_serialization_roundtrip(self, rechnungsposition):
+        """
+        Test de-/serialisation
+        """
+        assert_serialization_roundtrip(rechnungsposition, RechnungspositionSchema())
+
+    def test_missing_required_attribute(self):
+        with pytest.raises(TypeError) as excinfo:
+            _ = Rechnungsposition()
+        assert "missing 8 required" in str(excinfo.value)

tests/test_steuerbetrag.py

@@ -7,18 +7,20 @@
 from bo4e.enum.waehrungscode import Waehrungscode
 from tests.serialization_helper import assert_serialization_roundtrip  # type:ignore[import]
 
+example_steuerbetrag = Steuerbetrag(
+    steuerkennzeichen=Steuerkennzeichen.UST_7,
+    basiswert=Decimal(100),
+    steuerwert=Decimal(19),
+    waehrung=Waehrungscode.EUR,
+)
+
 
 class TestSteuerbetrag:
     @pytest.mark.parametrize(
         "steuerbetrag, expected_json_dict",
         [
             pytest.param(
-                Steuerbetrag(
-                    steuerkennzeichen=Steuerkennzeichen.UST_7,
-                    basiswert=Decimal(100),
-                    steuerwert=Decimal(19),
-                    waehrung=Waehrungscode.EUR,
-                ),
+                example_steuerbetrag,
                 {"steuerkennzeichen": "UST_7", "basiswert": "100", "steuerwert": "19", "waehrung": "EUR"},
             ),
         ],