"""Import utilities for catalog files."""
import json
from enum import StrEnum
from json import JSONDecodeError
from typing import Tuple, Iterable
from jsonschema.validators import Draft202012Validator
from ferrosoft.apps.emiflow.models import (
VehicleType,
EmissionCategory,
LoadCharacteristics,
BookingStatus,
DistanceType,
DataSource,
EmitterEpitype,
OperationEpitype,
IntensityCapableEntity,
)
from ferrosoft.apps.ferrobase.models import (
MeasurementUnit,
Country,
)
from ferrosoft.apps.ferrobase.services.static_files import static_file_name
[docs]
class CatalogSchema(StrEnum):
BASE = "https://emiflow.de/schema/catalog.schema.json"
TOC = "https://emiflow.de/schema/toc.schema.json"
TRANSPORT = "https://emiflow.de/schema/transport.schema.json"
[docs]
class CatalogConversionError(Exception):
pass
[docs]
class CatalogValidationError(Exception):
pass
[docs]
class IdResolver:
def __init__(self):
self.unit_cache = {}
[docs]
def lookup_unit(self, symbol: str) -> str:
unit_id = self.unit_cache.get(symbol, None)
if unit_id is not None:
return unit_id
result = MeasurementUnit.objects.filter(symbol=symbol).values("id").first()
if result is None:
raise CatalogConversionError("Unit not found: %s" % symbol)
unit_id = str(result["id"])
if unit_id == "":
raise CatalogConversionError("Unit not found: %s" % symbol)
self.unit_cache[symbol] = unit_id
return unit_id
[docs]
@staticmethod
def lookup_reference(id_cache: dict, external_id: str) -> str:
model_id = id_cache.get(external_id, None)
if model_id is None:
raise CatalogConversionError("Invalid reference %s" % external_id)
return model_id
[docs]
def emiflow_static_file(file_name: str) -> str:
"""Find static file inside emiflow app.
Raises:
RuntimeError if the file is not found.
"""
return static_file_name("emiflow", file_name)
[docs]
def convert_catalog_files(
catalog_file_names: Iterable[str], schema_base_name: str
) -> list:
fixtures = []
for catalog_path in catalog_file_names:
with open(catalog_path, mode="rb") as catalog_file:
catalog = json.load(catalog_file)
schema, errors = validate_catalog(catalog, schema_base_name)
if len(errors) > 0:
raise CatalogValidationError(
"Invalid catalog file:\n" + "\n".join(errors)
)
converter = get_converter(schema)
fixtures.append(converter(catalog))
# Flatten fixtures.
return [x for xs in fixtures for x in xs]
[docs]
def validate_catalog(
catalog: dict, schema_base_name: str
) -> Tuple[CatalogSchema, list]:
"""Validate catalog dict against emiflow catalog schema."""
schema_path = emiflow_static_file(schema_base_name)
with open(schema_path, "rb") as schema_file:
try:
schema = json.load(schema_file)
except JSONDecodeError as e:
raise RuntimeError("Invalid schema file: %s" % schema_path) from e
schema_id = schema.get("$id", None)
if schema_id not in CatalogSchema:
raise CatalogValidationError("Unsupported schema: %s" % schema_id)
validator = Draft202012Validator(schema)
return CatalogSchema(schema_id), [
"%s at %s" % (err.message, err.json_path)
for err in validator.iter_errors(instance=catalog)
]
def _convert_enum(enum_type, external_value: str):
if hasattr(enum_type, external_value):
return enum_type[external_value]
raise CatalogConversionError(
"Invalid %s: %s" % (enum_type.__name__, external_value)
)
[docs]
def catalog_to_fixtures(catalog: dict) -> Iterable[dict]:
"""Convert (presumed valid) catalog to fixtures list."""
catalog_id = catalog["id"]
resolver = IdResolver()
fixtures = [
{
"model": "emiflow.emissionscatalog",
"pk": catalog_id,
"fields": {
"data_source": _convert_enum(DataSource, catalog["dataSource"]).value,
"name": catalog["name"],
"copyright": catalog["copyright"],
"license_uri": catalog["license"].get("uri", ""),
"license_text": catalog["license"].get("text", ""),
},
},
]
for vehicle in catalog.get("vehicles", []):
fixtures.append(
{
"model": "emiflow.emitter",
"pk": vehicle["id"],
"fields": {
"catalog": catalog_id,
"epitype": EmitterEpitype.VEHICLE,
"name": vehicle["name"],
"vehicle_type": _convert_enum(
VehicleType, vehicle.get("vehicleType", "ROAD")
).value,
"read_only": True,
},
}
)
for vehicle in catalog.get("hubEquipment", []):
fixtures.append(
{
"model": "emiflow.emitter",
"pk": vehicle["id"],
"fields": {
"catalog": catalog_id,
"epitype": _convert_enum(
EmitterEpitype, vehicle.get("type", "MATERIAL_HANDLER")
),
"name": vehicle["name"],
"vehicle_type": None,
"read_only": True,
},
}
)
for carrier in catalog.get("energyCarriers", []):
fixtures.append(
{
"model": "emiflow.energycarrier",
"pk": carrier["id"],
"fields": {
"catalog": catalog_id,
"name": carrier["name"],
"composition": carrier.get("composition", ""),
"application": carrier.get("application", ""),
"lower_heating": carrier["lowerHeating"],
"density": carrier["density"],
"read_only": True,
},
}
)
for factor in carrier["factors"]:
fixtures.append(
{
"model": "emiflow.emissionfactor",
"pk": factor["id"],
"fields": {
"energy_carrier": carrier["id"],
"value": factor["value"],
"unit": resolver.lookup_unit(factor["unit"]),
"category": _convert_enum(
EmissionCategory, factor["category"]
).value,
"read_only": True,
},
}
)
return fixtures
[docs]
def toc_to_fixtures(catalog: dict) -> Iterable[dict]:
fixtures = []
resolver = IdResolver()
for operation in catalog["operations"]:
intensity = operation.get("emissionIntensity", None)
if isinstance(intensity, dict):
fixtures.append(
{
"model": "emiflow.EmissionIntensity",
"pk": intensity["id"],
"fields": {
"owner_type": int(IntensityCapableEntity.TRANSPORT_OPERATION),
"owner_id": operation["id"],
# Output variables
"operation_value": intensity["operationIntensity"],
"energy_provision_value": intensity["energyProvisionIntensity"],
"unit": resolver.lookup_unit(intensity["intensityUnit"]),
# Input variables
"factor_operation_value": intensity["operationFactor"],
"factor_energy_provision_value": intensity[
"energyProvisionFactor"
],
"quantity_value": intensity["quantity"]["value"],
"quantity_unit": resolver.lookup_unit(
intensity["quantity"]["unit"]
),
"mass_value": intensity["mass"]["value"],
"mass_unit": resolver.lookup_unit(intensity["mass"]["unit"]),
"distance_value": intensity["distance"]["value"],
"distance_unit": resolver.lookup_unit(
intensity["distance"]["unit"]
),
"read_only": True,
},
}
)
fixtures.append(
{
"model": "emiflow.TransportOperation",
"pk": operation["id"],
"fields": {
"reference": operation["reference"],
"name": operation["name"],
"activity_type": _convert_enum(
OperationEpitype, operation["activityType"]
).value,
"emitter": operation["emitter"],
"energy_carrier": operation["energyCarrier"],
"emission_intensity": (
None if intensity is None else intensity["id"]
),
"read_only": True,
},
}
)
return fixtures
def _address_fields(address: dict) -> dict:
return {
"salutation": address.get("salutation", None),
"first_name": address.get("firstName", ""),
"last_name": address.get("lastName", ""),
"company": address.get("company", ""),
"additional_lines": address.get("additionalLines", ""),
"street": address["street"],
"house_no": address.get("houseNumber", ""),
"city": address["city"],
"postal_code": address["postalCode"],
"country": str(Country.get_by_iso_code(address["country"]).pk),
}
def _coordinates_fields(coordinates: dict) -> dict:
return {
"latitude": coordinates["lat"],
"longitude": coordinates["lon"],
}
def _tce_point_fields(point: dict) -> dict:
return _coordinates_fields(point["position"]) | _address_fields(point["address"])
[docs]
def transport_to_fixtures(catalog: dict):
fixtures = []
resolver = IdResolver()
for material in catalog["materials"]:
fixtures.append(
{
"model": "ferrobase.Material",
"pk": material["id"],
"fields": {
"code": material["reference"],
"name": material["name"],
},
}
)
for partner in catalog["partners"]:
fixtures.append(
{
"model": "ferrobase.BusinessPartner",
"pk": partner["id"],
"fields": {
"name": partner["name"],
"first_name": partner.get("firstName", ""),
"last_name": partner.get("lastName", ""),
"company": partner.get("company", ""),
"additional_lines": partner.get("additionalLines", ""),
"street": partner["street"],
"house_no": partner.get("houseNumber", ""),
"postal_code": partner["postalCode"],
"city": partner["city"],
"country": str(Country.get_by_iso_code(partner["country"]).pk),
},
}
)
for chain in catalog["transportChains"]:
chain_id = chain["id"]
fixtures.append(
{
"model": "emiflow.TransportChain",
"pk": chain_id,
"fields": {
"reference": chain["reference"],
"booking_date": chain["bookingDate"],
"status": _convert_enum(BookingStatus, chain["status"]).value,
},
}
)
for element in chain["elements"]:
dist = element["activityDistance"]
fixtures.append(
{
"model": "emiflow.TransportChainElement",
"pk": element["id"],
"fields": {
"chain": chain_id,
"position": element["position"],
"toc": element["toc"],
"operation": element["operation"],
"activity_distance_type": _convert_enum(
DistanceType, dist["type"]
).value,
"activity_distance_value": dist["value"],
"activity_distance_unit": resolver.lookup_unit(dist["unit"]),
"distance_adjustment_factor": element["daf"],
},
}
)
origin = dist.get("origin", None)
if origin is not None:
fixtures.append(
{
"model": "emiflow.TCEGeoOrigin",
"pk": origin["id"],
"fields": _tce_point_fields(origin) | {"tce": element["id"]},
}
)
destination = dist.get("destination", None)
if destination is not None:
fixtures.append(
{
"model": "emiflow.TCEGeoDestination",
"pk": destination["id"],
"fields": _tce_point_fields(destination)
| {"tce": element["id"]},
}
)
for freight_item in chain["freightItems"]:
weight = freight_item["weight"]
fixtures.append(
{
"model": "emiflow.FreightItem",
"pk": freight_item["id"],
"fields": {
"chain": chain_id,
"weight_value": weight["value"],
"weight_unit": resolver.lookup_unit(weight["unit"]),
"material": freight_item["material"],
},
}
)
return fixtures
_converters = {
CatalogSchema.BASE: catalog_to_fixtures,
CatalogSchema.TOC: toc_to_fixtures,
CatalogSchema.TRANSPORT: transport_to_fixtures,
}
[docs]
def get_converter(schema_id: CatalogSchema):
if schema_id in _converters:
return _converters[schema_id]
raise CatalogConversionError("Converter not defined for schema %s" % schema_id)