ferrosoft.apps.emiflow.services package

Service layer for the emiflow application.

This package holds the fat services / thin views business logic for emission calculation, reporting, integration, and supporting concerns:

• builder — fluent builders that compose TransportChains and their related rows.

• businesscentral — adapters for the Microsoft Dynamics 365 Business Central connector.

• emissions — transport-chain emission calculator and booking service.

• geographic — strategy for resolving activity distance by distance type.

• pcf — product carbon footprint roll-up across all chains carrying a material.

• reporting — document generators that turn booking records into renderable reports.

• session — per-request storage of the current transport chain in the user session.

• tonnage — per-tenant / per-organisation tonnage queries used for subscription accounting.

• treatment — treatment emission calculator and booking service.

• update — periodic update hooks that seed predefined lifecycle categories per tenant.

• vehicle — heuristic that guesses a VehicleType from freight weight.

• visitor — visitor implementations executed against TransportChain walks (debug printer, geocoding/routing).

Subpackages

• ferrosoft.apps.emiflow.services.businesscentral package
  • Submodules
  • ferrosoft.apps.emiflow.services.businesscentral.connector module
    • BcConnectorClient
      • BcConnectorClient.get_instance()
      • BcConnectorClient.post_emission()
    • BcConnectorHTTPClient
      • BcConnectorHTTPClient.Config
      • BcConnectorHTTPClient.get_instance()
      • BcConnectorHTTPClient.post_emission()
    • ClientError
    • Error
  • ferrosoft.apps.emiflow.services.businesscentral.rapidstart module
    • CatalogSerializer
      • CatalogSerializer.model
      • CatalogSerializer.number_prefix
      • CatalogSerializer.page_id
      • CatalogSerializer.serialize_fields()
      • CatalogSerializer.table_id
      • CatalogSerializer.table_name
    • ClassAttributes
      • ClassAttributes.require_all()
    • Element()
    • EmitterSerializer
      • EmitterSerializer.model
      • EmitterSerializer.number_prefix
      • EmitterSerializer.page_id
      • EmitterSerializer.serialize_fields()
      • EmitterSerializer.table_id
      • EmitterSerializer.table_name
    • EnergyCarrierSerializer
      • EnergyCarrierSerializer.model
      • EnergyCarrierSerializer.number_prefix
      • EnergyCarrierSerializer.page_id
      • EnergyCarrierSerializer.serialize_fields()
      • EnergyCarrierSerializer.table_id
      • EnergyCarrierSerializer.table_name
    • MeasurementUnitSerializer
      • MeasurementUnitSerializer.get_queryset()
      • MeasurementUnitSerializer.model
      • MeasurementUnitSerializer.page_id
      • MeasurementUnitSerializer.serialize_fields()
      • MeasurementUnitSerializer.table_id
      • MeasurementUnitSerializer.table_name
    • ModelSerializer
      • ModelSerializer.get_all()
      • ModelSerializer.get_list_element()
      • ModelSerializer.get_queryset()
      • ModelSerializer.model
      • ModelSerializer.number_prefix
      • ModelSerializer.page_id
      • ModelSerializer.serialize_fields()
      • ModelSerializer.table_id
      • ModelSerializer.table_name
    • ModelSerializerMeta
    • RapidstartExporter
      • RapidstartExporter.get_datalist()
      • RapidstartExporter.serialize()
    • SerializerSwitch
      • SerializerSwitch.register()
      • SerializerSwitch.serialize()
    • TransportOperationSerializer
      • TransportOperationSerializer.model
      • TransportOperationSerializer.page_id
      • TransportOperationSerializer.serialize_fields()
      • TransportOperationSerializer.table_id
      • TransportOperationSerializer.table_name
    • TreatmentOperationInputLineSerializer
      • TreatmentOperationInputLineSerializer.get_queryset()
      • TreatmentOperationInputLineSerializer.model
      • TreatmentOperationInputLineSerializer.page_id
      • TreatmentOperationInputLineSerializer.primary_key
      • TreatmentOperationInputLineSerializer.serialize_fields()
      • TreatmentOperationInputLineSerializer.table_id
      • TreatmentOperationInputLineSerializer.table_name
    • TreatmentOperationOutputLineSerializer
      • TreatmentOperationOutputLineSerializer.model
      • TreatmentOperationOutputLineSerializer.page_id
      • TreatmentOperationOutputLineSerializer.primary_key
      • TreatmentOperationOutputLineSerializer.serialize_fields()
      • TreatmentOperationOutputLineSerializer.table_id
      • TreatmentOperationOutputLineSerializer.table_name
    • TreatmentOperationSerializer
      • TreatmentOperationSerializer.get_queryset()
      • TreatmentOperationSerializer.model
      • TreatmentOperationSerializer.page_id
      • TreatmentOperationSerializer.serialize_fields()
      • TreatmentOperationSerializer.table_id
      • TreatmentOperationSerializer.table_name
    • derive_bc_number()

