# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
"""Export emiflow reference data as a Business Central RapidStart package.
RapidStart is BC's bulk-import format: a gzipped XML document with one
``<TableNameList>`` per BC table, each holding the row data plus
field metadata (``TableID``, ``PageID``, ``ValidateField``,
``ProcessingOrder``, ``PrimaryKey``). The
:class:`ModelSerializerMeta` metaclass registers each subclass with
:class:`SerializerSwitch` so :class:`RapidstartExporter` can iterate
the registered models in dependency order and assemble the complete
``DataList`` element.
Numbers in BC have a maximum length (20 chars), so primary keys are
derived from the last seven characters of each Ferrosoft UUID via
:func:`derive_bc_number`.
"""
import gzip
import xml.etree.ElementTree as ET
from io import BytesIO
from typing import Iterable, Any, Type
from uuid import UUID
from ferrosoft.apps.emiflow.models import (
EnergyCarrier,
EmissionsCatalog,
Emitter,
TransportOperation,
TreatmentOperation,
TreatmentOperationInputLine,
TreatmentOperationOutputLine,
TreatmentOperationStatus,
)
from ferrosoft.apps.ferrobase.models import MeasurementUnit
[docs]
def derive_bc_number(default_prefix: str):
"""Return a function that turns a UUID into a BC-friendly *No.* string.
The returned closure derives a string of the form
``"{prefix}{LAST_7_HEX_UPPER}"``; ``prefix`` defaults to
``default_prefix`` but each call can override it (e.g. to reuse
another model's prefix when emitting a foreign-key reference).
"""
def derive(pk: UUID, prefix: str | None = None):
return "%s%s" % (prefix or default_prefix, str(pk)[-7:].upper())
return derive
# noinspection PyPep8Naming
[docs]
def Element(
tag: str,
text: str | None = None,
children: Iterable[ET.Element] | None = None,
attrib: dict | None = None,
**extra,
) -> ET.Element:
"""Construct an ``ElementTree`` element in one expression.
Convenience wrapper that accepts ``text`` and ``children`` inline
so RapidStart documents read close to the XML they produce.
"""
element = ET.Element(tag, attrib or {}, **extra)
if text is not None:
element.text = text
if children is not None:
element.extend(children)
return element
[docs]
class ClassAttributes:
"""Validator for class-attribute presence used by :class:`ModelSerializerMeta`."""
def __init__(self, attrs: dict[str, Any]):
"""Capture the metaclass ``attrs`` dict so it can be checked later."""
self._attrs = attrs
[docs]
def require_all(self, *required) -> "ClassAttributes":
"""Raise unless every named attribute is present in the wrapped dict.
Returns ``self`` so calls can be chained when more validators
are added later.
"""
missing = list(filter(lambda key: key not in self._attrs, required))
if len(missing) > 0:
raise RuntimeError(
"Required class attributes missing: %s" % ", ".join(missing)
)
return self
[docs]
class ModelSerializer(metaclass=ModelSerializerMeta):
"""Base class for a serializer that emits one BC RapidStart table.
Subclasses declare ``model``, ``table_id``, ``table_name``, and
``page_id`` as class attributes and implement
:meth:`serialize_fields` to return the per-record XML children.
The first record gets full field metadata
(``ValidateField``/``ProcessingOrder``/``PrimaryKey``); later
records re-use it implicitly. Set ``primary_key`` to a list of
field names when the table has a composite primary key.
"""
abstract = True
model = None
table_id: int = None
table_name: str = None
page_id: int = None
number_prefix: str = None
def __init__(self):
if self.number_prefix is not None:
self.derive_number = derive_bc_number(self.number_prefix)
else:
self.derive_number = derive_bc_number("EMI")
[docs]
def serialize_fields(self, instance) -> Iterable[ET.Element]:
"""Return the ordered field elements for a single ``instance``."""
raise NotImplementedError
[docs]
def get_queryset(self):
"""Return the queryset whose rows should be exported."""
return self.model.objects.all()
[docs]
def get_all(self) -> ET.Element:
"""Build the ``<TableNameList>`` element with one child per queryset row."""
list_element = self.get_list_element()
list_element.extend(
[
Element(
self.table_name,
children=self._add_bc_field_metadata(
i, self.serialize_fields(instance)
),
)
for i, instance in enumerate(self.get_queryset())
]
)
return list_element
[docs]
def get_list_element(self):
"""Return the empty ``<TableNameList>`` shell with table/page metadata."""
children = [Element("TableID", text=str(self.table_id))]
if self.page_id != 0:
children.append(Element("PageID", text=str(self.page_id)))
return Element("%sList" % self.table_name, children=children)
def _add_bc_field_metadata(
self, record_number: int, record_fields: Iterable[ET.Element]
) -> Iterable[ET.Element]:
"""Attach validate / order / PK attributes to the first record's fields only."""
if record_number != 0:
# Metadata are added only to the first record.
return record_fields
for i, field in enumerate(record_fields):
field.attrib["ValidateField"] = "1"
field.attrib["ProcessingOrder"] = str(i + 1)
field.attrib["PrimaryKey"] = (
"1" if self._is_primary_key(i, field.tag) else "0"
)
return record_fields
def _is_primary_key(self, field_number: int, field_name: str) -> bool:
"""Decide whether the given field is part of the table's primary key.
When ``primary_key`` is a list on the subclass, membership is
determined by field name; otherwise only the first field is
treated as primary.
"""
pk = getattr(self, "primary_key", None)
if isinstance(pk, list):
return field_name in pk
return field_number == 0
[docs]
class SerializerSwitch:
"""Registry mapping :class:`Model` classes to their :class:`ModelSerializer`."""
_SERIALIZERS = {}
[docs]
@classmethod
def register(cls, model, serializer: Type[ModelSerializer]):
"""Register ``serializer`` as the serializer for ``model``."""
cls._SERIALIZERS[model.__name__] = serializer
[docs]
@classmethod
def serialize(cls, model) -> ET.Element:
"""Run the registered serializer for ``model`` and return its element.
Raises :class:`RuntimeError` when no serializer is registered.
"""
if model.__name__ not in cls._SERIALIZERS:
raise RuntimeError(f"Serializer for {model.__name__} not found")
serializer = cls._SERIALIZERS[model.__name__]
return serializer().get_all()
[docs]
class MeasurementUnitSerializer(ModelSerializer):
"""Serializer mapping ferrobase :class:`MeasurementUnit` to BC ``Unit of Measure``."""
model = MeasurementUnit
table_id = 204
page_id = 209
table_name = "UnitofMeasure"
[docs]
def serialize_fields(self, instance) -> Iterable[ET.Element]:
return [
Element("Code", instance.symbol.upper()),
Element("Description", instance.name),
]
[docs]
def get_queryset(self):
return super().get_queryset().order_by("symbol")
[docs]
class CatalogSerializer(ModelSerializer):
"""Serializer mapping :class:`EmissionsCatalog` to the ``CatalogFER`` table."""
model = EmissionsCatalog
table_name = "CatalogFER"
table_id = 73060577
page_id = 0
number_prefix = "CAT"
[docs]
def serialize_fields(self, instance: EmissionsCatalog) -> Iterable[ET.Element]:
return [
Element("No", self.derive_number(instance.pk)),
Element("PublicID", str(instance.pk)),
Element("Name", instance.name),
Element("DataSource", str(instance.data_source)),
Element("Copyright", instance.copyright),
Element("LicenseURI", instance.license_uri),
Element("LicenseText", instance.license_text),
]
[docs]
class EnergyCarrierSerializer(ModelSerializer):
"""Serializer mapping :class:`EnergyCarrier` to the ``EnergyCarrierFER`` table."""
model = EnergyCarrier
table_name = "EnergyCarrierFER"
table_id = 73060578
page_id = 0
number_prefix = "ENR"
[docs]
def serialize_fields(self, instance: EnergyCarrier) -> Iterable[ET.Element]:
return [
Element("No", self.derive_number(instance.pk)),
Element("PublicID", str(instance.pk)),
Element("Name", instance.name),
Element(
"CatalogNo",
self.derive_number(
instance.catalog_id, CatalogSerializer.number_prefix
),
),
Element("Composition", instance.composition),
Element("Application", instance.application),
Element("LowerHeating", str(instance.lower_heating or 0)),
Element("Density", str(instance.density or 0)),
]
[docs]
class EmitterSerializer(ModelSerializer):
"""Serializer mapping :class:`Emitter` to the ``EmitterFER`` table."""
model = Emitter
table_name = "EmitterFER"
table_id = 73060579
page_id = 0
number_prefix = "EMT"
[docs]
def serialize_fields(self, instance: Emitter) -> Iterable[ET.Element]:
intensity = instance.emission_intensity
return [
Element("No", self.derive_number(instance.pk)),
Element("PublicID", str(instance.pk)),
Element("Name", instance.name),
Element(
"CatalogNo",
self.derive_number(
instance.catalog_id, CatalogSerializer.number_prefix
),
),
Element("Epitype", str(instance.epitype)),
Element("VehicleType", str(instance.vehicle_type or 0)),
Element(
"EnergyCarrierNo",
""
if instance.energy_carrier_id is None
else self.derive_number(
instance.energy_carrier_id, EnergyCarrierSerializer.number_prefix
),
),
Element(
"EmissionIntOperation",
str(0 if intensity is None else intensity.operation_value),
),
Element(
"EmissionIntEnergyProv",
str(0 if intensity is None else intensity.energy_provision_value),
),
Element(
"EmissionIntUnit",
"" if intensity is None else intensity.unit.symbol.upper(),
),
]
[docs]
class TransportOperationSerializer(ModelSerializer):
"""Serializer mapping :class:`TransportOperation` to the ``TransportOperationFER`` table."""
model = TransportOperation
table_name = "TransportOperationFER"
table_id = 73060580
page_id = 0
[docs]
def serialize_fields(self, instance: TransportOperation) -> Iterable[ET.Element]:
intensity = instance.emission_intensity
return [
Element("No", instance.reference),
Element("PublicID", str(instance.pk)),
Element("Name", instance.name),
Element("ActivityType", str(instance.activity_type)),
Element(
"EmitterNo",
self.derive_number(
instance.emitter_id, EmitterSerializer.number_prefix
),
),
Element(
"EnergyCarrierNo",
self.derive_number(
instance.energy_carrier_id, EnergyCarrierSerializer.number_prefix
),
),
Element(
"EmissionIntOperation",
str(0 if intensity is None else intensity.operation_value),
),
Element(
"EmissionIntEnergyProv",
str(0 if intensity is None else intensity.energy_provision_value),
),
Element(
"EmissionIntUnit",
"" if intensity is None else intensity.unit.symbol.upper(),
),
]
[docs]
class TreatmentOperationSerializer(ModelSerializer):
"""Serializer mapping :class:`TreatmentOperation` to ``TreatmentOperationFER``.
Only certified operations are exported — open operations are still
being authored and would mislead downstream BC bookings.
"""
model = TreatmentOperation
table_name = "TreatmentOperationFER"
table_id = 73060581
page_id = 0
[docs]
def get_queryset(self):
"""Limit the export to operations in the ``CERTIFIED`` status."""
return super().get_queryset().filter(status=TreatmentOperationStatus.CERTIFIED)
[docs]
def serialize_fields(self, instance: TreatmentOperation) -> Iterable[ET.Element]:
intensity = instance.emission_intensity
return [
Element("No", instance.reference),
Element("PublicID", str(instance.pk)),
Element("Name", instance.name),
Element("AllocationMethod", str(instance.allocation_method)),
Element(
"EmitterNo",
self.derive_number(
instance.emitter_id, EmitterSerializer.number_prefix
),
),
Element(
"EnergyCarrierNo",
self.derive_number(
instance.energy_carrier_id, EnergyCarrierSerializer.number_prefix
),
),
Element(
"EmissionIntOperation",
str(0 if intensity is None else intensity.operation_value),
),
Element(
"EmissionIntEnergyProv",
str(0 if intensity is None else intensity.energy_provision_value),
),
Element(
"EmissionIntUnit",
"" if intensity is None else intensity.unit.symbol.upper(),
),
]
[docs]
class TreatmentOperationOutputLineSerializer(ModelSerializer):
"""Serializer for :class:`TreatmentOperationOutputLine` rows (composite PK)."""
model = TreatmentOperationOutputLine
table_name = "TreatmentOpOutputLineFER"
table_id = 73060583
page_id = 0
primary_key = ["OperationNo", "MaterialNo"]
[docs]
def serialize_fields(
self, instance: TreatmentOperationOutputLine
) -> Iterable[ET.Element]:
return [
Element("OperationNo", str(instance.operation.reference)),
Element("MaterialNo", instance.material.code),
Element("PublicID", str(instance.pk)),
Element("Weight", str(instance.weight_value)),
Element("WeightUnitCode", instance.weight_unit.symbol.upper()),
Element("EmissionRatio", str(instance.allocated_fraction)),
]
[docs]
class RapidstartExporter:
"""Assemble and gzip the complete RapidStart ``DataList`` package.
Iterates :attr:`models` in dependency order (units → catalog →
energy carriers → emitters → operations → operation lines) so that
foreign-key targets always precede the rows that reference them
in the resulting XML.
"""
def __init__(self):
self.validate_field = True
self.models = [
MeasurementUnit,
EmissionsCatalog,
EnergyCarrier,
Emitter,
TransportOperation,
TreatmentOperation,
TreatmentOperationInputLine,
TreatmentOperationOutputLine,
]
[docs]
@staticmethod
def serialize(data_list: ET.Element) -> bytes:
"""Serialise ``data_list`` to UTF-8 XML and return its gzipped bytes.
``mtime=0`` keeps the gzip output reproducible so two runs
with the same data produce byte-identical packages.
"""
tree = ET.ElementTree(data_list)
buffer = BytesIO()
# noinspection PyTypeChecker
tree.write(buffer, encoding="utf-8", xml_declaration=True)
buffer.seek(0)
return gzip.compress(buffer.getvalue(), mtime=0)
[docs]
def get_datalist(self, package_name: str, code: str) -> ET.Element:
"""Build the top-level ``<DataList>`` element for the package.
Args:
package_name: Human-readable package name shown in BC.
code: RapidStart package code (typically alphanumeric).
Returns:
ET.Element: The fully populated ``<DataList>`` element
with one table list per registered model.
"""
data_list = Element(
"DataList",
attrib={
"MinCountForAsyncImport": "5",
"ExcludeConfigTables": "1",
"ProductVersion": "",
"PackageName": package_name,
"Code": code,
},
)
data_list.extend([SerializerSwitch.serialize(model) for model in self.models])
return data_list