diff --git a/.flake8 b/.flake8 index 05ec680..c7e5c32 100755 --- a/.flake8 +++ b/.flake8 @@ -3,5 +3,5 @@ exclude = .git,__pycache__,documentation max-line-length = 160 max-complexity = 13 select = B,C,E,F,W,B9 -ignore = E501, W503, E203 +ignore = E501, W503, E203, E231 diff --git a/.gitignore b/.gitignore index f44df2b..705b2f4 100644 --- a/.gitignore +++ b/.gitignore @@ -177,3 +177,6 @@ cython_debug/ *.pickle flexmeasures.log *.png + +#MacOS +.DS_Store diff --git a/README.md b/README.md index 4b31527..d13909c 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,14 @@ using the FLEXMEASURES_PLUGINS setting (a list). Alternatively, if you installed this plugin as a package (e.g. via `python setup.py install`, `pip install -e` or `pip install flexmeasures_s2` should this project be on Pypi), then "flexmeasures_s2" suffices. -2. +2. Config settings (during development) + +In your `flexmeasures.cfg`, you can set the following config settings for this plugin: + +- `FLEXMEASURES_S2_TARGET_MODE`: "energy" or "costs" +- `FLEXMEASURES_S2_PRICE_SENSOR`: sensor ID of the price sensor + +To do: move these settings to the db asset (preferred scheduler and flex-context). ## Development diff --git a/flexmeasures_s2/profile_steering/cluster_target.py b/flexmeasures_s2/profile_steering/cluster_target.py index b097746..0db49c3 100644 --- a/flexmeasures_s2/profile_steering/cluster_target.py +++ b/flexmeasures_s2/profile_steering/cluster_target.py @@ -225,3 +225,17 @@ def set_congestion_point_target( if elements is not None: self._congestion_point_targets[congestion_point_id].elements = elements + + def contains_energy_target(self) -> bool: + """ + Check if this cluster target contains any energy targets (JouleElement instances). + + Returns: + True if there are any JouleElement instances in the global target profile, False otherwise + """ + from flexmeasures_s2.profile_steering.common.target_profile import TargetProfile + + for element in self._global_target_profile.elements: + if isinstance(element, TargetProfile.JouleElement): + return True + return False diff --git a/flexmeasures_s2/profile_steering/common/proposal.py b/flexmeasures_s2/profile_steering/common/proposal.py index 003a8c0..fee5d03 100644 --- a/flexmeasures_s2/profile_steering/common/proposal.py +++ b/flexmeasures_s2/profile_steering/common/proposal.py @@ -40,22 +40,25 @@ def get_global_improvement_value(self) -> float: ) return self.global_improvement_value - # def get_cost_improvement_value(self) -> float: - # if self.cost_improvement_value is None: - # self.cost_improvement_value = self.get_cost( - # self.old_plan, self.global_diff_target - # ) - self.get_cost(self.proposed_plan, self.global_diff_target) - # return self.cost_improvement_value + def get_cost_improvement_value(self) -> float: + if self.cost_improvement_value is None: + self.cost_improvement_value = self.get_cost( + self.old_plan, self.global_diff_target + ) - self.get_cost(self.proposed_plan, self.global_diff_target) + return self.cost_improvement_value - # @staticmethod - # def get_cost(plan: JouleProfile, target_profile: TargetProfile) -> float: - # cost = 0.0 - # for i in range(target_profile.metadata.get_nr_of_timesteps()): - # joule_usage = plan.get_elements()[i] - # target_element = target_profile.get_elements()[i] - # if isinstance(target_element, TargetProfile.TariffElement): - # cost += (joule_usage / 3_600_000) * target_element.get_tariff() - # return cost + @staticmethod + def get_cost(plan: JouleProfile, target_profile: TargetProfile) -> float: + cost = 0.0 + for i in range(target_profile.metadata.nr_of_timesteps): + joule_usage = plan.elements[i] + target_element = target_profile.elements[i] + if ( + isinstance(target_element, TargetProfile.TariffElement) + and joule_usage is not None + ): + cost += (joule_usage / 3_600_000) * target_element.get_tariff() + return cost def get_congestion_improvement_value(self) -> float: if self.congestion_improvement_value is None: @@ -112,8 +115,8 @@ def is_preferred_to(self, other: "Proposal") -> bool: elif ( self.get_global_improvement_value() == other.get_global_improvement_value() - # and self.get_cost_improvement_value() - # > other.get_cost_improvement_value() + and self.get_cost_improvement_value() + > other.get_cost_improvement_value() ): return True return False diff --git a/flexmeasures_s2/profile_steering/common/target_profile.py b/flexmeasures_s2/profile_steering/common/target_profile.py index f57fd67..6b4f6ec 100644 --- a/flexmeasures_s2/profile_steering/common/target_profile.py +++ b/flexmeasures_s2/profile_steering/common/target_profile.py @@ -15,6 +15,13 @@ class JouleElement(Element): def __init__(self, joules: int): self.joules = joules + class TariffElement(Element): + def __init__(self, tariff: float): + self.tariff = tariff + + def get_tariff(self) -> float: + return self.tariff + class NullElement(Element): pass @@ -169,3 +176,30 @@ def from_joule_profile(joule_profile: JouleProfile) -> "TargetProfile": joule_profile.metadata.timestep_duration, [TargetProfile.JouleElement(e) for e in joule_profile.elements], ) + + @staticmethod + def from_tariff_values( + metadata: ProfileMetadata, tariff_values: List[float] + ) -> "TargetProfile": + """Create a TargetProfile with TariffElement instances from tariff values. + + Args: + metadata: ProfileMetadata for the target profile + tariff_values: List of tariff values (cost per kWh) + + Returns: + TargetProfile with TariffElement instances + """ + if len(tariff_values) != metadata.nr_of_timesteps: + raise ValueError( + f"Number of tariff values ({len(tariff_values)}) must match nr_of_timesteps ({metadata.nr_of_timesteps})" + ) + + elements: List[Union["TargetProfile.Element", None]] = [ + TargetProfile.TariffElement(tariff) for tariff in tariff_values + ] + return TargetProfile( + metadata.profile_start, + metadata.timestep_duration, + elements, + ) diff --git a/flexmeasures_s2/profile_steering/device_planner/frbc/fill_level_target_util.py b/flexmeasures_s2/profile_steering/device_planner/frbc/fill_level_target_util.py index 4824ee5..bc101e0 100644 --- a/flexmeasures_s2/profile_steering/device_planner/frbc/fill_level_target_util.py +++ b/flexmeasures_s2/profile_steering/device_planner/frbc/fill_level_target_util.py @@ -30,7 +30,7 @@ def from_fill_level_target_profile(fill_level_target_profile): elements = [] start = fill_level_target_profile.start_time.astimezone(timezone.utc) for element in fill_level_target_profile.elements: - end = start + timedelta(seconds=element.duration.__root__) + end = start + timedelta(seconds=element.duration.root) elements.append( FillLevelTargetElement( start, diff --git a/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_device_planner.py b/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_device_planner.py index 568467b..a5c2db7 100644 --- a/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_device_planner.py +++ b/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_device_planner.py @@ -1,6 +1,8 @@ from datetime import datetime from typing import Optional +from flask import current_app as app + from flexmeasures_s2.profile_steering.common.joule_profile import JouleProfile from flexmeasures_s2.profile_steering.common.proposal import Proposal from flexmeasures_s2.profile_steering.common.target_profile import TargetProfile @@ -21,9 +23,11 @@ from flexmeasures_s2.profile_steering.device_planner.device_planner_abstract import ( DevicePlanner, ) - +from s2python.frbc import FRBCInstruction # make sure this is a DevicePlanner + + class S2FrbcDevicePlanner(DevicePlanner): def __init__( self, @@ -201,6 +205,18 @@ def current_profile(self) -> JouleProfile: def get_device_plan(self) -> Optional[DevicePlan]: if self.accepted_plan is None: return None + app.logger.debug( + dict( + device_id=self.device_id, + device_name=self.device_name, + connection_id=self.connection_id, + energy_profile=self.accepted_plan.energy, + fill_level_profile=self.accepted_plan.fill_level, + instruction_profile=self.convert_plan_to_instructions( + self.profile_metadata, self.accepted_plan + ), + ) + ) return DevicePlan( device_id=self.device_id, device_name=self.device_name, @@ -216,16 +232,44 @@ def get_device_plan(self) -> Optional[DevicePlan]: def convert_plan_to_instructions( profile_metadata: ProfileMetadata, device_plan: S2FrbcPlan ) -> S2FrbcInstructionProfile: + import uuid + from datetime import timedelta + elements = [] actuator_configurations_per_timestep = device_plan.get_operation_mode_id() + if actuator_configurations_per_timestep is not None: - for actuator_configurations in actuator_configurations_per_timestep: - new_element = S2FrbcInstructionProfile.Element( - not actuator_configurations, actuator_configurations + for timestep_index, actuator_configurations_dict in enumerate( + actuator_configurations_per_timestep + ): + # Calculate execution time for this timestep + execution_time = profile_metadata.profile_start + timedelta( + seconds=timestep_index + * profile_metadata.timestep_duration.total_seconds() ) - elements.append(new_element) - else: - elements = [None] * profile_metadata.nr_of_timesteps + # Ensure timezone-aware datetime + if execution_time.tzinfo is None: + from datetime import timezone + + execution_time = execution_time.replace(tzinfo=timezone.utc) + + # Create instructions for each actuator at this timestep + for ( + actuator_id, + actuator_config, + ) in actuator_configurations_dict.items(): + if actuator_config is not None: + instruction = FRBCInstruction( + message_id=str(uuid.uuid4()), + id=str(uuid.uuid4()), + actuator_id=str(actuator_id), + operation_mode=str(actuator_config.operation_mode_id), + operation_mode_factor=float(actuator_config.factor), + execution_time=execution_time, + abnormal_condition=False, + ) + elements.append(instruction) + return S2FrbcInstructionProfile( profile_start=profile_metadata.profile_start, timestep_duration=profile_metadata.timestep_duration, diff --git a/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_device_state_wrapper.py b/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_device_state_wrapper.py index e62b596..a4ec7de 100644 --- a/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_device_state_wrapper.py +++ b/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_device_state_wrapper.py @@ -311,7 +311,7 @@ def get_timer_duration_milliseconds( timer = next( (t for t in actuator_description.timers if str(t.id) == timer_id), None ) - return timer.duration.__root__ if timer else 0 + return timer.duration.root if timer else 0 @staticmethod @lru_cache(maxsize=None) diff --git a/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_instruction_profile.py b/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_instruction_profile.py index c2ce3a2..c3fe6d8 100644 --- a/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_instruction_profile.py +++ b/flexmeasures_s2/profile_steering/device_planner/frbc/s2_frbc_instruction_profile.py @@ -1,35 +1,20 @@ -from typing import Dict, List +from typing import List from datetime import datetime, timedelta -from flexmeasures_s2.profile_steering.device_planner.frbc.s2_frbc_actuator_configuration import ( - S2ActuatorConfiguration, -) +from s2python.frbc import FRBCInstruction class S2FrbcInstructionProfile: - class Element: - def __init__( - self, - idle: bool, - actuator_configuration: Dict[str, S2ActuatorConfiguration], - ): - self.idle = idle - self.actuator_configuration = actuator_configuration - - def is_idle(self) -> bool: - return self.idle - + # This class represents a profile generated by the Reflex as a list of + # S2 FRBCInstruction s def __init__( self, profile_start: datetime, timestep_duration: timedelta, - elements: List[Element], + elements: List[FRBCInstruction], ): self.profile_start = profile_start self.timestep_duration = timestep_duration self.elements = elements - def default_value(self) -> "S2FrbcInstructionProfile.Element": - return S2FrbcInstructionProfile.Element(True, {}) - def __str__(self) -> str: return f"S2FrbcInstructionProfile(elements={self.elements})" diff --git a/flexmeasures_s2/profile_steering/device_planner/frbc/usage_forecast_util.py b/flexmeasures_s2/profile_steering/device_planner/frbc/usage_forecast_util.py index e277e71..9f9cbd6 100644 --- a/flexmeasures_s2/profile_steering/device_planner/frbc/usage_forecast_util.py +++ b/flexmeasures_s2/profile_steering/device_planner/frbc/usage_forecast_util.py @@ -48,7 +48,7 @@ def from_storage_usage_profile(usage_forecast): start = usage_forecast.start_time start = start.astimezone(timezone.utc) for element in usage_forecast.elements: - end = start + timedelta(seconds=element.duration.__root__) + end = start + timedelta(seconds=element.duration.root) elements.append( UsageForecastElement(start, end, element.usage_rate_expected) ) diff --git a/flexmeasures_s2/profile_steering/examples/energy_profile-D=1_B=100_S=20_T=288.json b/flexmeasures_s2/profile_steering/examples/energy_profile-D=1_B=100_S=20_T=288.json new file mode 100644 index 0000000..cd297bb --- /dev/null +++ b/flexmeasures_s2/profile_steering/examples/energy_profile-D=1_B=100_S=20_T=288.json @@ -0,0 +1 @@ +[0.0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 16245000, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 8550000, 7695000, 5985000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 5130000, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 5985000, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 5130000, 5130000, 5130000, 5130000, 5130000, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] \ No newline at end of file diff --git a/flexmeasures_s2/profile_steering/examples/example_schedule_frbc.py b/flexmeasures_s2/profile_steering/examples/example_schedule_frbc.py index 31a796b..3887b73 100644 --- a/flexmeasures_s2/profile_steering/examples/example_schedule_frbc.py +++ b/flexmeasures_s2/profile_steering/examples/example_schedule_frbc.py @@ -45,7 +45,7 @@ ids = [] # -> todo: plot of run time vs D, vs B, vs S and vs T -D = 10 # number of devices -> todo: multiprocessing on create_improved_planning +D = 2 # number of devices -> todo: multiprocessing on create_improved_planning B = 100 # number of buckets -> todo: vectorize computation of next state from current state S = 20 # number of stratification layers PLANNING_WINDOW = pd.Timedelta("PT24H") @@ -59,8 +59,9 @@ """ # Ideas for speeding up -- Parallelize device planning -- Precompute UUIDs +- Parallelize device planning - done +- Precompute UUIDs - done +- Use numpy for vectorized computation instead of list operations - done """ @@ -473,6 +474,8 @@ def plot_planning_results( nr_of_timesteps, predicted_energy_elements, target_energy_elements, + cost_elements=None, + target_cost_elements=None, ): """ Plots the energy, fill level, actuator usage, and operation mode ID lists using matplotlib. @@ -481,10 +484,9 @@ def plot_planning_results( :param nr_of_timesteps: Number of timesteps. :param predicted_energy_elements: List of predicted energy values. :param target_energy_elements: List of target energy values. + :param cost_elements: Optional list of cost values. + :param target_cost_elements: Optional list of target cost values (tariffs). """ - # Create a figure with a single subplot - fig, ax = plt.subplots(1, 1, figsize=(12, 8)) - # Generate timestep_start_times timestep_start_times = [ datetime(1970, 1, 1, tzinfo=timezone.utc) @@ -492,38 +494,97 @@ def plot_planning_results( for i in range(nr_of_timesteps) ] - # Plot both lines on the same subplot - ax.plot( - timestep_start_times, - predicted_energy_elements, - label="Predicted Energy", - color="green", - ) - ax.plot( - timestep_start_times, - target_energy_elements, - label="Target Energy", - color="red", - linestyle="dotted", - linewidth=2, - ) + if cost_elements is not None or target_cost_elements is not None: + # Create a figure with dual y-axes for energy and cost + fig, ax1 = plt.subplots(1, 1, figsize=(14, 8)) + + # Energy plot on left y-axis + ax1.plot( + timestep_start_times, + predicted_energy_elements, + label="Predicted Energy", + color="green", + linewidth=2, + ) + ax1.plot( + timestep_start_times, + target_energy_elements, + label="Target Energy", + color="red", + linestyle="dotted", + linewidth=2, + ) + ax1.set_ylabel("Energy (Joules)", color="green") + ax1.tick_params(axis="y", labelcolor="green") + ax1.grid(True, alpha=0.3) + + # Cost plot on right y-axis + ax2 = ax1.twinx() + if cost_elements is not None: + ax2.plot( + timestep_start_times, + cost_elements, + label="Predicted Cost", + color="blue", + linewidth=2, + ) + if target_cost_elements is not None: + ax2.plot( + timestep_start_times, + target_cost_elements, + label="Target Cost (Tariffs)", + color="orange", + linestyle="dashed", + linewidth=2, + ) + ax2.set_ylabel("Cost ($/kWh)", color="blue") + ax2.tick_params(axis="y", labelcolor="blue") + + # Combined legend + lines1, labels1 = ax1.get_legend_handles_labels() + lines2, labels2 = ax2.get_legend_handles_labels() + ax1.legend(lines1 + lines2, labels1 + labels2, loc="upper left") + + ax1.set_title("Energy and Cost Planning Results") + else: + # Create a figure with a single subplot for energy only + fig, ax1 = plt.subplots(1, 1, figsize=(12, 8)) + + # Plot both lines on the same subplot + ax1.plot( + timestep_start_times, + predicted_energy_elements, + label="Predicted Energy", + color="green", + ) + ax1.plot( + timestep_start_times, + target_energy_elements, + label="Target Energy", + color="red", + linestyle="dotted", + linewidth=2, + ) - # Set labels and grid - ax.set_ylabel("Energy (Joules)") - ax.set_title("Predicted vs Target Energy") - ax.legend(loc="best") - ax.grid(True) + # Set labels and grid + ax1.set_ylabel("Energy (Joules)") + ax1.set_title("Predicted vs Target Energy") + ax1.legend(loc="best") + ax1.grid(True) # Format the x-axis to show time and set ticks every 30 minutes - ax.xaxis.set_major_locator(mdates.MinuteLocator(interval=30)) - ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S")) + ax1.xaxis.set_major_locator(mdates.MinuteLocator(interval=30)) + ax1.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S")) fig.autofmt_xdate() # Adjust layout plt.tight_layout() # Save it in a directory called plots os.makedirs("plots", exist_ok=True) - plt.savefig(f"plots/my plot - D = {D} - B = {B} - S = {S} - T = {T}") + suffix = ( + "_cost" if cost_elements is not None or target_cost_elements is not None else "" + ) + plt.savefig(f"plots/my plot - D = {D} - B = {B} - S = {S} - T = {T}{suffix}") def create_device_state( @@ -594,6 +655,44 @@ def get_target_profile_elements(number_of_elements: int): return target_elements[:number_of_elements] +def get_cost_target_profile_elements(number_of_elements: int): + """Create cost target profile elements with dynamic pricing pattern.""" + cost_elements = [] + # Simulate time-of-use pricing for a 24-hour period (288 = 5-minute intervals) + intervals_per_hour = number_of_elements // 24 + + for hour in range(24): + if 0 <= hour < 6: # Night hours (00:00-06:00) - lowest cost + tariff = 0.10 # 10 cents per kWh + elif 6 <= hour < 9: # Morning ramp-up (06:00-09:00) + tariff = 0.15 # 15 cents per kWh + elif 9 <= hour < 17: # Peak hours (09:00-17:00) - highest cost + tariff = 0.30 # 30 cents per kWh + elif 17 <= hour < 21: # Evening peak (17:00-21:00) + tariff = 0.25 # 25 cents per kWh + else: # Late evening (21:00-24:00) + tariff = 0.15 # 15 cents per kWh + + # Add the tariff for all intervals in this hour + cost_elements.extend([tariff] * intervals_per_hour) + + return cost_elements[:number_of_elements] + + +def calculate_cost_from_energy_and_tariffs(energy_elements, tariff_elements): + """Calculate cost from energy usage and tariff rates.""" + cost_elements = [] + for energy, tariff in zip(energy_elements, tariff_elements): + if energy is not None and tariff is not None: + # Convert Joules to kWh and multiply by tariff + kwh = energy / 3_600_000 # 1 kWh = 3,600,000 Joules + cost = kwh * tariff + cost_elements.append(cost) + else: + cost_elements.append(0.0) + return cost_elements + + def test_planning_service_impl_with_ev_device(): """Test the PlanningServiceImpl with an EV device.""" print("Test the PlanningServiceImpl with an EV device.") @@ -691,8 +790,8 @@ def test_planning_service_impl_with_ev_device(): energy_profile = cluster_plan.get_joule_profile() # Save the energy profile to a file, - with open(f"energy_profile-D={D}_B={B}_S={S}_T={T}.json", "w") as f: - json.dump(energy_profile.elements, f) + # with open(f"energy_profile-D={D}_B={B}_S={S}_T={T}.json", "w") as f: + # json.dump(energy_profile.elements, f) # Plot the planning results plot_planning_results( @@ -702,6 +801,10 @@ def test_planning_service_impl_with_ev_device(): target_energy_elements=target_profile_elements, ) + if D == 3 or D == 10 or D == 5: + with open(f"energy_profile-D={D}_B={B}_S={S}_T={T}.json", "r") as f: + assert energy_profile.elements == json.load(f) + print("Energy profile matches expected values") # Get only the non-None plans device_plans = [plan for plan in device_plans if plan is not None] @@ -714,6 +817,131 @@ def test_planning_service_impl_with_ev_device(): assert len(energy_profile.elements) == target_metadata.nr_of_timesteps +def test_planning_service_impl_with_cost_target(): + """Test the PlanningServiceImpl with cost targeting.""" + print("Test the PlanningServiceImpl with cost targeting.") + + # Create profile metadata + target_metadata = ProfileMetadata( + profile_start=datetime(1970, 1, 1, tzinfo=timezone.utc), + timestep_duration=timedelta(seconds=TIMESTEP_DURATION), + nr_of_timesteps=T, + ) + plan_due_by_date = target_metadata.profile_start + timedelta(seconds=10) + + # Create cost target profile instead of energy target + cost_target_elements = get_cost_target_profile_elements(T) + + # Create a cost-based target profile + cost_target_profile = TargetProfile.from_tariff_values( + target_metadata, cost_target_elements + ) + + device_states = [ + create_device_state( + f"battery{i + 1}", + datetime.fromtimestamp(3600), + timezone(timedelta(hours=1)), + ) + for i in range(D) + ] + + # Create Dictionary of device states + device_states_dict = { + device_state.device_id: device_state for device_state in device_states + } + + # Create congestion points mapping + congestion_points_by_connection_id = { + device_id: "" for device_id in device_states_dict.keys() + } + + # Create a cluster state using the list of device states + cluster_state = ClusterState( + datetime.now(), device_states_dict, congestion_points_by_connection_id + ) + + # Create a cluster target with cost target + cluster_target = ClusterTarget( + datetime.now(), + None, + None, + global_target_profile=cost_target_profile, + congestion_point_targets={}, + ) + + config = PlanningServiceConfig( + energy_improvement_criterion=10.0, + cost_improvement_criterion=1.0, + congestion_retry_iterations=10, + multithreaded=False, + ) + + print("Generating cost-optimized plan!") + + # Create planning service implementation + service = PlanningServiceImpl(config) + + # Act - Generate a plan + start_time = time.time() + cluster_plan = service.plan( + state=cluster_state, + target=cluster_target, + planning_window=TIMESTEP_DURATION * T, # Full planning window in seconds + reason="Testing Cost planning", + plan_due_by_date=plan_due_by_date, + optimize_for_target=True, + max_priority_class=1, + ) + end_time = time.time() + execution_time = end_time - start_time + + # Log information + print(f"Cost-optimized plan generated in {execution_time: .2f} seconds") + + # Assert + assert cluster_plan is not None + print("Got cost-optimized cluster plan") + + if cluster_plan is None: + print("Cluster plan is None") + return + + # Get the plan for our device + energy_profile = cluster_plan.get_joule_profile() + + # Calculate the actual costs from the energy profile and tariffs + cost_elements = calculate_cost_from_energy_and_tariffs( + energy_profile.elements, cost_target_elements + ) + + print( + f"Total energy consumption: {sum(energy_profile.elements) / 3_600_000:.2f} kWh" # + ) + print(f"Total cost: ${sum(cost_elements):.2f}") + + # Plot the cost-optimized planning results with dual axes + plot_planning_results( + timestep_duration=timedelta(seconds=TIMESTEP_DURATION), + nr_of_timesteps=T, + predicted_energy_elements=energy_profile.elements, + target_energy_elements=[0] * T, # No energy target for cost optimization + cost_elements=cost_elements, + target_cost_elements=cost_target_elements, + ) + + # Basic assertion - the energy profile should have the expected number of elements + assert len(energy_profile.elements) == target_metadata.nr_of_timesteps + + print("Cost targeting test completed successfully!") + + # Main function if __name__ == "__main__": + # Test energy targeting (original functionality) test_planning_service_impl_with_ev_device() + + print("\n" + "=" * 60 + "\n") + + # Test cost targeting (new functionality) + test_planning_service_impl_with_cost_target() diff --git a/flexmeasures_s2/profile_steering/planning_service_impl.py b/flexmeasures_s2/profile_steering/planning_service_impl.py index 2f45931..823944a 100644 --- a/flexmeasures_s2/profile_steering/planning_service_impl.py +++ b/flexmeasures_s2/profile_steering/planning_service_impl.py @@ -131,10 +131,14 @@ def create_controller_tree( Returns: A RootPlanner with appropriate device planners added """ + # Always accepting all targets is NOT possible if there is an energy target + always_accept_all_proposals = not target.contains_energy_target() + root_planner = RootPlanner( target.get_global_target_profile(), self.config.energy_improvement_criterion(), self.config.cost_improvement_criterion(), + always_accept_all_proposals, self.context, ) diff --git a/flexmeasures_s2/profile_steering/root_planner.py b/flexmeasures_s2/profile_steering/root_planner.py index 29f905b..7ad1eb3 100644 --- a/flexmeasures_s2/profile_steering/root_planner.py +++ b/flexmeasures_s2/profile_steering/root_planner.py @@ -1,4 +1,4 @@ -from typing import List, Any +from typing import List, Any, Optional from datetime import datetime from flexmeasures_s2.profile_steering.common.joule_profile import JouleProfile from .congestion_point_planner import CongestionPointPlanner @@ -13,6 +13,7 @@ def __init__( target: TargetProfile, energy_iteration_criterion: float, cost_iteration_criterion: float, + always_accept_all_proposals: bool, context: Any, ): """ @@ -20,12 +21,14 @@ def __init__( It must support a method get_profile_start(), have attributes timestep_duration and nr_of_timesteps, and (for optimization purposes) support a subtract() method. + always_accept_all_proposals: If True, accept all valid proposals (for pure cost targets) context: In a full implementation, this might be an executor or similar. Here it is passed along. """ self.context = context self.target = self.remove_null_values(target) self.energy_iteration_criterion = energy_iteration_criterion self.cost_iteration_criterion = cost_iteration_criterion + self.always_accept_all_proposals = always_accept_all_proposals # Create an empty JouleProfile. # We assume that target exposes get_profile_start(), timestep_duration and nr_of_timesteps. self.empty_profile = JouleProfile( @@ -46,7 +49,7 @@ def add_congestion_point_controller(self, cpc: CongestionPointPlanner): def get_congestion_point_controller( self, cp_id: str - ) -> CongestionPointPlanner | None: + ) -> Optional[CongestionPointPlanner]: for cp in self.cp_controllers: if cp.congestion_point_id == cp_id: return cp @@ -95,6 +98,7 @@ def plan( # noqa: C901 best_proposal = None # Get proposals from each congestion point controller + proposals = [] for cpc in self.cp_controllers: print("Improving------------------------->") try: @@ -104,6 +108,7 @@ def plan( # noqa: C901 plan_due_by_date, ) if proposal is not None: + proposals.append(proposal) if best_proposal is None or proposal.is_preferred_to( best_proposal ): @@ -116,49 +121,40 @@ def plan( # noqa: C901 # No proposal could be generated; exit inner loop. break - # Update the root controller's planning based on the best proposal. - self.root_ctrl_planning = self.root_ctrl_planning.subtract( - best_proposal.old_plan - ) - self.root_ctrl_planning = self.root_ctrl_planning.add( - best_proposal.proposed_plan - ) - - # Find the real device object and explicitly update its state. - # This is crucial for parallel execution as best_proposal.origin is a copy. - cp_id_of_origin = best_proposal.origin.congestion_point_id - congestion_point_controller = self.get_congestion_point_controller( - cp_id_of_origin - ) - if congestion_point_controller is None: - raise Exception( - f"Could not find CongestionPointController with id {cp_id_of_origin}" - ) - - real_device = None - for dev in congestion_point_controller.devices: - if dev.device_id == best_proposal.origin.device_id: - real_device = dev - break - - if real_device is None: - raise Exception( - f"Could not find device with id {best_proposal.origin.device_id} in CP {cp_id_of_origin}" - ) - plan_to_accept = best_proposal.origin.get_latest_plan() - real_device.set_accepted_plan(plan_to_accept) + # Update root controller's own planning + accepted_proposals = 0 + if self.always_accept_all_proposals: + # If we're NOT optimizing for a Joule target or congestion target, we can + # accept all proposals for efficiency + accepted_proposals = len(proposals) + for proposal in proposals: + self._accept_proposal(proposal) + else: + # If we're optimizing for a Joule target or congestion target we only accept + # one proposal + accepted_proposals = 1 + self._accept_proposal(best_proposal) i += 1 print( - f"Root controller: selected best controller '{best_proposal.origin.device_name}' with global energy impr {best_proposal.get_global_improvement_value()}, congestion impr {best_proposal.get_congestion_improvement_value()}, iteration {i}." + f"Root controller: selected best controller '{best_proposal.origin.device_name}' with global energy impr {best_proposal.get_global_improvement_value()}, cost impr {best_proposal.get_cost_improvement_value()}, CP impr {best_proposal.get_congestion_improvement_value()}. Accepted {accepted_proposals} proposal(s). Iteration {i}." ) # Check stopping criteria: if improvement values are below thresholds or max iterations reached. - if ( - best_proposal.get_global_improvement_value() - <= self.energy_iteration_criterion - ) or i >= self.MAX_ITERATIONS: - break + if self.always_accept_all_proposals: + # For pure cost targets, check cost improvement + if ( + best_proposal.get_cost_improvement_value() + <= self.cost_iteration_criterion + ) or i >= self.MAX_ITERATIONS: + break + else: + # For energy targets, check energy improvement (original behavior) + if ( + best_proposal.get_global_improvement_value() + <= self.energy_iteration_criterion + ) or i >= self.MAX_ITERATIONS: + break print( f"Optimizing priority class {priority_class} was done after {i} iterations." @@ -167,3 +163,33 @@ def plan( # noqa: C901 print( f"Warning: Optimization stopped due to iteration limit. Priority class: {priority_class}, Iterations: {i}" ) + + def _accept_proposal(self, proposal): + """Accept a proposal and update the planning accordingly.""" + # Update the root controller's planning based on the proposal. + self.root_ctrl_planning = self.root_ctrl_planning.subtract(proposal.old_plan) + self.root_ctrl_planning = self.root_ctrl_planning.add(proposal.proposed_plan) + + # Find the real device object and explicitly update its state. + # This is crucial for parallel execution as proposal.origin is a copy. + cp_id_of_origin = proposal.origin.congestion_point_id + congestion_point_controller = self.get_congestion_point_controller( + cp_id_of_origin + ) + if congestion_point_controller is None: + raise Exception( + f"Could not find CongestionPointController with id {cp_id_of_origin}" + ) + + real_device = None + for dev in congestion_point_controller.devices: + if dev.device_id == proposal.origin.device_id: + real_device = dev + break + + if real_device is None: + raise Exception( + f"Could not find device with id {proposal.origin.device_id} in CP {cp_id_of_origin}" + ) + plan_to_accept = proposal.origin.get_latest_plan() + real_device.set_accepted_plan(plan_to_accept) diff --git a/flexmeasures_s2/scheduler/schedulers.py b/flexmeasures_s2/scheduler/schedulers.py index d3a708c..ec509f9 100644 --- a/flexmeasures_s2/scheduler/schedulers.py +++ b/flexmeasures_s2/scheduler/schedulers.py @@ -2,8 +2,13 @@ from datetime import datetime from typing import Any, Dict, Optional import logging +from flask import current_app as app -from flexmeasures import Scheduler +from flexmeasures import Scheduler, Sensor +from flexmeasures.data import db +from flexmeasures.data.models.planning.utils import initialize_index +from flexmeasures.data.queries.utils import simplify_index +from flexmeasures.utils.flexmeasures_inflection import pluralize # Profile steering imports from flexmeasures_s2.profile_steering.root_planner import RootPlanner @@ -128,10 +133,14 @@ def create_controller_tree( Returns: A RootPlanner with appropriate device planners added """ + # Always accepting all targets is NOT possible if there is an energy target + always_accept_all_proposals = not target.contains_energy_target() + root_planner = RootPlanner( target.get_global_target_profile(), self.config.energy_improvement_criterion(), self.config.cost_improvement_criterion(), + always_accept_all_proposals, self.context, ) @@ -150,7 +159,9 @@ def create_controller_tree( # This is a dummy congestion point. We will give it an empty profile. congestion_point_target = JouleRangeProfile( target.get_global_target_profile().metadata, - elements=congestion_point_target.elements, # type: ignore[union-attr] + elements=congestion_point_target.elements + if congestion_point_target is not None + else [], ) cpc = CongestionPointPlanner(congestion_point, congestion_point_target, self.config.multithreaded()) # type: ignore[arg-type] @@ -264,13 +275,14 @@ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.planning_service = None self.config_deserialized = False + self.frbc_device_data = None # Store FRBC device data from WebSocket def compute(self, *args, **kwargs): """ Compute the optimal energy profile using the S2 profile steering algorithm. Returns: - pandas.Series: Energy profile with timestamps as index and energy values + List: List containing FRBCInstruction objects and metadata dicts """ if not self.config_deserialized: self.deserialize_config() @@ -280,17 +292,134 @@ def compute(self, *args, **kwargs): # 1. Create ClusterState from FlexMeasures data # 2. Create ClusterTarget from flex_model # 3. Call planning_service.plan() - # 4. Convert the result to pandas.Series + # 4. Extract FRBCInstructions from device plans - raise NotImplementedError("S2Scheduler.compute() needs to be implemented") + try: + # Create planning service + config = PlanningServiceConfig() + planning_service = PlanningServiceImpl(config) - # Placeholder return - return pd.Series( - self.sensor.get_attribute("capacity_in_mw"), - index=pd.date_range( - self.start, self.end, freq=self.resolution, inclusive="left" - ), - ) + # Use FRBC device data if available, otherwise use mock data + if self.frbc_device_data is not None: + logger.info("Creating device states and cluster target from FRBC data") + device_states = self.create_device_states_from_frbc_data() + cluster_target = self.create_cluster_target_from_frbc_data() + else: + # Fallback to mock data for testing + logger.info("Using mock device states and cluster target") + device_states = self.create_mock_device_states() + cluster_target = self.create_mock_cluster_target() + + logger.info(f"Created cluster state with {len(device_states)} devices") + cluster_state = ClusterState(self.start, device_states, {}) + + # Generate plan + logger.info("Starting planning service...") + try: + cluster_plan = planning_service.plan( + state=cluster_state, + target=cluster_target, + planning_window=int((self.end - self.start).total_seconds()), + reason="S2 scheduling", + plan_due_by_date=self.start + pd.Timedelta(seconds=10), + optimize_for_target=True, + max_priority_class=1, + ) + logger.info("Planning service completed successfully") + except Exception as planning_error: + logger.error(f"Planning service failed: {planning_error}") + # Generate instructions directly without planning + if self.frbc_device_data is not None: + logger.info("Falling back to simple FRBC instruction generation") + instructions = self._generate_simple_frbc_instructions() + else: + instructions = [] + + # Create empty energy data + num_timesteps = int((self.end - self.start) / self.resolution) + energy_data = { + "sensor": self.sensor, + "data": pd.Series( + [0] * num_timesteps, + index=pd.date_range( + self.start, self.end, freq=self.resolution, inclusive="left" + ), + ), + } + return instructions + [energy_data] + + # Extract FRBCInstructions from device plans + instructions = [] + device_plans = cluster_plan.get_plan_data().get_device_plans() + + for device_plan in device_plans: + if device_plan and device_plan.instruction_profile: + instructions.extend(device_plan.instruction_profile.elements) + + # If we have FRBC device data but no instructions from planning, + # generate simple instructions based on the received data + if not instructions and self.frbc_device_data is not None: + logger.info( + "No instructions from planning service, generating simple FRBC instructions" + ) + instructions = self._generate_simple_frbc_instructions() + + # Add energy data entry for potential storage + try: + energy_profile = cluster_plan.get_joule_profile() + if energy_profile is not None and energy_profile.elements is not None: + energy_data = { + "sensor": self.sensor, + "data": pd.Series( + energy_profile.elements, + index=pd.date_range( + self.start, + self.end, + freq=self.resolution, + inclusive="left", + ), + ), + } + else: + # Create empty energy data if profile is None or has no elements + num_timesteps = int((self.end - self.start) / self.resolution) + energy_data = { + "sensor": self.sensor, + "data": pd.Series( + [0] * num_timesteps, + index=pd.date_range( + self.start, + self.end, + freq=self.resolution, + inclusive="left", + ), + ), + } + except Exception as profile_error: + logger.warning( + f"Failed to get energy profile: {profile_error}, using fallback" + ) + # Create empty energy data as fallback + num_timesteps = int((self.end - self.start) / self.resolution) + energy_data = { + "sensor": self.sensor, + "data": pd.Series( + [0] * num_timesteps, + index=pd.date_range( + self.start, self.end, freq=self.resolution, inclusive="left" + ), + ), + } + + return instructions + [energy_data] + + except Exception as e: + logger.error(f"Error in S2Scheduler.compute(): {e}") + import traceback + + logger.debug(f"Traceback: {traceback.format_exc()}") + # Return empty list on error + return [] def deserialize_config(self): """Deserialize the flex configuration from asset attributes.""" @@ -350,3 +479,329 @@ def create_cluster_target(self, target_profile: Any) -> ClusterTarget: """ # TODO: Implement conversion from flex_model to ClusterTarget raise NotImplementedError("create_cluster_target() needs to be implemented") + + def create_mock_device_states(self) -> Dict[str, Any]: + """ + Create mock device states for testing. + + Returns: + Dict[str, Any]: Dictionary of mock device states + """ + from flexmeasures_s2.profile_steering.device_planner.frbc.s2_frbc_device_state import ( + S2FrbcDeviceState, + ) + from s2python.common import PowerValue, CommodityQuantity + + # Create a simple mock device state for testing + mock_device_state = S2FrbcDeviceState( + device_id="mock_device_1", + device_name="Mock Device 1", + connection_id="mock_connection_1", + priority_class=1, + timestamp=self.start, + energy_in_current_timestep=PowerValue( + value=0, commodity_quantity=CommodityQuantity.ELECTRIC_POWER_L1 + ), + is_online=True, + power_forecast=None, + system_descriptions=[], + leakage_behaviours=[], + usage_forecasts=[], + fill_level_target_profiles=[], + computational_parameters=S2FrbcDeviceState.ComputationalParameters(100, 20), + actuator_statuses=[], + storage_status=[], + ) + + return {"mock_device_1": mock_device_state} + + def create_mock_cluster_target(self) -> ClusterTarget: + """ + Create a mock cluster target for testing. + + Returns: + ClusterTarget: Mock cluster target + """ + from flexmeasures_s2.profile_steering.common.target_profile import TargetProfile + from flexmeasures_s2.profile_steering.common.profile_metadata import ( + ProfileMetadata, + ) + + # Create mock profile metadata + profile_metadata = ProfileMetadata( + profile_start=self.start, + timestep_duration=self.resolution, + nr_of_timesteps=int((self.end - self.start) / self.resolution), + ) + + # Create a simple target profile with zeros + target_elements = [0] * profile_metadata.nr_of_timesteps + # Type ignore: TargetProfile constructor accepts List[int] and converts to JouleElement + global_target_profile = TargetProfile( + profile_start=profile_metadata.profile_start, + timestep_duration=profile_metadata.timestep_duration, + elements=target_elements, # type: ignore[arg-type] + ) + + return ClusterTarget( + generated_at=self.start, + parent_id=None, + generated_by=None, + global_target_profile=global_target_profile, + congestion_point_targets={}, + ) + + def create_device_states_from_frbc_data(self) -> Dict[str, Any]: + """ + Create device states from received FRBC data. + + Returns: + Dict[str, Any]: Dictionary of device states created from FRBC data + """ + from flexmeasures_s2.profile_steering.device_planner.frbc.s2_frbc_device_state import ( + S2FrbcDeviceState, + ) + from s2python.common import PowerValue, CommodityQuantity + + if self.frbc_device_data is None: + return self.create_mock_device_states() + + # Extract information from received FRBC data + device_id = self.frbc_device_data.resource_id or "frbc_device_1" + system_desc = self.frbc_device_data.system_description + storage_status = self.frbc_device_data.storage_status + + # Create device state using the received FRBC data + device_state = S2FrbcDeviceState( + device_id=device_id, + device_name=f"FRBC Device {device_id}", + connection_id=f"{device_id}_connection", + priority_class=1, + timestamp=self.start, + energy_in_current_timestep=PowerValue( + value=0, commodity_quantity=CommodityQuantity.ELECTRIC_POWER_L1 + ), + is_online=True, + power_forecast=None, + system_descriptions=[system_desc] if system_desc else [], + leakage_behaviours=[], + usage_forecasts=[], + fill_level_target_profiles=[self.frbc_device_data.fill_level_target_profile] + if self.frbc_device_data.fill_level_target_profile + else [], + computational_parameters=S2FrbcDeviceState.ComputationalParameters(100, 20), + actuator_statuses=[self.frbc_device_data.actuator_status] + if self.frbc_device_data.actuator_status + else [], + storage_status=[storage_status] if storage_status else [], + ) + + return {device_id: device_state} + + def create_cluster_target_from_frbc_data(self) -> ClusterTarget: + """ + Create a cluster target from FRBC device data. + + Returns: + ClusterTarget: Cluster target based on FRBC data + """ + from flexmeasures_s2.profile_steering.common.target_profile import TargetProfile + from flexmeasures_s2.profile_steering.common.profile_metadata import ( + ProfileMetadata, + ) + + if ( + self.frbc_device_data is None + or self.frbc_device_data.fill_level_target_profile is None + ): + return self.create_mock_cluster_target() + + # Create profile metadata + profile_metadata = ProfileMetadata( + profile_start=self.start, + timestep_duration=self.resolution, + nr_of_timesteps=int((self.end - self.start) / self.resolution), + ) + + # Convert FRBC fill level target to energy target + # This is a simplified conversion - in practice you'd need more sophisticated logic + target_elements = self._convert_frbc_fill_level_to_energy_target( + self.frbc_device_data.fill_level_target_profile, + profile_metadata.nr_of_timesteps, + ) + + # Type ignore: TargetProfile constructor accepts List[int] and converts to JouleElement + target_mode = app.config.get("FLEXMEASURES_S2_TARGET_MODE", "costs") + app.logger.debug(f"Target mode = {target_mode}") + if target_mode == "energy": + global_target_profile = TargetProfile( + profile_start=profile_metadata.profile_start, + timestep_duration=profile_metadata.timestep_duration, + elements=target_elements, # type: ignore[arg-type] + ) + elif target_mode == "costs": + price_sensor_id = app.config.get("FLEXMEASURES_S2_PRICE_SENSOR", 2) + price_sensor = db.session.get(Sensor, price_sensor_id) + if price_sensor is None: + logger.warning( + f"Price sensor with ID {price_sensor_id} not found, falling back to energy target mode" + ) + # Fall back to creating a simple energy target + global_target_profile = TargetProfile( + profile_start=profile_metadata.profile_start, + timestep_duration=profile_metadata.timestep_duration, + elements=target_elements, # type: ignore[arg-type] + ) + else: + assert price_sensor.unit == "EUR/MWh" + tariffs = price_sensor.search_beliefs( + event_starts_after=self.start, + event_ends_before=self.end, + resolution=self.resolution, + one_deterministic_belief_per_event=True, + ) + if ( + n_missing_prices := (self.end - self.start) // self.resolution + - len(tariffs) + ) > 0: + tariffs = simplify_index(tariffs) + tariffs = tariffs.reindex( + initialize_index( + start=self.start, end=self.end, resolution=self.resolution + ) + ) + if n_missing_prices == len(tariffs): + app.logger.warning( + f"All prices are missing in the period {self.start.isoformat()} until {self.end.isoformat()}; assuming a constant energy price of 1 EUR/MWh" + ) + tariffs = tariffs.fillna(1) + else: + app.logger.warning( + f"Forward filling {n_missing_prices} {pluralize('price', n_missing_prices)} in the period {self.start.isoformat()} until {self.end.isoformat()}" + ) + tariffs = tariffs.ffill() + global_target_profile = TargetProfile.from_tariff_values( + metadata=profile_metadata, + tariff_values=tariffs.values, + ) + else: + raise ValueError(f"Unknown FLEXMEASURES_S2_TARGET_MODE='{target_mode}'") + + return ClusterTarget( + generated_at=self.start, + parent_id=None, + generated_by=None, + global_target_profile=global_target_profile, + congestion_point_targets={}, + ) + + def _convert_frbc_fill_level_to_energy_target( + self, fill_level_profile, nr_timesteps: int + ) -> list: + """ + Convert FRBC fill level target profile to energy target profile. + + This is a simplified conversion for demonstration. + """ + # Simple conversion: aim for charging when fill level targets are higher + target_elements = [] + + if fill_level_profile and fill_level_profile.elements: + # For each target element, determine if we need to charge/discharge + current_timestep = 0 + + for element in fill_level_profile.elements: + duration_seconds = ( + getattr(element.duration, "value", 0) / 1000 + ) # Convert ms to seconds + timesteps_for_element = max( + 1, int(duration_seconds / self.resolution.total_seconds()) + ) + + # Simple logic: if target range is high, we want to charge (positive energy) + target_range_avg = ( + element.fill_level_range.start_of_range + + element.fill_level_range.end_of_range + ) / 2 + + if target_range_avg > 50: # Above 50% - charging target + energy_target = 10000 # 10kJ per timestep + else: # Below 50% - minimal energy + energy_target = 1000 # 1kJ per timestep + + # Add target for this duration + for _ in range( + min(timesteps_for_element, nr_timesteps - current_timestep) + ): + target_elements.append(energy_target) + current_timestep += 1 + + if current_timestep >= nr_timesteps: + break + + # Fill remaining timesteps with zeros + while len(target_elements) < nr_timesteps: + target_elements.append(0) + + return target_elements[:nr_timesteps] + + def _generate_simple_frbc_instructions(self) -> list: + """ + Generate simple FRBC instructions based on received device data. + + Returns: + List of FRBCInstruction objects + """ + from s2python.frbc import FRBCInstruction + from datetime import timezone + import uuid + + if ( + self.frbc_device_data is None + or self.frbc_device_data.system_description is None + ): + return [] + + instructions = [] + system_desc = self.frbc_device_data.system_description + fill_level_target = self.frbc_device_data.fill_level_target_profile + + # Extract actuator and operation mode information + if system_desc.actuators: + actuator = system_desc.actuators[0] # Use first actuator + if actuator.operation_modes: + operation_mode = actuator.operation_modes[0] # Use first operation mode + + # Determine operation mode factor based on fill level targets + operation_mode_factor = 0.5 # Default + + if fill_level_target and fill_level_target.elements: + # Simple logic: higher target ranges get higher factors + for element in fill_level_target.elements: + target_avg = ( + element.fill_level_range.start_of_range + + element.fill_level_range.end_of_range + ) / 2 + if target_avg > 50: + operation_mode_factor = 0.8 # Higher factor for charging + else: + operation_mode_factor = 0.2 # Lower factor for maintaining + break # Use first element for simplicity + + # Generate instruction for near future + execution_time = self.start.replace(tzinfo=timezone.utc) + + instruction = FRBCInstruction( + message_id=str(uuid.uuid4()), + id=str(uuid.uuid4()), + actuator_id=str(actuator.id), + operation_mode=str(operation_mode.id), + operation_mode_factor=operation_mode_factor, + execution_time=execution_time, + abnormal_condition=False, + ) + + instructions.append(instruction) + logger.info(f"Generated FRBC instruction: {instruction.to_json()}") + + return instructions diff --git a/requirements/app.in b/requirements/app.in index dd57aa2..b4916df 100644 --- a/requirements/app.in +++ b/requirements/app.in @@ -1,4 +1,4 @@ flexmeasures>=0.27.0 s2-python cryptography>=44.0.2 -pydantic<2.0.0 +inflect>=7.5.0 diff --git a/run_mypy.sh b/run_mypy.sh index f7b3c44..52258c9 100755 --- a/run_mypy.sh +++ b/run_mypy.sh @@ -2,5 +2,5 @@ set -e pip install mypy pip install types-pytz types-requests types-Flask types-click types-redis types-tzlocal types-python-dateutil types-setuptools -files=$(find . -name \*.py -not -path "./venv/*" -not -path ".egg/*" -not -path "./build/*" -not -path "./dist/*") +files=$(find . -name \*.py -not -path "./venv/*" -not -path "./.eggs/*" -not -path "./build/*" -not -path "./dist/*" -not -path "./.mypy_cache/*") mypy --follow-imports skip --ignore-missing-imports $files diff --git a/setup.cfg b/setup.cfg index b0f88ac..291efbb 100644 --- a/setup.cfg +++ b/setup.cfg @@ -7,5 +7,5 @@ exclude = .git,__pycache__,documentation max-line-length = 160 max-complexity = 13 select = B,C,E,F,W,B9 -ignore = E501, W503, E203, C901 +ignore = E501, W503, E203, C901, E231