Submodules

ferrosoft.apps.emiflow.services.builder module

Fluent builders for assembling transport chains and related rows.

Each ModelBuilder subclass wraps a Django model instance and exposes chainable setter methods so that complex transport chains can be constructed declaratively. Builders compose via BuilderOfBuilders (a chain holds builders for its elements and freight items), and can either save() rows one-by-one or defer everything into a BulkCreator for one bulk upsert per model class.

class ferrosoft.apps.emiflow.services.builder.BuilderOfBuilders

Bases: ModelBuilder

Aggregate of child builders that fans save/add_bulk out to each.

add_bulk(bulk)

Register every child’s instance with bulk.

append(builder)

Add builder to the aggregate and return it for chaining.

Return type:

ModelBuilder

save(*args, **kwargs)

Save every child builder.

class ferrosoft.apps.emiflow.services.builder.BulkCreator(create_order)

Bases: object

Collects unsaved model instances and upserts them in declared order.

The model classes are grouped by name; on upsert() the creator iterates through create_order so foreign keys are satisfied (e.g. TransportChain before TransportChainElement) and each model’s manager-level upsert is invoked once with all instances of that class.

add_model(model)

Add model to the pending bulk for its class.

upsert()

Run Manager.upsert(...) for each pending model class in declared order.

Returns:

A mapping of class name to the list of upserted instances. Classes with no pending rows are omitted.

Return type:

dict[str, list]

Raises:

RuntimeError – When a model class’s manager has no upsert method.

class ferrosoft.apps.emiflow.services.builder.FreightItemBuilder(item)

Bases: ModelBuilder

Fluent builder for one FreightItem.

add_bulk(bulk)

Register the built instance with bulk for deferred persistence.

Return type:

FreightItem

chain(chain)

Return type:

FreightItemBuilder

material(material)

Return type:

FreightItemBuilder

classmethod new()

Start a fresh builder around an empty FreightItem.

Return type:

FreightItemBuilder

save(*args, **kwargs)

Persist the built model instance and return it.

Return type:

FreightItem

weight(weight)

Return type:

FreightItemBuilder

class ferrosoft.apps.emiflow.services.builder.GeoPointBuilder(point)

Bases: ModelBuilder

Fluent builder around a TCEGeoPoint (origin or destination).

add_bulk(bulk)

Register the built instance with bulk for deferred persistence.

Return type:

TCEGeoPoint

address(address)

Return type:

GeoPointBuilder

coordinates(coords)

Return type:

GeoPointBuilder

classmethod new()

Start a fresh builder around an empty TCEGeoPoint.

Return type:

GeoPointBuilder

save(*args, **kwargs)

Persist the built model instance and return it.

Return type:

TCEGeoPoint

class ferrosoft.apps.emiflow.services.builder.MaterialBuilder(material)

Bases: ModelBuilder

Fluent builder around a Material.

add_bulk(bulk)

Register the built instance with bulk for deferred persistence.

Return type:

Material

code(code)

Return type:

MaterialBuilder

name(name)

Return type:

MaterialBuilder

classmethod new()

Start a fresh builder around an empty Material.

Return type:

MaterialBuilder

save(*args, **kwargs)

Persist the built model instance and return it.

Return type:

Material

class ferrosoft.apps.emiflow.services.builder.ModelBuilder

Bases: ABC

Abstract fluent builder around a single Django model instance.

Subclasses expose chainable setter methods returning self and must implement save() to persist the instance individually or add_bulk() to defer persistence into a BulkCreator.

add_bulk(bulk)

Register the built instance with bulk for deferred persistence.

abstractmethod save(*args, **kwargs)

Persist the built model instance and return it.

class ferrosoft.apps.emiflow.services.builder.TransportChainBuilder(chain)

Bases: ModelBuilder

Top-level fluent builder that composes the whole chain.

Owns nested builders for chain elements (auto-numbered by insertion order) and freight items. save cascades to all children so the entire object graph can be persisted from a single call.

add_bulk(bulk)

Register the built instance with bulk for deferred persistence.

Return type:

TransportChain

add_element()

Append a new chain element (auto-numbered by insertion order) and return its builder.

Return type:

TransportChainElementBuilder

add_freight_item()

Append a new freight item to the chain and return its builder.

Return type:

