refactor(shock): derive shock-paraeter registry from PDL (#14)

src/provider_simenv/agents/farmer.py

@@ -168,12 +168,12 @@ def _step_bra(self):
         raises per-unit cost, which raises unit_price automatically.
         """
         env = self.model.environment
-        farm_capacity = env.get_effective_value("farm_capacity_bra")
+        farm_capacity = env.get_effective_value("brazil_farms", "supply")
         self.quantity_available = self.base_yield * farm_capacity
 
         if self.quantity_available > 0:
             # fertilizer price factor raises effective fixed costs this step
-            fertilizer_factor = env.get_effective_value("fertilizer_price_factor")
+            fertilizer_factor = env.get_effective_value("fertilizer_supply", "price")
             effective_costs = self.fixed_costs * fertilizer_factor
             self.unit_price = (effective_costs / self.quantity_available) * (1.0 + self.margin)
         else:
@@ -220,7 +220,7 @@ def _step_arg(self):
         farm_capacity_arg allows ARG-specific shocks to be modelled independently.
         Defaults to 1.0 = always unshocked.
         """
-        farm_capacity = self.model.environment.get_effective_value("farm_capacity_arg")
+        farm_capacity = self.model.environment.get_effective_value("argentina_farms", "supply")
         self.quantity_available = self.base_yield * farm_capacity
 
         if self.quantity_available > 0:

src/provider_simenv/agents/process.py

@@ -71,12 +71,12 @@ def step(self):
     # Shared helper: receive from upstream list, convert, compute price
     # ------------------------------------------------------------------
 
-    def _process(self, upstream_list, peer_list, scenario_param: str = ""):
+    def _process(self, upstream_list, peer_list, shock_key: tuple[str, str] | None = None):
         """
         Pull an equal share of upstream output, apply conversion_ratio,
         and compute unit_price accounting for yield loss.
 
-        scenario_param: scenario param whose effective value scales output.
+        shock_key: PDL (entity, field) whose effective value scales output.
                     Models indirect capacity reduction from soja shortage.
 
         For every 1 unit of output, (1 / conversion_ratio) input units
@@ -95,7 +95,7 @@ def _process(self, upstream_list, peer_list, scenario_param: str = ""):
             self.unit_price = 0.0
             return
 
-        effective_factor = self.model.environment.get_effective_value(scenario_param) if scenario_param else 1.0
+        effective_factor = self.model.environment.get_effective_value(*shock_key) if shock_key else 1.0
 
         total_input = sum(a.quantity_available for a in active_upstream)
 
@@ -135,13 +135,13 @@ def _step_processor(self):
         self._process(
             upstream_list=combined_eu,
             peer_list=self.model.processors,
-            scenario_param="oil_mill_capacity",
+            shock_key=("eu_oil_mills", "supply"),
         )
 
     def _step_feed_manufacturer(self):
         """Receive soja meal from processors, compound to animal feed."""
         self._process(
             upstream_list=self.model.processors,
             peer_list=self.model.feed_manufacturers,
-            scenario_param="feed_mill_capacity",
+            shock_key=("feed_mills", "supply"),
         )

src/provider_simenv/agents/transport.py

@@ -109,12 +109,12 @@ def step(self):
     # cap at capacity, compute all-in unit_price (commodity + freight)
     # ------------------------------------------------------------------
 
-    def _move(self, upstream, scenario_param: str = ""):
+    def _move(self, upstream, shock_key: tuple[str, str] | None = None):
         """
         Pull an equal share of upstream output, ca at own capacity,
         and compute the all-in price passed to the next chain node.
 
-        scenario_param: scenario param whose effective value scales this agent's capacity, via env.get_effective_value()
+        shock_key: PDL (entity, field) whoe effective value scales this agent's capacity
         """
         margin = self.scenario.margin_transport
 
@@ -135,7 +135,7 @@ def _move(self, upstream, scenario_param: str = ""):
         volume_in = total_volume / n_self
 
         env = self.model.environment
-        effective_factor = env.get_effective_value(scenario_param) if scenario_param else 1.0
+        effective_factor = env.get_effective_value(*shock_key) if shock_key else 1.0
 
         # effective capacity after applying port capacity shock
         effective_capacity = self.capacity * effective_factor
@@ -151,21 +151,21 @@ def _move(self, upstream, scenario_param: str = ""):
         # price = commodity price + freight fee per unit
         # energy price factor raises transport operation costs
         if self.quantity_available > 0:
-            energy_factor = env.get_effective_value("energy_price_factor")
+            energy_factor = env.get_effective_value("gas_supply", "price")
             effective_costs = self.fixed_costs * energy_factor
             freight_fee = (effective_costs / self.quantity_available) * (1.0 + margin)
             self.unit_price = upstream_price + freight_fee
         else:
             self.unit_price = 0.0
 
 
-    def _move_split(self, upstream_list, share: float, scenario_param: str = "", exclude_arg=False, exclude_usa=False):
+    def _move_split(self, upstream_list, share: float, shock_key: tuple[str, str] | None = None, exclude_arg=False, exclude_usa=False):
         """
         Like _move, but routes only share fraction of total upstream volume through this port.
         Used to split wholesaler output between Santos and Paranagua.
 
         :param share: fraction of total wholesaler output for this port (e.g. 0.7 for Santos, 0.3 for Paranagua).
-        :param scenario_param: scenario param whose effective value scales this agent's capacity.
+        :param shock_key: PDL (entity, field) whoe effective value scales this agent's capacity
         """
         margin = self.scenario.margin_transport
         active_upstream = upstream_list.filter(lambda a: a.active)
@@ -192,7 +192,7 @@ def _move_split(self, upstream_list, share: float, scenario_param: str = "", exc
         volume_in = (routable_volume * share) / n_self
 
         env = self.model.environment
-        effective_factor = env.get_effective_value(scenario_param) if scenario_param else 1.0
+        effective_factor = env.get_effective_value(*shock_key) if shock_key else 1.0
         effective_capacity = self.capacity * effective_factor
 
         self.quantity_available = min(volume_in, effective_capacity)
@@ -204,7 +204,7 @@ def _move_split(self, upstream_list, share: float, scenario_param: str = "", exc
         upstream_price = (total_value / total_volume) if total_volume > 0 else 0.0
 
         if self.quantity_available > 0:
-            energy_factor = env.get_effective_value("energy_price_factor")
+            energy_factor = env.get_effective_value("gas_supply", "price")
             effective_costs = self.fixed_costs * energy_factor
             freight_fee = (effective_costs / self.quantity_available) * (1.0 + margin)
             self.unit_price = upstream_price + freight_fee
@@ -244,7 +244,7 @@ def _step_sa_santos(self):
         self._move_split(
             self.model.wholesalers,
             share=self.scenario.santos_share,
-            scenario_param="port_capacity_santos",
+            shock_key=("santos_port", "supply"),
             exclude_arg=True,
             exclude_usa=True,
         )
@@ -257,7 +257,7 @@ def _step_sa_paranagua(self):
         self._move_split(
             self.model.wholesalers,
             share=1.0 - self.scenario.santos_share,
-            scenario_param="port_capacity_paranagua",
+            shock_key=("paranagua_port", "supply"),
             exclude_arg=True,
             exclude_usa=True,
         )
@@ -315,7 +315,7 @@ def _step_sea_arg(self):
         upstream_price = total_value / total_arg
 
         if self.quantity_available > 0:
-            energy_factor = self.model.environment.get_effective_value("energy_price_factor")
+            energy_factor = self.model.environment.get_effective_value("gas_supply", "price")
             effective_costs = self.fixed_costs * energy_factor
             freight_fee = (effective_costs / self.quantity_available) * (1.0 + margin)
             self.unit_price = upstream_price + freight_fee
@@ -364,7 +364,7 @@ def _step_sea_usa(self):
         upstream_price = total_value / total_usa
 
         if self.quantity_available > 0:
-            energy_factor = self.model.environment.get_effective_value("energy_price_factor")
+            energy_factor = self.model.environment.get_effective_value("gas_supply", "price")
             effective_costs = self.fixed_costs * energy_factor
             freight_fee = (effective_costs / self.quantity_available) * (1.0 + margin)
             self.unit_price = upstream_price + freight_fee
@@ -382,7 +382,7 @@ def _step_eu_rtm(self):
             + self.model.sea_lane_arg.filter(lambda a: a.active)
             + self.model.sea_lane_usa.filter(lambda a: a.active)
         )
-        self._move(combined, scenario_param="port_capacity_rotterdam")
+        self._move(combined, shock_key=("rotterdam_port", "supply"))
 
 
     def _step_eu_ham(self):
@@ -393,6 +393,6 @@ def _step_eu_ham(self):
         """
         self._move(
             self.model.sea_lane_paranagua,
-            scenario_param="port_capacity_hamburg",
+            shock_key=("hamburg_port", "supply"),
         )
 

src/provider_simenv/data/input/SimulatorScenarios.csv

@@ -1,4 +1,3 @@
-id,run_num,period_num,n_bra_farmers,n_arg_farmers,n_usa_farmers,n_wholesalers,n_transport_sa_santos,n_transport_sa_paranagua,n_sea_lane_santos,n_sea_lane_paranagua,n_sea_lane_arg,n_sea_lane_usa,n_transport_eu_rtm,n_transport_eu_ham,n_processors,n_feed_manufacturers,n_feed_traders,n_eu_farmers,farm_capacity_bra,farm_capacity_arg,port_capacity_santos,port_capacity_paranagua,port_capacity_rotterdam,port_capacity_hamburg,santos_share,fertilizer_price_factor,energy_price_factor,oil_mill_capacity,feed_mill_capacity,shock_ramp_steps,shock_onset_farm_bra,shock_end_farm_bra,shock_onset_farm_arg,shock_end_farm_arg,shock_onset_port_santos,shock_end_port_santos,shock_onset_port_paranagua,shock_end_port_paranagua,shock_onset_port_rotterdam,shock_end_port_rotterdam,shock_onset_port_hamburg,shock_end_port_hamburg,shock_onset_fertilizer,shock_end_fertilizer,shock_onset_energy,shock_end_energy,shock_onset_oil_mill,shock_end_oil_mill,shock_onset_feed_mill,shock_end_feed_mill,farm_size_sigma_bra,farm_size_sigma_eu,farm_size_seed,wholesaler_storage_capacity
-0,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,1.0,1.0,1.0,1.0,1.0,1.0,0.7,1.0,1.0,1.0,1.0,0,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0.0,0.0,42,2857.0
-1,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.6,1.0,1.0,1.0,1.0,1.0,0.7,1.0,1.0,1.0,1.0,0,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0.4,0.4,42,2857.0
-2,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.6,1.0,0.5,1.0,1.0,1.0,0.7,1.0,1.0,1.0,1.0,0,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0.4,0.4,42,2857.0
+id,run_num,period_num,n_bra_farmers,n_arg_farmers,n_usa_farmers,n_wholesalers,n_transport_sa_santos,n_transport_sa_paranagua,n_sea_lane_santos,n_sea_lane_paranagua,n_sea_lane_arg,n_sea_lane_usa,n_transport_eu_rtm,n_transport_eu_ham,n_processors,n_feed_manufacturers,n_feed_traders,n_eu_farmers,santos_share,shock_ramp_steps,farm_size_sigma_bra,farm_size_sigma_eu,farm_size_seed,wholesaler_storage_capacity
+0,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.7,0,0.0,0.0,42,2857.0
+1,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.7,0,0.0,0.0,42,2857.0

src/provider_simenv/data/input/SimulatorScenarios_template.csv

@@ -1,4 +1,4 @@
-id,run_num,period_num,n_bra_farmers,n_arg_farmers,n_usa_farmers,n_wholesalers,n_transport_sa_santos,n_transport_sa_paranagua,n_sea_lane_santos,n_sea_lane_paranagua,n_sea_lane_arg,n_sea_lane_usa,n_transport_eu_rtm,n_transport_eu_ham,n_processors,n_feed_manufacturers,n_feed_traders,n_eu_farmers,farm_capacity_bra,farm_capacity_arg,port_capacity_santos,port_capacity_paranagua,port_capacity_rotterdam,port_capacity_hamburg,santos_share,fertilizer_price_factor,energy_price_factor,oil_mill_capacity,feed_mill_capacity,shock_ramp_steps,shock_onset_farm_bra,shock_end_farm_bra,shock_onset_farm_arg,shock_end_farm_arg,shock_onset_port_santos,shock_end_port_santos,shock_onset_port_paranagua,shock_end_port_paranagua,shock_onset_port_rotterdam,shock_end_port_rotterdam,shock_onset_port_hamburg,shock_end_port_hamburg,shock_onset_fertilizer,shock_end_fertilizer,shock_onset_energy,shock_end_energy,shock_onset_oil_mill,shock_end_oil_mill,shock_onset_feed_mill,shock_end_feed_mill,farm_size_sigma_bra,farm_size_sigma_eu,farm_size_seed,wholesaler_storage_capacity
-0,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,1.0,1.0,1.0,1.0,1.0,1.0,0.7,1.0,1.0,1.0,1.0,0,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0.0,0.0,42,2857.0
-1,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.6,1.0,1.0,1.0,1.0,1.0,0.7,1.0,1.0,1.0,1.0,0,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0.4,0.4,42,2857.0
-2,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.6,1.0,0.5,1.0,1.0,1.0,0.7,1.0,1.0,1.0,1.0,0,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0,365,0.4,0.4,42,2857.0
+id,run_num,period_num,n_bra_farmers,n_arg_farmers,n_usa_farmers,n_wholesalers,n_transport_sa_santos,n_transport_sa_paranagua,n_sea_lane_santos,n_sea_lane_paranagua,n_sea_lane_arg,n_sea_lane_usa,n_transport_eu_rtm,n_transport_eu_ham,n_processors,n_feed_manufacturers,n_feed_traders,n_eu_farmers,santos_share,shock_ramp_steps,farm_size_sigma_bra,farm_size_sigma_eu,farm_size_seed,wholesaler_storage_capacity
+0,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.7,0,0.0,0.0,42,2857.0
+1,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.7,0,0.4,0.4,42,2857.0
+2,1,365,10,5,8,3,1,1,1,1,1,1,1,1,3,3,3,10,0.7,0,0.4,0.4,42,2857.0

src/provider_simenv/environment.py

@@ -16,23 +16,9 @@
 from typing import TYPE_CHECKING
 from Melodie import Environment
 
-if TYPE_CHECKING:
-    from event_tracker import EventTracker
 
-
-# maps scenario param name -> (onset_field, end_field) on SupplyChainScenario
-_PARAM_TIMING_FIELDS: list[tuple[str, str, str]] = [
-    ("farm_capacity_bra", "shock_onset_farm_bra", "shock_end_farm_bra"),
-    ("farm_capacity_arg", "shock_onset_farm_arg", "shock_end_farm_arg"),
-    ("port_capacity_santos", "shock_onset_port_santos", "shock_end_port_santos"),
-    ("port_capacity_paranagua", "shock_onset_port_paranagua", "shock_end_port_paranagua"),
-    ("port_capacity_rotterdam", "shock_onset_port_rotterdam", "shock_end_port_rotterdam"),
-    ("port_capacity_hamburg", "shock_onset_port_hamburg", "shock_end_port_hamburg"),
-    ("fertilizer_price_factor", "shock_onset_fertilizer", "shock_end_fertilizer"),
-    ("energy_price_factor", "shock_onset_energy", "shock_end_energy"),
-    ("oil_mill_capacity", "shock_onset_oil_mill", "shock_end_oil_mill"),
-    ("feed_mill_capacity", "shock_onset_feed_mill", "shock_end_feed_mill"),
-]
+from .shock_registry import DROUGHT_KEY
+from .event_tracker import EventTracker
 
 
 class SupplyChainEnvironment(Environment):
@@ -47,7 +33,7 @@ class SupplyChainEnvironment(Environment):
     feed_price: float = 0.0
 
     # global shock intensity
-    # Agents should call get_shock_scale(param) instead of reading this directly.
+    # Agents should call get_shock_scale(entity, field) instead of reading this directly.
     shock_scale: float = 0.0
 
     # drought severity this step
@@ -78,10 +64,8 @@ def setup(self):
         self.transport_utilisation = 0.0
         self.current_step = 0
 
-        # per-parameter shock activation scale
-        self.shock_scales: dict[str, float] = {
-            param: 0.0 for param, _, _ in _PARAM_TIMING_FIELDS
-        }
+        # per (entity, field) shock activation scale.
+        self.shock_scales: dict[tuple[str, str], float] = {}
 
 
     def update_shock_scales(self, period: int):
@@ -94,43 +78,42 @@ def update_shock_scales(self, period: int):
         """
         if self._tracker is not None:
             self._tracker.step(period)
-            for param, _, _ in _PARAM_TIMING_FIELDS:
-                self.shock_scales[param] = self._tracker.get_shock_scale(param)
+            # seed the key once
+            if not self.shock_scales:
+                self.shock_scales = {key: 0.0 for key in self._tracker.known_keys()}
+            for key in self.shock_scales:
+                self.shock_scales[key] = self._tracker.get_shock_scale(*key)
         else:
-            for param, onset_field, end_field in _PARAM_TIMING_FIELDS:
-                onset = getattr(self.scenario, onset_field)
-                end = getattr(self.scenario, end_field)
-                value = getattr(self.scenario, param)
-                has_shock = value != 1.0
-                self.shock_scales[param] = (1.0 if has_shock and onset <= period < end else 0.0)
+            for key in self.shock_scales:
+                self.shock_scales[key] = 0.0
 
         self.shock_scale = max(self.shock_scales.values(), default=0.0)
 
-        # drought severity: use racker value if available
-        bra_scale = self.shock_scales.get("farm_capacity_bra", 0.0)
-        bra_value = self.get_effective_value("farm_capacity_bra")
-        self.drought_severity = (
-            bra_scale * (1.0 - bra_value)
-        )
+        # Drought severity is defined as brazil_farms supply degradation (DROUGHT_KEY)
+        bra_scale = self.shock_scales.get(DROUGHT_KEY, 0.0)
+        bra_value = self.get_effective_value(*DROUGHT_KEY)
+        self.drought_severity = (bra_scale * (1.0 - bra_value))
+
+
 
 
-    def get_shock_scale(self, param: str) -> float:
+    def get_shock_scale(self, entity: str, field: str) -> float:
         """
         Return the current shock actibation scale for a scenario parameter.
         """
-        return self.shock_scales.get(param, 0.0)
+        return self.shock_scales.get((entity, field), 0.0)
 
 
-    def get_effective_value(self, param: str) -> float:
+    def get_effective_value(self, entity: str, field: str) -> float:
         """
         Return the effective value for this step.
 
         Tracker mode: aggregated from currently active events only.
-        Static mode: reads fixed value from the scenario.
+        No tracker (baseline / non-PDL): unshocked, always 1.0
         """
         if self._tracker is not None:
-            return self._tracker.get_param_value(param)
-        return getattr(self.scenario, param, 1.0)
+            return self._tracker.get_param_value(entity, field)
+        return 1.0
 
 
     def step(self):