Lennart Schürmann <[email protected]>
 
 `MIT license <https://github.com/oemof/oemof-solph/blob/dev/LICENSE>`_
 
 """
-
 
 from oemof.solph import EnergySystem
 from oemof.solph import Model
 from oemof.solph import buses
 from oemof.solph import components as cmp

         \times P_{\mathrm{out},n}(p, t)
             \quad \forall t \in T, \forall n in {1,2} \\
         &
 
     """
+
     CONSTRAINT_GROUP = True
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 

     +--------------------+------------------------+------+--------------------------------------------+
 
     Note that :math:`P_{max}(t) \cdot Y(t)` is merged into one variable,
     called `status_nominal[n, o, p, t]`.
     """  # noqa: E501
+
     CONSTRAINT_GROUP = True
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 

             self.es = energysystem
         self.timeincrement = kwargs.get("timeincrement", self.es.timeincrement)
 
         self.objective_weighting = kwargs.get(
             "objective_weighting",
-            [1] * len(self.timeincrement)
-            if self.timeincrement is not None
-            else [1],
+            (
+                [1] * len(self.timeincrement)
+                if self.timeincrement is not None
+                else [1]
+            ),
         )
 
         self._constraint_groups = type(self).CONSTRAINT_GROUPS + kwargs.get(
             "constraint_groups", []
         )

             # Construct weighting from occurrences and order
             self.tsam_weighting = list(
                 self.es.tsa_parameters[p]["occurrences"][k]
                 for p in self.PERIODS
                 for k in range(len(self.es.tsa_parameters[p]["occurrences"]))
-                for _ in range(
-                    self.es.tsa_parameters[p]["timesteps"]
-                )
+                for _ in range(self.es.tsa_parameters[p]["timesteps"])
             )
             self.CLUSTERS = po.Set(
                 initialize=list(
                     range(
                         sum(

         """
         return [
             (p, k, t)
             for p in range(len(self.es.tsa_parameters))
             for k in range(len(self.es.tsa_parameters[p][cluster_type]))
-            for t in range(
-                self.es.tsa_parameters[p]["timesteps"] + offset
-            )
+            for t in range(self.es.tsa_parameters[p]["timesteps"] + offset)
         ]

     :math:`\eta_{el,min,woDH}`      `Eta_el_min_woDH[n,t]`  P    el. eff. at min. fuel flow w/o distr. heating
     :math:`\eta_{el,max,woDH}`      `Eta_el_max_woDH[n,t]`  P    el. eff. at max. fuel flow w/o distr. heating
     =============================== ======================= ==== =============================================
 
     """  # noqa: E501
+
     CONSTRAINT_GROUP = True
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 

     :class:`~oemof.solph.components.experimental._piecewise_linear_converter.PiecewiseLinearConverter`
 
     **The following constraints are created:**
 
     """
+
     CONSTRAINT_GROUP = True
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 

     def _outputs_tsam():
         OUTPUTS = po.Set(initialize=output_levels.keys())
         setattr(model, f"{name}_OUTPUTS", OUTPUTS)
 
         active_output = po.Var(
-            OUTPUTS, model.TIMEINDEX_TYPICAL_CLUSTER_OFFSET, domain=po.Binary, bounds=(0, 1)
+            OUTPUTS,
+            model.TIMEINDEX_TYPICAL_CLUSTER_OFFSET,
+            domain=po.Binary,
+            bounds=(0, 1),
         )
         setattr(model, f"{name}_active_output", active_output)
 
         constraint_name = f"{name}_output_active_constraint"
 

                 t = m.get_timestep_from_tsam_timestep(p, k, g)
                 for o in output_levels:
                     getattr(m, constraint_name).add(
                         (o, p, k, g),
                         m.GenericStorageBlock.storage_content_intra[
-                            storage_component,  p, k, g + 1
+                            storage_component, p, k, g + 1
                         ]
                         / storage_component.nominal_storage_capacity
                         >= active_output[o, p, k, g] * output_levels[o],
                     )
 

     def _inputs_tsam():
         INPUTS = po.Set(initialize=input_levels.keys())
         setattr(model, f"{name}_INPUTS", INPUTS)
 
         inactive_input = po.Var(
-            INPUTS, model.TIMEINDEX_TYPICAL_CLUSTER_OFFSET, domain=po.Binary, bounds=(0, 1)
+            INPUTS,
+            model.TIMEINDEX_TYPICAL_CLUSTER_OFFSET,
+            domain=po.Binary,
+            bounds=(0, 1),
         )
         setattr(model, f"{name}_active_input", inactive_input)
 
         constraint_name = f"{name}_input_active_constraint"
 

 
     for lb in labels:
         subset = (
             df.T.groupby(
                 lambda x1: (
-                    lambda fr, to, ty: "output"
-                    if (fr == lb and ty == "flow")
-                    else "input"
-                    if (to == lb and ty == "flow")
-                    else "level"
-                    if (fr == lb and ty != "flow")
-                    else None
+                    lambda fr, to, ty: (
+                        "output"
+                        if (fr == lb and ty == "flow")
+                        else (
+                            "input"
+                            if (to == lb and ty == "flow")
+                            else (
+                                "level"
+                                if (fr == lb and ty != "flow")
+                                else None
+                            )
+                        )
+                    )
                 )(*x1)
             )
             .sum()
             .T
         )

             bounds = (
                 n.nominal_storage_capacity * n.min_storage_level[t],
                 n.nominal_storage_capacity * n.max_storage_level[t],
             )
             return bounds