FreightItemBuilder

booking_date(booking_date)

Return type:

TransportChainBuilder

consignee(partner)

Return type:

TransportChainBuilder

consignor(partner)

Return type:

TransportChainBuilder

incoterms(incoterms)

Return type:

TransportChainBuilder

classmethod new()

Start a fresh builder around an empty TransportChain.

Return type:

TransportChainBuilder

reference(reference)

Return type:

TransportChainBuilder

save(*args, **kwargs)

Persist the built model instance and return it.

Return type:

TransportChain

simulation(simulation)

Return type:

TransportChainBuilder

status(status)

Return type:

TransportChainBuilder

transport_type(transport_type)

Return type:

TransportChainBuilder

class ferrosoft.apps.emiflow.services.builder.TransportChainElementBuilder(element)

Bases: ModelBuilder

Fluent builder for one TransportChainElement (with optional geo-points).

activity_distance(distance)

Return type:

TransportChainElementBuilder

activity_distance_type(dtype)

Return type:

TransportChainElementBuilder

add_bulk(bulk)

Register the built instance with bulk for deferred persistence.

Return type:

TransportChainElement

add_geo_destination()

Attach a new destination geo-point and return its builder for chaining.

Return type:

GeoPointBuilder

add_geo_origin()

Attach a new origin geo-point and return its builder for chaining.

Return type:

GeoPointBuilder

chain(chain)

Return type:

TransportChainElementBuilder

distance_adjustment_factor(daf)

Return type:

TransportChainElementBuilder

epitype(epitype)

Return type:

TransportChainElementBuilder

classmethod new()

Start a fresh builder around an empty TransportChainElement.

Return type:

TransportChainElementBuilder

operation(operation)

Return type:

TransportChainElementBuilder

position(position)

Return type:

TransportChainElementBuilder

save(*args, **kwargs)

Persist the built model instance and return it.

Return type:

TransportChainElement

ferrosoft.apps.emiflow.services.builder.resolve_model(needle, model_type)

Resolve needle to an instance of model_type.

Parameters:

• needle – str to look up by name, UUID to look up by primary key, an instance of model_type (returned unchanged), or None (returned as None).

• model_type – Django model class.

Returns:

The resolved model instance, or None when needle is None.

Raises:

• ValueError – When needle is none of the accepted types.

• model_type.DoesNotExist – When the lookup finds no row.

ferrosoft.apps.emiflow.services.emissions module

Transport-chain emission calculation and booking.

This module is the heart of the transport-side emiflow pipeline:

• TransportChainCalculator walks a chain’s elements, looks up each operation’s emission intensity, and produces per-element and per-chain results (transport activity, distance, emissions for the operation and energy provision categories).

• EmissionIntensityCalculator derives a new emission intensity from raw inputs (factor, fuel quantity, mass, distance), performing unit normalisation so callers can supply values in any supported unit.

• TransportBookingService posts a chain: it geocodes points, computes emissions, denormalises everything into EmissionBookingRecord rows (one per chain + one per TransportLCCValue), and flips the chain status to POSTED.

ISO 14083 terminology is used throughout: TCE = Transport Chain Element, transport activity = mass × distance (tkm), intensity = emission / activity.

class ferrosoft.apps.emiflow.services.emissions.EmissionIntensityCalculator

Bases: object

Compute emission intensities from input variables.

Either pass an existing EmissionIntensity row whose input fields are populated (the calculator pulls the energy carrier’s density from the related factor), or call calculate() directly with an EmissionIntensityInputs value.

calculate(inputs)

Calculate one emission intensity from the supplied inputs.

Steps: normalise distance (treat zero as a neutral multiplier for hub operations), normalise the factor and verify its divider matches the quantity unit, normalise the quantity (volume → mass via density when needed), compute activity distance (mass × distance), and finally emission / activity.

Raises:

IntensityCalculationError – When factor and quantity units are incompatible, or when activity distance resolves to zero (which would divide by zero).

Return type:

DecimalWithUnit

calculateFor(model)

Return the operation and energy-provision intensities for model.

Returns a two-element list in (energy-provision, operation) order — mirroring the order of factor fields on the model.

Return type:

List[DecimalWithUnit]

class ferrosoft.apps.emiflow.services.emissions.EmissionIntensityInputs(emission_factor, quantity, mass, distance, energy_carrier_density)

Bases: object

Raw inputs to EmissionIntensityCalculator.calculate().

distance: DecimalWithUnit

Distance in km or m unit.

Distance may be None if calculating emission intensity for HUB operations.

emission_factor: DecimalWithUnit

Factor with gCO2e/MJ or kgCO2e/kg unit.

