Cement plant and associated technologies

assume/common/forecaster.py

@@ -781,6 +781,132 @@ def initialize(
         initializing_unit.setup_model()
 
 
+class CementForecaster(DsmUnitForecaster):
+    """Forecaster for cement plant units.
+
+    Provides all DSM forecasts (see :class:`DsmUnitForecaster`) plus fuel prices, clinker
+    demand and the optional profiles that shape clinker production. After initialization,
+    DSM signals are copied to the unit and ``setup_model()`` is called.
+
+    Supports the same three operational strategies as :class:`SteelplantForecaster`:
+    1. **Profile-guided**: If ``normalized_load_profile`` is provided, production follows the profile shape.
+    2. **Min-demand**: If hourly minimum demand (``clinker_demand``) is provided, meets per-hour minimums.
+    3. **Cost-optimized**: If neither is provided, minimizes cost without shape constraints.
+
+    Attributes:
+        fuel_prices (dict[str, ForecastSeries]): Map of fuel type to forecasted fuel prices
+            (``natural_gas``, ``coal``, ``hydrogen``, ``co2``).
+        clinker_demand (ForecastSeries): Per-timestep clinker production demand in tonnes (optional).
+        normalized_load_profile (ForecastSeries): Normalized profile to guide production shape (optional).
+        thermal_storage_schedule (ForecastSeries): Binary charge/discharge schedule of a
+            long-term thermal storage (optional).
+        electricity_price_flex (ForecastSeries): Alternative price signal used by the
+            ``electricity_price_signal`` flexibility measure. Falls back to ``electricity_price``.
+        availability_profiles (dict[str, ForecastSeries]): Per-component availability
+            profiles (1 available, 0 unavailable), keyed by technology name.
+    """
+
+    def __init__(
+        self,
+        index: ForecastIndex,
+        fuel_prices: dict[str, ForecastSeries],
+        market_prices: dict[str, ForecastSeries] = None,
+        residual_load: dict[str, ForecastSeries] = None,
+        availability: ForecastSeries = 1,
+        forecast_algorithms: dict[str, str] = {},
+        forecast_registries: dict[str, dict] = None,
+        congestion_signal: ForecastSeries = 0.0,
+        renewable_utilisation_signal: ForecastSeries = 0.0,
+        electricity_price: ForecastSeries = None,
+        electricity_price_flex: ForecastSeries = None,
+        clinker_demand: ForecastSeries = None,
+        normalized_load_profile: ForecastSeries = None,
+        thermal_storage_schedule: ForecastSeries = 0,
+        availability_profiles: dict[str, ForecastSeries] = None,
+    ):
+        super().__init__(
+            index=index,
+            availability=availability,
+            forecast_algorithms=forecast_algorithms,
+            forecast_registries=forecast_registries,
+            market_prices=market_prices,
+            residual_load=residual_load,
+            congestion_signal=congestion_signal,
+            renewable_utilisation_signal=renewable_utilisation_signal,
+            electricity_price=electricity_price,
+        )
+        self.fuel_prices = self._dict_to_series(fuel_prices)
+        self.clinker_demand = (
+            self._to_series(clinker_demand) if clinker_demand is not None else None
+        )
+        self.normalized_load_profile = (
+            self._to_series(normalized_load_profile)
+            if normalized_load_profile is not None
+            else None
+        )
+        self.thermal_storage_schedule = self._to_series(thermal_storage_schedule)
+        self._electricity_price_flex = (
+            self._to_series(electricity_price_flex)
+            if electricity_price_flex is not None
+            else None
+        )
+        self.availability_profiles = {
+            tech: self._to_series(profile)
+            for tech, profile in (availability_profiles or {}).items()
+            if profile is not None
+        }
+
+    @property
+    def electricity_price_flex(self) -> FastSeries:
+        """Price signal for the ``electricity_price_signal`` measure.
+
+        Falls back to the current ``electricity_price`` when no dedicated flex price
+        was provided, so the fallback also tracks the price initialized later on.
+        """
+        if self._electricity_price_flex is None:
+            return self.electricity_price
+        return self._electricity_price_flex
+
+    @electricity_price_flex.setter
+    def electricity_price_flex(self, value: ForecastSeries) -> None:
+        self._electricity_price_flex = (
+            self._to_series(value) if value is not None else None
+        )
+
+    def get_price(self, fuel: str) -> FastSeries:
+        if fuel not in self.fuel_prices:
+            return self._to_series(0)
+        return self.fuel_prices[fuel]
+
+    def initialize(
+        self,
+        units: list[BaseUnit],
+        market_configs: list[MarketConfig],
+        forecast_df: ForecastSeries = None,
+        initializing_unit: BaseUnit = None,
+    ):
+        super().initialize(
+            units,
+            market_configs,
+            forecast_df,
+            initializing_unit,
+        )
+
+        # Always set standard DSM signals
+        initializing_unit.congestion_signal = self.congestion_signal
+        initializing_unit.renewable_utilisation_signal = (
+            self.renewable_utilisation_signal
+        )
+
+        if self.clinker_demand is not None:
+            initializing_unit.clinker_demand_per_timestep = self.clinker_demand
+
+        if self.normalized_load_profile is not None:
+            initializing_unit.normalized_load_profile = self.normalized_load_profile
+
+        initializing_unit.setup_model()
+
+
 class SteamgenerationForecaster(DsmUnitForecaster):
     """Forecaster for steam generation units.
 

assume/scenario/loader_csv.py

@@ -20,6 +20,7 @@
 from assume.common.fast_pandas import FastIndex
 from assume.common.forecaster import (
     BuildingForecaster,
+    CementForecaster,
     CustomUnitForecaster,
     DemandForecaster,
     DsmUnitForecaster,
@@ -814,6 +815,42 @@ def get_building_profile(column_name: str) -> pd.Series:
                         normalized_load_profile=normalized_profile,
                         steel_demand=steel_demand,
                     )
+                if type == "cement_plant":
+                    storage_schedule = get_unit_forecast_column(
+                        forecasts_df, id, "thermal_storage_schedule"
+                    )
+                    unit_forecasts[id] = CementForecaster(
+                        index=shared_unit_index,
+                        availability=availability.get(
+                            id, pd.Series(1.0, index, name=id)
+                        ),
+                        market_prices=unit.get("market_prices"),
+                        forecast_algorithms=unit_forecast_algorithms,
+                        forecast_registries=None,
+                        fuel_prices=fuel_prices_df,
+                        normalized_load_profile=get_unit_forecast_column(
+                            forecasts_df, id, "normalized_load_profile"
+                        ),
+                        clinker_demand=get_unit_forecast_column(
+                            forecasts_df, id, "clinker_demand"
+                        ),
+                        electricity_price_flex=get_unit_forecast_column(
+                            forecasts_df, id, "electricity_price_flex"
+                        ),
+                        thermal_storage_schedule=(
+                            storage_schedule if storage_schedule is not None else 0
+                        ),
+                        availability_profiles={
+                            tech: get_unit_forecast_column(
+                                forecasts_df, id, f"{tech}_availability"
+                            )
+                            for tech in (
+                                "preheater",
+                                "calciner",
+                                "kiln",
+                            )
+                        },
+                    )
                 if type == "hydrogen_plant":
                     unit_forecasts[id] = HydrogenForecaster(
                         index=shared_unit_index,

assume/units/__init__.py

@@ -8,6 +8,7 @@
 from assume.units.powerplant import PowerPlant
 from assume.units.storage import Storage
 from assume.units.steel_plant import SteelPlant
+from assume.units.cement_plant import CementPlant
 from assume.units.steam_generation_plant import SteamPlant
 from assume.units.hydrogen_plant import HydrogenPlant
 from assume.units.building import Building
@@ -19,6 +20,7 @@
     "exchange": Exchange,
     "storage": Storage,
     "steel_plant": SteelPlant,
+    "cement_plant": CementPlant,
     "hydrogen_plant": HydrogenPlant,
     "steam_plant": SteamPlant,
     "building": Building,