+
         def _storage_content_bound_intra_rule(block, n, p, k, g):
             """
             Rule definition for bounds of storage_content variable of
             storage n in timestep t.
             """

             bounds = (
                 n.nominal_storage_capacity * n.min_storage_level[t],
                 n.nominal_storage_capacity * n.max_storage_level[t],
             )
             return bounds
+
         if not m.TSAM_MODE:
             self.storage_content = Var(
                 self.STORAGES, m.TIMEPOINTS, bounds=_storage_content_bound_rule
             )
 

                     self.storage_content[n, 0].fix()
         else:
             for n in group:
                 if n.multiple_tsam_timegrid:
                     self.storage_content_inter = Var(
-                        self.STORAGES, m.CLUSTERS_OFFSET, within=NonNegativeReals
+                        self.STORAGES,
+                        m.CLUSTERS_OFFSET,
+                        within=NonNegativeReals,
                     )
                     self.storage_content_intra = Var(
                         self.STORAGES, m.TIMEINDEX_TYPICAL_CLUSTER_OFFSET
                     )
                 else:
                     self.storage_content_intra = Var(
-                        self.STORAGES, m.TIMEINDEX_TYPICAL_CLUSTER_OFFSET, bounds=_storage_content_bound_intra_rule
+                        self.STORAGES,
+                        m.TIMEINDEX_TYPICAL_CLUSTER_OFFSET,
+                        bounds=_storage_content_bound_intra_rule,
                     )
             # set the initial intra storage content
             # first timestep in intra storage is always zero
             for n in group:
                 for p, k in m.TYPICAL_CLUSTERS:
                     if n.multiple_tsam_timegrid:
                         self.storage_content_intra[n, p, k, 0] = 0
                         self.storage_content_intra[n, p, k, 0].fix()
                     else:
                         if n.initial_storage_level is not None:
-                            self.storage_content_intra[n, p, k, 0] = n.initial_storage_level * \
-                                                                     n.nominal_storage_capacity
+                            self.storage_content_intra[n, p, k, 0] = (
+                                n.initial_storage_level
+                                * n.nominal_storage_capacity
+                            )
                             self.storage_content_intra[n, p, k, 0].fix()
