Compile a SupplyChainModel into a GDSSpec.
Registers: types, spaces, entities, blocks, wirings.
Source code in packages/gds-business/gds_business/supplychain/compile.py
| def compile_scn(model: SupplyChainModel) -> GDSSpec:
"""Compile a SupplyChainModel into a GDSSpec.
Registers: types, spaces, entities, blocks, wirings.
"""
spec = GDSSpec(name=model.name, description=model.description)
# 1. Register types
spec.collect(InventoryType, ShipmentRateType, DemandType)
# 2. Register spaces
spec.collect(InventorySpace, ShipmentRateSpace, DemandSpace)
# 3. Register entities (one per supply node)
for node in model.nodes:
spec.register_entity(_build_node_entity(node))
# 4. Register blocks
for d in model.demand_sources:
spec.register_block(_build_demand_block(d))
for p in model.order_policies:
spec.register_block(_build_policy_block(p, model))
for node in model.nodes:
spec.register_block(_build_node_mechanism(node, model))
# 5. Register spec wirings
all_block_names = [b.name for b in spec.blocks.values()]
wires: list[Wire] = []
# Demand -> Policy wires
for d in model.demand_sources:
for p in model.order_policies:
if p.node == d.target:
wires.append(
Wire(source=d.name, target=p.name, space="SCN DemandSpace")
)
# Policy -> Mechanism wires
for p in model.order_policies:
wires.append(
Wire(
source=p.name,
target=_mechanism_block_name(p.node),
space="SCN ShipmentRateSpace",
)
)
# Mechanism -> Policy temporal wires (inventory feedback)
for node in model.nodes:
for p in model.order_policies:
if node.name in p.inputs:
wires.append(
Wire(
source=_mechanism_block_name(node.name),
target=p.name,
space="SCN InventorySpace",
)
)
if wires:
spec.register_wiring(
SpecWiring(
name=f"{model.name} Wiring",
block_names=all_block_names,
wires=wires,
description=f"Auto-generated wiring for SCN {model.name!r}",
)
)
return spec
|