The divider of the unit has to align with the unit of the quantity.

energy_carrier_density: Decimal | None

Density of energy carrier (L/kg). Only required if quantity unit is L.

mass: DecimalWithUnit

Material mass with kg or t unit.

quantity: DecimalWithUnit

Quantity of energy carrier (fuel or electricity) with kg, L or kWh unit.

Kilogram and Liter is converted to Metric Ton. Kilowatt hours is converted to Megajoule.

class ferrosoft.apps.emiflow.services.emissions.EmissionResult(emission_category, operation_epitype, emission)

Bases: object

One emission value tagged by category and operation epitype.

emission: DecimalWithUnit

emission_category: EmissionCategory

operation_epitype: OperationEpitype

class ferrosoft.apps.emiflow.services.emissions.EmissionResultMixin

Bases: object

Memoised summation helpers for classes that own a list of EmissionResult.

emission(emission_category, operation_epitype)

Return the single emission matching both category and epitype, or None.

Return type:

DecimalWithUnit | None

emission_of_category(emission_category)

Sum emissions in emission_category (zero when none match).

Return type:

DecimalWithUnit

property emission_operation

Sum of emissions in the OPERATION category.

property emission_total: DecimalWithUnit

Sum of all emissions; 0 gCO2e when the list is empty.

exception ferrosoft.apps.emiflow.services.emissions.EmissionServiceError

Bases: RuntimeError

Base class for emiflow emission-service errors.

exception ferrosoft.apps.emiflow.services.emissions.IntensityCalculationError

Bases: EmissionServiceError

Raised when emission intensity cannot be computed (bad units, zero activity).

class ferrosoft.apps.emiflow.services.emissions.TransportActivityGroups(groups=<factory>)

Bases: object

Helper container that buckets transport activity by DistanceType.

add(dtype, activity_distance)

Accumulate activity_distance into the bucket for dtype.

groups: dict[DistanceType, DecimalWithUnit]

class ferrosoft.apps.emiflow.services.emissions.TransportBookingService

Bases: object

Post a transport chain’s calculated emissions to the booking journal.

Refuses to double-post (already-POSTED chains are skipped to keep the subscription tonnage counter accurate), enforces that each vehicle type within the chain uses a single distance strategy, geocodes geo-points, runs the TransportChainCalculator, and writes one EmissionBookingRecord per chain plus one per TransportLCCValue. Finally moves the chain to POSTED.

book(transport)

Calculate and persist booking records, then mark the chain POSTED.

Returns the upserted records, or an empty list when the chain is not in OPEN status.

Raises:

BookingError – When chain elements within one vehicle type mix distance strategies (an ISO 14083 reporting violation).

Return type:

list[EmissionBookingRecord]

class ferrosoft.apps.emiflow.services.emissions.TransportChainCalculator

Bases: object

Compute per-element and per-chain emissions for one or more transport chains.

The calculator carries running totals across multiple chains so it can be reused via add_transport_chain() for set-level reporting; for a single-chain calculation use calculate_for() on a freshly constructed calculator.

add_transport_chain(chain)

Calculate chain and fold its result into set_result.

calculate_for(chain)

Compute the TransportChainResult for chain.

For each chain element the operation’s emission intensity is looked up and combined with the transport activity (mass × distance for transport TCEs, mass alone for hub TCEs) and the distance-adjustment factor to yield emissions in the operation and energy provision categories.

Elements whose operation lacks an intensity contribute an empty result so the chain can still be reported on, instead of failing the whole calculation.

Raises:

RuntimeError – When the chain’s summed freight weight has no measurement unit attached.

Return type:

TransportChainResult

tc_results: List[TransportChainResult]

tce_results: List[TransportChainElementResult]

class ferrosoft.apps.emiflow.services.emissions.TransportChainElementResult(distance_type, element, transport_activity, emissions=<factory>)

Bases: EmissionResultMixin

Calculation result for one TransportChainElement.

distance_type: DistanceType | None

element: TransportChainElement

emissions: List[EmissionResult]

transport_activity: DecimalWithUnit

class ferrosoft.apps.emiflow.services.emissions.TransportChainResult(chain, emissions=<factory>, tce_results=<factory>, transport_activity=<factory>, distance=<factory>, weight=<factory>)

Bases: EmissionResultMixin

Aggregated calculation result for one TransportChain.

Mirrors the ISO 14083 reporting layout: emissions are accumulated across all elements, distance is summed, and transport activity is kept grouped by DistanceType so it can be reported per strategy.

add_activity_distance(dtype, activity_distance)

Accumulate activity_distance into the bucket for dtype.