-                if n.initial_storage_level is not None and n.multiple_tsam_timegrid:
-                        self.storage_content_inter[n, 0] = (
-                            n.initial_storage_level * n.nominal_storage_capacity
-                        )
-                        self.storage_content_inter[n, 0].fix()
-
+                if (
+                    n.initial_storage_level is not None
+                    and n.multiple_tsam_timegrid
+                ):
+                    self.storage_content_inter[n, 0] = (
+                        n.initial_storage_level * n.nominal_storage_capacity
+                    )
+                    self.storage_content_inter[n, 0].fix()
 
         #  ************* Constraints ***************************
 
         def _storage_inter_minimum_level_rule(block):
             # See FINE implementation at

             k = m.es.tsa_parameters[p]["order"][ii]
 
             # Calculate inter losses over whole typical period
             t0 = m.get_timestep_from_tsam_timestep(p, k, 0)
             losses = math.prod(
-                (1 - n.loss_rate[t0 + s])
-                ** m.es.tsa_parameters[p]["segments"][(k, s)]
-                if "segments" in m.es.tsa_parameters[p]
-                else 1 - n.loss_rate[t0 + s]
+                (
+                    (1 - n.loss_rate[t0 + s])
+                    ** m.es.tsa_parameters[p]["segments"][(k, s)]
+                    if "segments" in m.es.tsa_parameters[p]
+                    else 1 - n.loss_rate[t0 + s]
+                )
                 for s in range(m.es.tsa_parameters[p]["timesteps"])
             )
 
             expr = 0
             expr += block.storage_content_inter[n, i + 1]

             )
         else:
             for n in self.STORAGES:
                 if n.multiple_tsam_timegrid:
                     self.balanced_cstr = Constraint(
-                    self.STORAGES_BALANCED, rule=_balanced_inter_storage_rule
+                        self.STORAGES_BALANCED,
+                        rule=_balanced_inter_storage_rule,
                     )
 
         def _power_coupled(block):
             """
             Rule definition for constraint to connect the input power

             )
             expr = 0
             expr += block.storage_content_inter[n, i + 1]
             expr += -block.storage_content_inter[n, i] * (
                 1 - n.loss_rate[t]
-            ) ** (
-                m.timeincrement[t]
-                * m.es.tsa_parameters[p]["timesteps"]
-            )
+            ) ** (m.timeincrement[t] * m.es.tsa_parameters[p]["timesteps"])
             expr += -self.storage_content_intra[
                 n, p, k, m.es.tsa_parameters[p]["timesteps"]
             ]
             return expr == 0
 

     # Extraction steps that are the same for both model types
     df = create_dataframe(model)
 
     # create a dict of dataframes keyed by oemof tuples
     df_dict = {
-        k
-        if len(k) > 1
-        else (k[0], None): v[["timestep", "variable_name", "value"]]
+        k if len(k) > 1 else (k[0], None): v[
+            ["timestep", "variable_name", "value"]
+        ]
         for k, v in df.groupby("oemof_tuple")
     }
 
     # Define index
     if model.es.tsa_parameters:

                 inter_value = soc["inter"].iloc[i_offset + i]["value"]
                 inter_series = (
                     pd.Series(
                         itertools.accumulate(
                             (
-                                (1 - storage.loss_rate[t])
-                                ** tsa_period["segments"][(k, t - t0)]
-                                if "segments" in tsa_period
-                                else 1 - storage.loss_rate[t]
+                                (
+                                    (1 - storage.loss_rate[t])
+                                    ** tsa_period["segments"][(k, t - t0)]
+                                    if "segments" in tsa_period
+                                    else 1 - storage.loss_rate[t]
+                                )
                                 for t in range(
-                                t0,
-                                t0 + timesteps - 1,
-                            )
+                                    t0,
+                                    t0 + timesteps - 1,
+                                )
                             ),
                             operator.mul,
                             initial=1,
                         )
                     )

                     soc_frame[:-1] if is_last_timestep else soc_frame,
                     tsa_period["segments"],
                     k,
                 )
                 if is_last_timestep:
-                    soc_disaggregated.loc[
-                        len(soc_disaggregated)
-                    ] = soc_frame.iloc[-1]
+                    soc_disaggregated.loc[len(soc_disaggregated)] = (
+                        soc_frame.iloc[-1]
+                    )
                 soc_frame = soc_disaggregated
 
             soc_frames.append(soc_frame)
         i_offset += len(tsa_period["order"])
         t_offset += i_offset * tsa_period["timesteps"]

     # Disaggregate segments by linear interpolation and remove
     # last timestep afterwards (only needed for interpolation)
     interpolated_soc = soc_ts.interpolate()
     return interpolated_soc.iloc[:-1]
 
+
 def _calculate_multiplexer_actives(values, multiplexer, tsa_parameters):
     """Calculate multiplexer actives"""
     actives_frames = []
     for p, tsa_period in enumerate(tsa_parameters):
         for i, k in enumerate(tsa_period["order"]):
             timesteps = tsa_period["timesteps"]
-            actives_frames.append(pd.DataFrame(
-                values[(p, k)].iloc[0:timesteps],
-                columns=["value"])
+            actives_frames.append(
+                pd.DataFrame(
+                    values[(p, k)].iloc[0:timesteps], columns=["value"]
                 )
+            )
     actives_frames_ts = pd.concat(actives_frames)
-    actives_frames_ts["variable_name"] = values[(p, k)]["variable_name"].values[0]
+    actives_frames_ts["variable_name"] = values[(p, k)][
+        "variable_name"
+    ].values[0]
     actives_frames_ts["timestep"] = range(len(actives_frames_ts))
     return actives_frames_ts
+
 
 def _get_storage_soc_flows_and_keys(flow_dict):
     """Detect storage flows in flow dict"""
     storages = {}
     storage_keys = []

     multiplexer = {}
     multiplexer_keys = []
     for oemof_tuple, data in flow_dict.items():
         if oemof_tuple[1] is not None and not isinstance(oemof_tuple[1], int):
             continue
-        if 'multiplexer_active' in data['variable_name'].values[0]:
-            multiplexer.setdefault(oemof_tuple[0],{})
+        if "multiplexer_active" in data["variable_name"].values[0]:
+            multiplexer.setdefault(oemof_tuple[0], {})
             multiplexer_keys.append(oemof_tuple)
-            multiplexer[oemof_tuple[0]][(oemof_tuple[1], oemof_tuple[2])] = data
+            multiplexer[oemof_tuple[0]][
+                (oemof_tuple[1], oemof_tuple[2])
+            ] = data
     return multiplexer, multiplexer_keys
+
+
 def _disaggregate_tsa_timeindex(period_index, tsa_parameters):
     """Disaggregate aggregated period timeindex by using TSA parameters"""
     return pd.date_range(
         start=period_index[0],
         periods=tsa_parameters["timesteps_per_period"]

     converted into strings that represent the object labels
     e.g. results[('pp1','bus1')].
     """
     if keep_none_type:
         converted = {
-            tuple([str(e) if e is not None else None for e in k])
-            if isinstance(k, tuple)
-            else str(k)
-            if k is not None
-            else None: v
+            (
+                tuple([str(e) if e is not None else None for e in k])
+                if isinstance(k, tuple)
+                else str(k) if k is not None else None
+            ): v
             for k, v in result.items()
         }
     else:
         converted = {
             tuple(map(str, k)) if isinstance(k, tuple) else str(k): v

 by the contributors recorded in the version control history of the file,
 available from its original location oemof/tests/test_components.py
 
 SPDX-License-Identifier: MIT
 """
+
 import pytest
 
 from oemof import solph