Select technology data for a single year

Author: tsmbland

src/muse/sectors/sector.py

@@ -216,14 +216,12 @@ def group_assets(x: xr.DataArray) -> xr.DataArray:
 
         # Investments
         # uses technology data from the investment year
+        techs = self.technologies.sel(year=investment_year, drop=True)
         for subsector in self.subsectors:
-            subsector.invest(
-                technologies=self.technologies.sel(year=investment_year),
-                market=market,
-            )
+            subsector.invest(technologies=techs, market=market)
 
         # Full output data
-        supply, consume, costs = self.market_variables(market, self.technologies)
+        supply, consume, costs = self.market_variables(market, techs)
         self.output_data = xr.Dataset(
             dict(
                 supply=supply,
@@ -293,6 +291,7 @@ def market_variables(self, market: xr.Dataset, technologies: xr.Dataset) -> Any:
         from muse.quantities import consumption
         from muse.utilities import broadcast_techs
 
+        assert "year" not in technologies.dims
         years = market.year.values
         capacity = self.capacity.interp(year=years, **self.interpolation)
 

src/muse/utilities.py

@@ -201,6 +201,9 @@ def broadcast_techs(
     This function broadcast the first representation to the shape and coordinates
     of the second.
 
+    This function does not support technologies with a 'year' dimension. Please select
+    technology data for a specific year before calling this function.
+
     Arguments:
         technologies: The dataset to broadcast
         template: the dataset or data-array to use as a template
@@ -212,6 +215,8 @@ def broadcast_techs(
         kwargs: further arguments are used initial filters over the
             `technologies` dataset.
     """
+    assert "year" not in technologies.dims
+
     # this assert will trigger if 'year' is changed to 'installed' in
     # technologies, because then this function should be modified.
     assert "installed" not in technologies.dims
@@ -226,19 +231,6 @@ def broadcast_techs(
     first_sel.update({k: v for k, v in kwargs.items() if k != "year"})
     techs = technologies.sel(first_sel)
 
-    if "year" in technologies.dims:
-        year = None
-        if installed_as_year and "installed" in names:
-            year = template["installed"]
-        elif (not installed_as_year) and "year" in template.dims:
-            year = template["year"]
-        if year is not None and len(year) > 0:
-            techs = techs.interp(
-                year=sorted(set(cast(Iterable, year.values))), method=interpolation
-            )
-        if installed_as_year and "installed" in names:
-            techs = techs.rename(year="installed")
-
     second_sel = {n: template[n] for n in template.coords if n in techs.dims}
 
     return techs.sel(second_sel)

tests/test_quantities.py

@@ -5,57 +5,62 @@
 from pytest import approx, fixture
 
 
+@fixture
+def _technologies(technologies):
+    return technologies.sel(year=2010, drop=True)
+
+
 @fixture
 def production(
-    technologies: xr.Dataset, capacity: xr.DataArray, timeslice
+    _technologies: xr.Dataset, capacity: xr.DataArray, timeslice
 ) -> xr.DataArray:
     from muse.timeslices import broadcast_timeslice, distribute_timeslice
 
     return (
         broadcast_timeslice(capacity)
-        * distribute_timeslice(technologies.fixed_outputs)
-        * broadcast_timeslice(technologies.utilization_factor)
+        * distribute_timeslice(_technologies.fixed_outputs)
+        * broadcast_timeslice(_technologies.utilization_factor)
     )
 
 
-def test_supply_enduse(technologies, capacity, timeslice):
+def test_supply_enduse(_technologies, capacity, timeslice):
     """End-use part of supply."""
     from muse.commodities import is_enduse
     from muse.quantities import maximum_production, supply
 
-    production = maximum_production(technologies, capacity)
+    production = maximum_production(_technologies, capacity)
     demand = production.sum("asset") + 1
-    spl = supply(capacity, demand, technologies).where(
-        is_enduse(technologies.comm_usage), 0
+    spl = supply(capacity, demand, _technologies).where(
+        is_enduse(_technologies.comm_usage), 0
     )
     assert (abs(spl - production) < 1e-12).all()
     assert (spl.sum("asset") < demand).all()
 
     demand = production.sum("asset") * 0.7
-    spl = supply(capacity, demand, technologies).where(
-        is_enduse(technologies.comm_usage), 0
+    spl = supply(capacity, demand, _technologies).where(
+        is_enduse(_technologies.comm_usage), 0
     )
     assert (spl <= production + 1e-12).all()
     assert (
         abs(spl.sum("asset") - demand.where(production.sum("asset") > 0, 0)) < 1e-12
     ).all()
 
 
-def test_supply_emissions(technologies, capacity, timeslice):
+def test_supply_emissions(_technologies, capacity, timeslice):
     """Emission part of supply."""
     from muse.commodities import is_enduse, is_pollutant
     from muse.quantities import emission, maximum_production, supply
 
-    production = maximum_production(technologies, capacity)
-    spl = supply(capacity, production.sum("asset") + 1, technologies)
-    msn = emission(spl.where(is_enduse(spl.comm_usage), 0), technologies.fixed_outputs)
+    production = maximum_production(_technologies, capacity)
+    spl = supply(capacity, production.sum("asset") + 1, _technologies)
+    msn = emission(spl.where(is_enduse(spl.comm_usage), 0), _technologies.fixed_outputs)
     actual, expected = xr.broadcast(
         spl.sel(commodity=is_pollutant(spl.comm_usage)), msn
     )
     assert actual.values == approx(expected.values)
 
 
-def test_gross_margin(technologies, capacity, market, timeslice):
+def test_gross_margin(_technologies, capacity, market, timeslice):
     from muse.commodities import is_enduse, is_fuel, is_pollutant
     from muse.quantities import gross_margin
 
@@ -67,22 +72,22 @@ def test_gross_margin(technologies, capacity, market, timeslice):
     # we modify the variables to have just the values we want for the testing
     selected = capacity.technology.values[0]
 
-    technologies = technologies.sel(technology=technologies.technology == selected)
+    _technologies = _technologies.sel(technology=_technologies.technology == selected)
     capa = capacity.where(capacity.technology == selected, drop=True)
 
     # Filtering commodity outputs
-    usage = technologies.comm_usage
+    usage = _technologies.comm_usage
 
-    technologies.var_par[:] = vp = 2
-    technologies.var_exp[:] = ve = 0.5
-    technologies.fixed_inputs[{"commodity": is_fuel(usage)}] = fuels = 2
-    technologies.fixed_outputs[{"commodity": is_pollutant(usage)}] = envs = 10
-    technologies.fixed_outputs[{"commodity": is_enduse(usage)}] = prod = 5
+    _technologies.var_par[:] = vp = 2
+    _technologies.var_exp[:] = ve = 0.5
+    _technologies.fixed_inputs[{"commodity": is_fuel(usage)}] = fuels = 2
+    _technologies.fixed_outputs[{"commodity": is_pollutant(usage)}] = envs = 10
+    _technologies.fixed_outputs[{"commodity": is_enduse(usage)}] = prod = 5
 
     market.prices[:] = prices = 3
     market.prices[{"commodity": is_pollutant(usage)}] = env_prices = 6
     # We expect a xr.DataArray with 1 replacement technology
-    actual = gross_margin(technologies, capa, market.prices)
+    actual = gross_margin(_technologies, capa, market.prices)
 
     revenues = prices * prod * sum(is_enduse(usage))
     env_costs = env_prices * envs * sum(is_pollutant(usage))
@@ -96,25 +101,25 @@ def test_gross_margin(technologies, capacity, market, timeslice):
     assert actual.values == approx(expected.values)
 
 
-def test_consumption(technologies, production, market):
+def test_consumption(_technologies, production, market):
     from muse.quantities import consumption
 
     # Prices not provided, so flexible inputs are ignored
-    consump = consumption(technologies, production)
+    consump = consumption(_technologies, production)
     assert set(production.dims) == set(consump.dims)
 
     # Prices provided, but no flexible inputs -> should be the same as above
-    technologies.flexible_inputs[:] = 0
-    consump2 = consumption(technologies, production, market.prices)
+    _technologies.flexible_inputs[:] = 0
+    consump2 = consumption(_technologies, production, market.prices)
     assert consump2.values == approx(consump.values)
 
     # Flexible inputs considered
-    consump3 = consumption(technologies, production, market.prices)
+    consump3 = consumption(_technologies, production, market.prices)
     assert set(production.dims) == set(consump3.dims)
 
 
 def test_production_aggregate_asset_view(
-    capacity: xr.DataArray, technologies: xr.Dataset
+    capacity: xr.DataArray, _technologies: xr.Dataset
 ):
     """Production when capacity has format of agent.sector.
 
@@ -123,36 +128,36 @@ def test_production_aggregate_asset_view(
     from muse.commodities import is_enduse
     from muse.quantities import maximum_production
 
-    technologies: xr.Dataset = technologies[  # type:ignore
+    _technologies: xr.Dataset = _technologies[  # type:ignore
         ["fixed_outputs", "utilization_factor"]
     ]
 
-    enduses = is_enduse(technologies.comm_usage)
+    enduses = is_enduse(_technologies.comm_usage)
     assert enduses.any()
 
-    technologies.fixed_outputs[:] = 1
-    technologies.utilization_factor[:] = 1
-    prod = maximum_production(technologies, capacity)
+    _technologies.fixed_outputs[:] = 1
+    _technologies.utilization_factor[:] = 1
+    prod = maximum_production(_technologies, capacity)
     assert set(prod.dims) == set(capacity.dims).union({"commodity", "timeslice"})
     assert prod.sel(commodity=~enduses).values == approx(0)
     prod, expected = xr.broadcast(
         prod.sel(commodity=enduses).sum("timeslice"), capacity
     )
     assert prod.values == approx(expected.values)
 
-    technologies.fixed_outputs[:] = fouts = 2
-    technologies.utilization_factor[:] = ufact = 0.5
-    prod = maximum_production(technologies, capacity)
+    _technologies.fixed_outputs[:] = fouts = 2
+    _technologies.utilization_factor[:] = ufact = 0.5
+    prod = maximum_production(_technologies, capacity)
     assert prod.sel(commodity=~enduses).values == approx(0)
     assert set(prod.dims) == set(capacity.dims).union({"commodity", "timeslice"})
     prod, expected = xr.broadcast(
         prod.sel(commodity=enduses).sum("timeslice"), capacity
     )
     assert prod.values == approx(fouts * ufact * expected.values)
 
-    technologies.fixed_outputs[:] = fouts = 3
-    technologies.utilization_factor[:] = ufact = 0.5
-    prod = maximum_production(technologies, capacity)
+    _technologies.fixed_outputs[:] = fouts = 3
+    _technologies.utilization_factor[:] = ufact = 0.5
+    prod = maximum_production(_technologies, capacity)
     assert prod.sel(commodity=~enduses).values == approx(0)
     assert set(prod.dims) == set(capacity.dims).union({"commodity", "timeslice"})
     prod, expected = xr.broadcast(
@@ -162,93 +167,93 @@ def test_production_aggregate_asset_view(
 
 
 def test_production_agent_asset_view(
-    capacity: xr.DataArray, technologies: xr.Dataset, timeslice
+    capacity: xr.DataArray, _technologies: xr.Dataset, timeslice
 ):
     """Production when capacity has format of agent.assets.capacity."""
     from muse.utilities import coords_to_multiindex, reduce_assets
 
     capacity = coords_to_multiindex(reduce_assets(capacity)).unstack("asset").fillna(0)
-    test_production_aggregate_asset_view(capacity, technologies)
+    test_production_aggregate_asset_view(capacity, _technologies)
 
 
-def test_capacity_in_use(production: xr.DataArray, technologies: xr.Dataset):
+def test_capacity_in_use(production: xr.DataArray, _technologies: xr.Dataset):
     from muse.commodities import is_enduse
     from muse.quantities import capacity_in_use
 
-    technologies: xr.Dataset = technologies[  # type: ignore
+    _technologies: xr.Dataset = _technologies[  # type: ignore
         ["fixed_outputs", "utilization_factor"]
     ]
     production[:] = prod = 10
-    technologies.fixed_outputs[:] = fout = 5
-    technologies.utilization_factor[:] = ufac = 2
+    _technologies.fixed_outputs[:] = fout = 5
+    _technologies.utilization_factor[:] = ufac = 2
 
-    enduses = is_enduse(technologies.comm_usage)
-    capa = capacity_in_use(production, technologies, max_dim=None)
+    enduses = is_enduse(_technologies.comm_usage)
+    capa = capacity_in_use(production, _technologies, max_dim=None)
     assert "commodity" in capa.dims
     capa, expected = xr.broadcast(capa, enduses * prod / fout / ufac)
     assert capa.values == approx(expected.values)
 
-    capa = capacity_in_use(production, technologies)
+    capa = capacity_in_use(production, _technologies)
     assert "commodity" not in capa.dims
     assert capa.values == approx(prod / fout / ufac)
 
     maxcomm = np.random.choice(production.commodity.sel(commodity=enduses).values)
     production.loc[{"commodity": maxcomm}] = prod = 11
-    capa = capacity_in_use(production, technologies)
+    capa = capacity_in_use(production, _technologies)
     assert "commodity" not in capa.dims
     assert capa.values == approx(prod / fout / ufac)
 
 
-def test_emission(production: xr.DataArray, technologies: xr.Dataset):
+def test_emission(production: xr.DataArray, _technologies: xr.Dataset):
     from muse.commodities import is_enduse, is_pollutant
     from muse.quantities import emission
 
-    envs = is_pollutant(technologies.comm_usage)
-    technologies = cast(xr.Dataset, technologies[["fixed_outputs"]])
-    technologies.fixed_outputs[{"commodity": envs}] = fout = 1.5
-    technologies.fixed_outputs[{"commodity": ~envs}] = 2
+    envs = is_pollutant(_technologies.comm_usage)
+    _technologies = cast(xr.Dataset, _technologies[["fixed_outputs"]])
+    _technologies.fixed_outputs[{"commodity": envs}] = fout = 1.5
+    _technologies.fixed_outputs[{"commodity": ~envs}] = 2
 
-    enduses = is_enduse(technologies.comm_usage.sel(commodity=production.commodity))
+    enduses = is_enduse(_technologies.comm_usage.sel(commodity=production.commodity))
     production[{"commodity": enduses}] = prod = 0.5
     production[{"commodity": ~enduses}] = 5
 
-    em = emission(production, technologies)
+    em = emission(production, _technologies)
     assert em.commodity.isin(envs.commodity).all()
     assert em.values == approx(fout * enduses.sum().values * prod)
 
 
-def test_min_production(technologies, capacity, timeslice):
+def test_min_production(_technologies, capacity, timeslice):
     """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 "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)
+    _technologies["minimum_service_factor"] = 0.5
+    production = minimum_production(_technologies, capacity)
     assert not (production == 0).all()
-    assert (production <= maximum_production(technologies, capacity)).all()
+    assert (production <= maximum_production(_technologies, capacity)).all()
 
 
-def test_supply_capped_by_min_service(technologies, capacity, timeslice):
+def test_supply_capped_by_min_service(_technologies, capacity, timeslice):
     """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)
+    _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 = supply(capacity, demand, _technologies)
     spl = spl.sel(commodity=spl.comm_usage == CommodityUsage.PRODUCT).sum(
         ["year", "asset"]
     )
@@ -259,8 +264,8 @@ def test_supply_capped_by_min_service(technologies, capacity, timeslice):
 
     # 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)
+    del _technologies["minimum_service_factor"]
+    spl = supply(capacity, demand, _technologies)
     spl = spl.sel(commodity=spl.comm_usage == CommodityUsage.PRODUCT).sum(
         ["year", "asset"]
     )
@@ -271,8 +276,8 @@ def test_supply_capped_by_min_service(technologies, capacity, timeslice):
     assert (spl <= minprod).all()
 
 
-def test_production_amplitude(production, technologies):
+def test_production_amplitude(production, _technologies):
     from muse.quantities import production_amplitude
 
-    result = production_amplitude(production, technologies)
+    result = production_amplitude(production, _technologies)
     assert set(result.dims) == set(production.dims) - {"commodity"}