chain: TransportChain

distance: DecimalWithUnit

emission_intensity(category)

Return the chain’s emission intensity for category (emission / activity).

Returns a zero DecimalWithUnit when there is no transport activity (a chain made entirely of hub elements).

Return type:

DecimalWithUnit

property emission_intensity_operation

Operation-only emission intensity.

property emission_intensity_total

Sum of operation and energy-provision intensities.

emissions: List[EmissionResult]

tce_results: List[TransportChainElementResult]

transport_activity: dict[DistanceType, DecimalWithUnit]

transport_activity_sum()

Total transport activity across every distance-type bucket.

Return type:

DecimalWithUnit

weight: DecimalWithUnit

class ferrosoft.apps.emiflow.services.emissions.TransportChainSetResult(emission_total, emission_operation, transport_activity, emission_intensity_total, emission_intensity_operation, distance, weight)

Bases: object

Aggregated result over a set of transport chains (cross-chain reporting).

Supports set_result += chain_result so callers can fold a chain result into the running totals.

distance: DecimalWithUnit

emission_intensity_operation: DecimalWithUnit

emission_intensity_total: DecimalWithUnit

emission_operation: DecimalWithUnit

emission_total: DecimalWithUnit

transport_activity: dict[DistanceType, DecimalWithUnit]

weight: DecimalWithUnit

class ferrosoft.apps.emiflow.services.emissions.TransportChainVisitorImportCallback(visitor)

Bases: ImportCallback

Import callback that runs a TransportChainVisitor after import finishes.

Collects every TransportChain instance the import emits; once on_finish fires (all rows committed), each chain accepts the wrapped visitor. Used to defer geocoding/routing until imported rows are queryable.

on_finish()

Walk every buffered chain with the wrapped visitor.

on_object(instance)

Buffer the instance when it’s a TransportChain.

ferrosoft.apps.emiflow.services.geographic module

Strategy that resolves activity distance from a TCE’s distance type.

For actual distance the user-entered value is treated as authoritative. For great-circle (GCD) and shortest-feasible (SFD) distances the configured router computes the distance between origin and destination geo-points. Results are normalised to kilometres.

class ferrosoft.apps.emiflow.services.geographic.DistanceStrategyCalculator(gcd_router, sfd_router)

Bases: object

Resolve a TCE’s activity distance according to its DistanceType.

distance(activity, force_calculation=False)

Return the activity distance for activity.

Parameters:

• activity (ActivityDistanceParameters) – Bundles distance type, the already-entered distance, the DAF, and origin/destination geo-points.

• force_calculation – When True, ignore any pre-existing activity.distance value and re-route from origin/destination.

Returns:

The resolved distance, or None for the actual distance strategy (the caller should keep the user-entered value).

Return type:

DecimalWithUnit | None

Raises:

GeographicError – When the inputs are inconsistent with the strategy (zero actual distance, missing origin or destination, unrouted route, or unknown distance type).

exception ferrosoft.apps.emiflow.services.geographic.GeographicError

Bases: Exception

Raised when distance resolution cannot satisfy the requested strategy.

ferrosoft.apps.emiflow.services.pcf module

Product Carbon Footprint (PCF) calculation across all chains for a material.

The ProductCarbonFootprintService aggregates transport emissions for every TransportChain that carries a given Material, normalises units across chains, and reports the emission-per-tonne intensity in kgCO2e.

class ferrosoft.apps.emiflow.services.pcf.LifecycleResult(label, value)

Bases: object

One row of a PCF table: a labelled lifecycle contribution.

label: str

value: DecimalWithUnit

class ferrosoft.apps.emiflow.services.pcf.ProductCarbonFootprintService

Bases: object

Calculate the Product Carbon Footprint of one material across all transports.

Uses a raw SQL query to find the transport chains carrying the material, then runs the chain calculator against each one and divides the summed emissions by the summed weight to express the result as kgCO2e per weight unit.

calculate_for(material)

Compute the PCF for material and return one transport row.

Returns an empty list when no chains carry the material. Raises RuntimeError when the source chains mix weight or emission units, since combining mismatched units would silently falsify the result.

Return type:

List[LifecycleResult]

ferrosoft.apps.emiflow.services.reporting module

Document generators that turn booking records into renderable reports.

Each DocumentGenerator subclass binds a Report (and its filtered queryset of booking records) to a list of DocumentOutput objects. The base generate() renders each output’s template into HTML, and prepare_report_models() persists the rendered HTML as short-lived StoredFile-backed reports that can later be turned into PDF by the upstream reporting service.

DocumentGeneratorRegistry maps ReportType values to the matching generator class.

