diff --git a/src/muse/quantities.py b/src/muse/quantities.py index 7daf87b15..016be9fc0 100644 --- a/src/muse/quantities.py +++ b/src/muse/quantities.py @@ -49,6 +49,7 @@ def supply( production_method = maximum_production maxprod = production_method(technologies, capacity) + minprod = minimum_production(technologies, capacity) size = np.array(maxprod.region).size # in presence of trade demand needs to map maxprod dst_region if ( @@ -88,12 +89,14 @@ def supply( expanded_demand = (demand * maxprod / maxprod.sum(demsum)).fillna(0) expanded_maxprod = (maxprod * demand / demand.sum(prodsum)).fillna(0) - + expanded_minprod = (minprod * demand / demand.sum(prodsum)).fillna(0) expanded_demand = expanded_demand.reindex_like(maxprod) + expanded_minprod = expanded_minprod.reindex_like(maxprod) result = expanded_demand.where( expanded_demand <= expanded_maxprod, expanded_maxprod ) + result = result.where(result >= expanded_minprod, expanded_minprod) # add production of environmental pollutants env = is_pollutant(technologies.comm_usage) @@ -524,6 +527,61 @@ def group_assets(x: xr.DataArray) -> xr.DataArray: return result +def minimum_production(technologies: xr.Dataset, capacity: xr.DataArray, **filters): + r"""Minimum production for a given capacity. + + Given a capacity :math:`\mathcal{A}_{t, \iota}^r`, the minimum service factor + :math:`\alpha^r_{t, \iota}` and the the fixed outputs of each technology + :math:`\beta^r_{t, \iota, c}`, then the result production is: + + .. math:: + + P_{t, \iota}^r = + \alpha^r_{t, \iota}\beta^r_{t, \iota, c}\mathcal{A}_{t, \iota}^r + + The dimensions above are only indicative. The function should work with many + different input values, e.g. with capacities expanded over time-slices :math:`t` or + agents :math:`i`. + + Arguments: + capacity: Capacity of each technology of interest. In practice, the capacity can + refer to asset capacity, the max capacity, or the capacity-in-use. + technologies: xr.Dataset describing the features of the technologies of + interests. It should contain `fixed_outputs` and `minimum_service_factor`. + Its shape is matched to `capacity` using `muse.utilities.broadcast_techs`. + filters: keyword arguments are used to filter down the capacity and + technologies. Filters not relevant to the quantities of interest, i.e. + filters that are not a dimension of `capacity` or `technologies`, are + silently ignored. + + Return: + `capacity * fixed_outputs * minimum_service_factor`, whittled down according to + the filters and the set of technologies in `capacity`. + """ + from muse.commodities import is_enduse + from muse.utilities import broadcast_techs, filter_input + + capa = filter_input( + capacity, **{k: v for k, v in filters.items() if k in capacity.dims} + ) + + if "minimum_service_factor" not in technologies: + return xr.zeros_like(capa) + + btechs = broadcast_techs( # type: ignore + cast( + xr.Dataset, + technologies[["fixed_outputs", "minimum_service_factor"]], + ), + capa, + ) + ftechs = filter_input( + btechs, **{k: v for k, v in filters.items() if k in btechs.dims} + ) + result = capa * ftechs.fixed_outputs * ftechs.minimum_service_factor + return result.where(is_enduse(result.comm_usage), 0) + + def capacity_to_service_demand( demand: xr.DataArray, technologies: xr.Dataset, diff --git a/tests/test_quantities.py b/tests/test_quantities.py index 2d0711ea5..f0cf73072 100644 --- a/tests/test_quantities.py +++ b/tests/test_quantities.py @@ -598,3 +598,57 @@ def test_costed_production_with_minimum_service(market, capacity, technologies, assert (actual[dim] == maxdemand[dim]).all() assert (actual >= 0.9 * maxdemand - 1e-8).all() assert (result >= minprod - 1e-8).all() + + +def test_min_production(technologies, capacity): + """Test minimum production quantity.""" + from muse.quantities import maximum_production, minimum_production + + # If no minimum service factor is defined, the minimum production is zero + assert "minimum_service_factor" not in technologies + production = minimum_production(technologies, capacity) + assert (production == 0).all() + + # If minimum service factor is defined, then the minimum production is not zero + # and it is less than the maximum production + technologies["minimum_service_factor"] = 0.5 + production = minimum_production(technologies, capacity) + assert not (production == 0).all() + assert (production <= maximum_production(technologies, capacity)).all() + + +def test_supply_capped_by_min_service(technologies, capacity): + """Test supply is capped by the minimum service.""" + from muse.commodities import CommodityUsage + from muse.quantities import minimum_production, supply + + technologies["minimum_service_factor"] = 0.3 + minprod = minimum_production(technologies, capacity) + + # If minimum service factor is defined, then the minimum production is not zero + assert not (minprod == 0).all() + + # And even if the demand is smaller than the minimum production, the supply + # should be equal to the minimum production + demand = minprod / 2 + spl = supply(capacity, demand, technologies) + spl = spl.sel(commodity=spl.comm_usage == CommodityUsage.PRODUCT).sum( + ["year", "asset"] + ) + minprod = minprod.sel(commodity=minprod.comm_usage == CommodityUsage.PRODUCT).sum( + ["year", "asset"] + ) + assert (spl == approx(minprod)).all() + + # But if there is not minimum service factor, the supply should be equal to the + # demand and should not be capped by the minimum production + del technologies["minimum_service_factor"] + spl = supply(capacity, demand, technologies) + spl = spl.sel(commodity=spl.comm_usage == CommodityUsage.PRODUCT).sum( + ["year", "asset"] + ) + demand = demand.sel(commodity=demand.comm_usage == CommodityUsage.PRODUCT).sum( + ["year", "asset"] + ) + assert (spl == approx(demand)).all() + assert (spl <= minprod).all()