class ferrosoft.apps.emiflow.services.reporting.DocumentGenerator(report, booking_records=[], template=None)

Bases: ABC

Generate HTML document for Emiflow reports.

booking_records: QuerySet

Reports center around booking records.

They can be used in get_context_data to make queries.

generate(template=None)

Generate document outputs.

Outputs are retrieved via get_document_outputs, which must be implemented in child classes. For each output, the template defined in the output is rendered and the resulting HTML is stored in output content field.

Parameters:

template (ReportTemplate | None) – Content customization parameters overriding default parameters.

Return type:

list[DocumentOutput]

Returns: List of outputs to be saved as StoredFile associated with a Report model.

get_context_data()

Get context for report template rendering.

This provides the base context for all document outputs. DocumentOutput itself also provides a get_contex_data method, which adds to the base context.

Return type:

dict

abstractmethod get_document_outputs()

Return which outputs to create.

A generator can return one or more outputs. Each non-empty output is saved into a StoredFile associated with a Report model. The user can select which output to view or to export to PDF.

See also generate method on how the outputs are used.

Return type:

list[DocumentOutput]

prepare_report_models(outputs, owner)

Create Report models (of the reporting app) with the content of given outputs.

Return type:

list[Report]

report: Report

The report for which the content is to be generated.

template: ReportTemplate | None

Content customization parameters

class ferrosoft.apps.emiflow.services.reporting.DocumentGeneratorRegistry

Bases: object

Maps ReportType to the matching DocumentGenerator class.

generator_classes = {ReportType.PCF: <class 'ferrosoft.apps.emiflow.services.reporting.PCFDocumentGenerator'>, ReportType.TRANSPORT: <class 'ferrosoft.apps.emiflow.services.reporting.TransportEmissionsDocumentGenerator'>, ReportType.WAREHOUSE: <class 'ferrosoft.apps.emiflow.services.reporting.WarehouseOverviewGenerator'>}

classmethod get_generator(report_type, **kwargs)

Instantiate the generator registered for report_type.

kwargs are forwarded to the generator class constructor (typically report, booking_records, template).

Return type:

DocumentGenerator

class ferrosoft.apps.emiflow.services.reporting.DocumentOutput(template_name, title='Report', content='')

Bases: object

Document outputs are produced by report generators.

content: str

HTML content (set by the generator)

get_context_data(**kwargs)

Get context data in addition to report generator context data.

Return type:

dict

template_name: str

Template from which to generate the document

title: str

Document title

class ferrosoft.apps.emiflow.services.reporting.PCFDocumentGenerator(report, booking_records=[], template=None)

Bases: DocumentGenerator

Generator that emits one PCF document per distinct material.

get_document_outputs()

Return one PCFDocumentOutput per material in the booking records.

Return type:

list[DocumentOutput]

class ferrosoft.apps.emiflow.services.reporting.PCFDocumentOutput(template_name, title='Report', content='', material_name='')

Bases: DocumentOutput

Output for one PCF report — one document per material.

get_context_data(**kwargs)

Aggregate booking records for this material and build the table context.

Partner columns are hidden from the rendered table because PCF reports are typically forwarded to third parties.

material_name: str

class ferrosoft.apps.emiflow.services.reporting.TransportEmissionsDocumentGenerator(report, booking_records=[], template=None)

Bases: DocumentGenerator

Generator for ISO 14083 transport-emissions reports (summary + listing).

get_context_data()

Aggregate transport bookings (distance, intensity, totals) into the template context.

Restricts the records to the three transport-related lifecycle categories (incoming, outgoing, drop shipment) and computes distance- and intensity-based aggregates from the denormalised extra JSON fields. Unit symbols are hard-coded because booking records always store transport distances and weights in those units.

Return type:

dict

get_document_outputs()

Return the two standard outputs: summary and full listing.

Return type:

list[DocumentOutput]

class ferrosoft.apps.emiflow.services.reporting.WarehouseOverviewGenerator(report, booking_records=[], template=None)

Bases: DocumentGenerator

Generator for the warehouse overview report (one table grouped by material).

get_context_data()

Provide the warehouse overview table, sourced from the PCF aggregate query.

Return type:

dict

get_document_outputs()

Return a single output containing the warehouse overview table.

Return type:

list[DocumentOutput]

ferrosoft.apps.emiflow.services.reporting.format_html(template, context=None, title='', timestamp=None, params=None)

Format report template as HTML.

Parameters:

• template (str | Template) – Django template. It should extend reporting/report.html.

• context (dict | None) – Context passed to template.

• title (str) – Default document title. It is overridden by ReportTemplate.document_title if not empty.

• timestamp (datetime | None) – Timestamp to show in document. If None, current time is used.

• params (ReportTemplate | None) – Override visual appearance and document title.

Return type:

str

Returns:

HTML content

ferrosoft.apps.emiflow.services.reporting.generate_report_document(report)

Read the report’s stored HTML and convert it to PDF bytes.

Return type:

bytes

ferrosoft.apps.emiflow.services.session module

Per-user-session storage of the current transport chain.

When the user is mid-way through editing a transport chain across multiple views, the chain’s primary key is stashed in the Django session so subsequent views can resume without it being passed through every URL. All helpers degrade silently when the request has no session (e.g. during management commands or unit tests).

ferrosoft.apps.emiflow.services.session.clear_session(request)

Forget the current chain. No-op when nothing is stored.

ferrosoft.apps.emiflow.services.session.get_current_transport(request)

Return the primary key of the current chain, or None when unset.

Return type:

UUID | None

ferrosoft.apps.emiflow.services.session.set_current_transport(request, transport)

Remember transport as the user’s currently-edited chain.

ferrosoft.apps.emiflow.services.tonnage module

Tonnage accounting for subscription billing.

Customers are billed on the cumulative weight of (non-simulation) emission bookings. This module exposes two query paths:

• A direct database aggregate per tenant (TonnageService.get_summary()) used at startup to seed the Prometheus counter and on demand for reconciliation. Expensive — it opens a connection to every active tenant database.

• A cheap VictoriaMetrics query (TonnageServiceVictoria.get_org_stats()) that returns the rolling 4-week tonnage and configured limit per organisation from the metrics store.

class ferrosoft.apps.emiflow.services.tonnage.TonnageService

Bases: ABC

Abstract tonnage service. The concrete singleton is TonnageServiceVictoria.

classmethod get_instance()

Return the configured production tonnage service.

Return type:

TonnageService

abstractmethod get_org_stats(org)

Return the billable tonnage and limit for org (implementation-specific).

Return type:

TonnageStats

get_summary()

Return per-tenant booked tonnage across every active tenant.

Iterates over every organisation and active tenant, opening each tenant database to read its booking-record aggregate. Misconfigured databases (ImproperlyConfigured) yield a zero-tonnage row so the summary remains complete. Costly; prefer get_org_stats() whenever per-organisation totals are sufficient.

Return type:

list[TonnageSummary]

static get_tenant_tonnage(tenant)

Return tenant’s booked tonnage from its own database.

Simulation records are excluded so dry-run bookings don’t count toward the customer’s subscription quota.

class ferrosoft.apps.emiflow.services.tonnage.TonnageServiceVictoria(config)

Bases: TonnageService

Tonnage service backed by VictoriaMetrics Prometheus-compatible queries.

Uses a single PromQL union over the emiflow_tonnage_limit gauge and the 4-week increase of the emiflow_tonnage_total counter to return both values in one HTTP call. Returns zero-stats on query failure so callers can render a safe fallback.

class Config(base_url, timeout=10)

Bases: object

VictoriaMetrics endpoint configuration.

base_url: str

classmethod from_django_settings()

Build a config from the VICTORIA_BASE_URL Django setting.

timeout: float = 10

classmethod get_instance()

Instantiate the service with config loaded from Django settings.

get_org_stats(org)

Return org’s current tonnage and limit from VictoriaMetrics.

Falls back to a zero stats tuple (and logs the response) when the query fails, returns a non-success status, or yields an unexpected result type.

Return type:

TonnageStats

class ferrosoft.apps.emiflow.services.tonnage.TonnageStats(billable_tonnage, tonnage_limit)

Bases: object

Billing snapshot for an organisation: current usage versus configured limit.

billable_tonnage: DecimalWithUnit

tonnage_limit: DecimalWithUnit

class ferrosoft.apps.emiflow.services.tonnage.TonnageSummary(organization, tenant, tonnage)

Bases: object

Aggregated booked tonnage for one (organization, tenant) pair.

organization: Organization

tenant: Tenant

tonnage: Decimal

ferrosoft.apps.emiflow.services.treatment module

Emission calculation and booking for treatments.

TreatmentEmissionCalculator multiplies the treatment’s output mass by the operation’s emission intensity (for both operation and energy provision categories) to derive the total emission. TreatmentBookingService then splits that total across output lines using each line’s allocated_fraction and creates one EmissionBookingRecord per output and per attached TreatmentLCCValue.

class ferrosoft.apps.emiflow.services.treatment.EmissionResult(emission, intensity)

Bases: object

One category’s calculated emission paired with the intensity that produced it.

emission: DecimalWithUnit

intensity: DecimalWithUnit

class ferrosoft.apps.emiflow.services.treatment.TreatmentBookingService

Bases: object

Persist EmissionBookingRecord rows for a treatment.

For every output line, the treatment’s total emission is multiplied by the line’s allocated_fraction to produce one booking record; each attached TreatmentLCCValue is similarly split across output lines.

book(treatment)

Calculate emissions for treatment and bulk-upsert booking records.

Returns the list of persisted records (one per output line plus one per LCC value per output line).

Return type:

list[EmissionBookingRecord]

class ferrosoft.apps.emiflow.services.treatment.TreatmentEmissionCalculator

Bases: object

Compute operation and energy-provision emissions for a treatment.

calculate_for(treatment)

Calculate emissions for treatment.

Raises:

NormalizationError – When the operation has no emission intensity, or the intensity unit cannot be normalised to kgCO2e/t.

Return type:

TreatmentResult

class ferrosoft.apps.emiflow.services.treatment.TreatmentResult(activity, emissions)

Bases: object

Per-treatment activity (output mass) and per-category emission results.

activity: DecimalWithUnit

emissions: Mapping[EmissionCategory, EmissionResult]

ferrosoft.apps.emiflow.services.update module

Emiflow hooks into the ferrobase update pipeline.

The ferrobase ferrobase_update management command iterates over registered global and per-tenant updaters at deploy time. This module registers an updater that ensures the four predefined LifecycleCategory rows exist in every database — both the shared catalog database (global pass) and each tenant database (tenant pass).

ferrosoft.apps.emiflow.services.update.create_lifecycle_categories(writer, tenant=None)

Upsert the predefined lifecycle categories in the appropriate database.

Parameters:

• writer (StatusWriter) – Status writer used to report progress back to the update pipeline.

• tenant (Tenant | None) – When provided, the upsert is routed to that tenant’s database; when None, the default (catalog) database is used so the same function serves both the global and the per-tenant pass.

ferrosoft.apps.emiflow.services.update.init()

Register create_lifecycle_categories() with the global and tenant updater registries.

Called from EmiflowConfig.ready() so the updaters are wired up at Django app initialisation.

ferrosoft.apps.emiflow.services.vehicle module

Heuristic mapping from freight weight to a likely VehicleType.

Used during bulk import when the source data does not specify the vehicle type explicitly. The thresholds are hard-coded approximations and will likely become tenant-configurable.

ferrosoft.apps.emiflow.services.vehicle.detect_vehicle_type(weight)

Guess the vehicle type that would typically carry weight.

Returns ROAD for less than 30 t, RAIL for 30..60 t, and INLAND_SHIPPING above. Kilogram input is converted to metric tons before comparison.

Return type:

VehicleType

ferrosoft.apps.emiflow.services.visitor module

Visitor implementations driven by TransportChain.accept().

Visitors here perform actions over an entire chain’s structure (elements, freight items, geo-points). DebuggerVisitor dumps every field to stdout for development; GeocodingRoutingVisitor populates missing coordinates and activity distances before emission calculation.

class ferrosoft.apps.emiflow.services.visitor.DebuggerVisitor

Bases: TransportChainVisitor

Visitor that prints every field of every visited model to stdout.

Intended for ad-hoc debugging only — production code should not depend on it.

before_chain(chain)

Called once before any freight items or elements are visited.

on_element(element)

Called for each chain element, in position order.

on_freight_item(item)

Called for each freight item, in material-code order.

exception ferrosoft.apps.emiflow.services.visitor.GeocodingRoutingError

Bases: RuntimeError

Raised when geocoding or routing fails for a chain element.

class ferrosoft.apps.emiflow.services.visitor.GeocodingRoutingVisitor(geocoder, distance_calc, force_geocoding=False, force_distance_calculation=False)

Bases: TransportChainVisitor

Fill in missing coordinates and activity distances on each TCE.

For every transport (non-hub) element with origin or destination geo-points whose coordinates are still (0, 0), the address is geocoded and the coordinates are persisted. Then, when the activity distance has not been entered (or force_distance_calculation is set), the configured DistanceStrategyCalculator is used to compute and store it. Run before booking to make sure emissions can actually be calculated.

classmethod get_instance(**kwargs)

Build a visitor wired up with the default address-router-backed services.

on_element(element)

Geocode geo-points and resolve activity distance for element.

Hub elements are skipped — distance is irrelevant there. For transport elements an origin and destination are required for any distance type other than actual distance. Geocoding and routing failures are re-raised as GeocodingRoutingError with a user-facing message.