Add universal yaw controller for estimation and wake steering

hycon/controllers/__init__.py

@@ -10,6 +10,7 @@
 from hycon.controllers.hydrogen_plant_controller import HydrogenPlantController
 from hycon.controllers.lookup_based_wake_steering_controller import (
     LookupBasedWakeSteeringController,
+    YawSetpointPassthroughController,
 )
 from hycon.controllers.solar_passthrough_controller import SolarPassthroughController
 from hycon.controllers.wake_steering_rosco_standin import WakeSteeringROSCOStandin

hycon/controllers/lookup_based_wake_steering_controller.py

@@ -113,3 +113,17 @@ def wake_steering_angles(self, wind_directions):
         self.controls_dict = {"yaw_angles": yaw_setpoint}
 
         return {"yaw_angles": yaw_setpoint}
+
+
+class YawSetpointPassthroughController(ControllerBase):
+    """
+    YawSetpointPassthroughController is a simple controller that passes through wind directions
+    as yaw setpoints without modification.
+    """
+
+    def __init__(self, interface: InterfaceBase, verbose: bool = False):
+        super().__init__(interface, verbose=verbose)
+
+    def compute_controls(self, measurements_dict):
+        # Simply pass through the yaw setpoints as the received wind directions
+        return {"yaw_angles": measurements_dict["wind_farm"]["wind_directions"]}

hycon/controllers/wind_farm_yaw_controller.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from hycon.controllers import YawSetpointPassthroughController
+from hycon.controllers.controller_base import ControllerBase
+from hycon.estimators import WindDirectionPassthroughEstimator
+from hycon.estimators.estimator_base import EstimatorBase
+from hycon.interfaces.interface_base import InterfaceBase
+
+
+class WindFarmYawController(ControllerBase):
+    """
+    WindFarmYawController is a top-level controller that manages a combined wind estimator
+    and yaw setpoint controller for a wind farm.
+    """
+
+    def __init__(
+        self,
+        interface: InterfaceBase,
+        yaw_setpoint_controller: ControllerBase | None = None,
+        wind_estimator: EstimatorBase | None = None,
+        verbose: bool = False,
+    ):
+        """
+        Constructor for WindFarmYawController.
+
+        Args:
+            interface (InterfaceBase): Interface object for communicating with the plant.
+            input_dict (dict): Optional dictionary of input parameters.
+            controller_parameters (dict): Optional dictionary of controller parameters.
+            yaw_setpoint_controller (ControllerBase): Optional yaw controller to set control
+                setpoints of individual wind turbines.
+            wind_estimator (EstimatorBase): Optional wind estimator to provide wind direction
+                estimates for individual turbines.
+        """
+        super().__init__(interface, verbose=verbose)
+
+        # Establish defaults for yaw setpoint controller and wind estimator and store on self
+        if yaw_setpoint_controller is None:
+            yaw_setpoint_controller = YawSetpointPassthroughController(interface, verbose=verbose)
+        if wind_estimator is None:
+            wind_estimator = WindDirectionPassthroughEstimator(interface, verbose=verbose)
+        self.yaw_setpoint_controller = yaw_setpoint_controller
+        self.wind_estimator = wind_estimator
+
+    def compute_controls(self, measurements_dict):
+        estimates_dict = self.wind_estimator.compute_estimates(measurements_dict)
+        controls_dict = self.yaw_setpoint_controller.compute_controls(estimates_dict)
+
+        return controls_dict

hycon/estimators/__init__.py

@@ -0,0 +1 @@
+from hycon.estimators.wind_direction_estimator import WindDirectionPassthroughEstimator

hycon/estimators/estimator_base.py

@@ -0,0 +1,39 @@
+from abc import ABCMeta, abstractmethod
+
+
+class EstimatorBase(metaclass=ABCMeta):
+    def __init__(self, interface, verbose=True):
+        self._s = interface
+        self.verbose = verbose
+
+        self._measurements_dict = {}
+        self._estimates_dict = {}
+
+    def _receive_measurements(self, input_dict=None):
+        self._measurements_dict = self._s.get_measurements(input_dict)
+        return None
+
+    def _send_estimates(self, input_dict=None):
+        # TODO: how should I use this? Could just return an empty dict for estimates, presumably.
+        # self._s.check_estimates(self._estimates_dict) # _s.check_controls?
+        # output_dict = self._s.send_estimates(input_dict, **self._estimates_dict) # send_controls?
+
+        return input_dict  # output_dict # Or simply return input_dict? May work.
+
+    def step(self, input_dict=None):
+        """
+        Only used if an estimator alone is being run (without an overarching controller).
+        """
+        self._receive_measurements(input_dict)
+
+        self._estimates_dict = self.compute_estimates(self._measurements_dict)
+
+        output_dict = self._send_estimates(input_dict)
+
+        return output_dict
+
+    @abstractmethod
+    def compute_estimates(self, measurements_dict: dict) -> dict:
+        raise NotImplementedError(
+            "compute_estimates method must be implemented in the child class of EstimatorBase."
+        )

hycon/estimators/wind_direction_estimator.py

@@ -0,0 +1,16 @@
+from hycon.estimators.estimator_base import EstimatorBase
+from hycon.interfaces.interface_base import InterfaceBase
+
+
+class WindDirectionPassthroughEstimator(EstimatorBase):
+    """
+    WindDirectionPassthroughEstimator is a simple estimator that passes through the wind
+    direction measurements without modification.
+    """
+
+    def __init__(self, interface: InterfaceBase, verbose: bool = False):
+        super().__init__(interface, verbose=verbose)
+
+    def compute_estimates(self, measurements_dict):
+        # Simply pass through the wind directions as estimates
+        return {"wind_directions": measurements_dict["wind_farm"]["wind_directions"]}

tests/controller_library_test.py

@@ -13,6 +13,7 @@
     SolarPassthroughController,
     WindFarmPowerDistributingController,
     WindFarmPowerTrackingController,
+    YawSetpointPassthroughController,
 )
 from hycon.controllers.wind_farm_power_tracking_controller import POWER_SETPOINT_DEFAULT
 from hycon.interfaces import (
@@ -46,6 +47,7 @@ def test_controller_instantiation(test_interface_standin, test_hercules_v1_dict)
         interface=test_interface_standin, input_dict=test_hercules_v1_dict
     )
     _ = BatteryController(interface=test_interface_standin, input_dict=test_hercules_v1_dict)
+    _ = YawSetpointPassthroughController(interface=test_interface_standin)
 
 
 def test_LookupBasedWakeSteeringController(test_hercules_v1_dict, test_interface_hercules_ad):
@@ -682,3 +684,22 @@ def test_HydrogenPlantController(test_hercules_v1_dict, test_interface_hercules_
         generator_controller=hybrid_controller,
         controller_parameters=external_controller_parameters,
     )
+
+
+def test_YawSetpointPassthroughController(test_hercules_v1_dict, test_interface_hercules_ad):
+    """
+    Tests that the YawSetpointPassthroughController simply passes through the yaw setpoints
+    from the interface.
+    """
+    test_controller = YawSetpointPassthroughController(
+        test_interface_hercules_ad, test_hercules_v1_dict
+    )
+
+    # Check that the controller can be stepped
+    test_hercules_v1_dict["time"] = 20
+    test_dict_out = test_controller.step(input_dict=test_hercules_v1_dict)
+
+    assert np.allclose(
+        test_dict_out["hercules_comms"]["amr_wind"]["test_farm"]["turbine_yaw_angles"],
+        test_hercules_v1_dict["hercules_comms"]["amr_wind"]["test_farm"]["turbine_wind_directions"],
+    )

tests/estimator_base_test.py

@@ -0,0 +1,44 @@
+import pytest
+from hycon.estimators.estimator_base import EstimatorBase
+
+
+class InheritanceTestClassBad(EstimatorBase):
+    """
+    Class that is missing necessary methods (compute_estimates).
+    """
+
+    def __init__(self, interface):
+        super().__init__(interface)
+
+
+class InheritanceTestClassGood(EstimatorBase):
+    """
+    Class that is missing necessary methods.
+    """
+
+    def __init__(self, interface):
+        super().__init__(interface)
+
+    def compute_estimates(self):
+        pass
+
+
+def test_EstimatorBase_methods(test_interface_standin):
+    """
+    Check that the base interface class establishes the correct methods.
+    """
+    estimator_base = InheritanceTestClassGood(test_interface_standin)
+    assert hasattr(estimator_base, "_receive_measurements")
+    # assert hasattr(estimator_base, "_send_estimates") # Not yet sure we want this.
+    assert hasattr(estimator_base, "step")
+    assert hasattr(estimator_base, "compute_estimates")
+
+
+def test_inherited_methods(test_interface_standin):
+    """
+    Check that a subclass of InterfaceBase inherits methods correctly.
+    """
+    with pytest.raises(TypeError):
+        _ = InheritanceTestClassBad(test_interface_standin)
+
+    _ = InheritanceTestClassGood(test_interface_standin)

tests/estimator_library_test.py

@@ -0,0 +1,39 @@
+import numpy as np
+from hycon.estimators import WindDirectionPassthroughEstimator
+
+
+def test_estimator_instantiation(test_interface_standin):
+    """
+    Tests whether all controllers can be imported correctly and that they
+    each implement the required methods specified by ControllerBase.
+    """
+    _ = WindDirectionPassthroughEstimator(interface=test_interface_standin)
+
+
+def test_WindDirectionPassthroughEstimator(test_interface_hercules_ad, test_hercules_v1_dict):
+    """
+    Tests that the YawSetpointPassthroughController simply passes through the yaw setpoints
+    from the interface.
+    """
+    test_estimator = WindDirectionPassthroughEstimator(
+        test_interface_hercules_ad, test_hercules_v1_dict
+    )
+
+    # Check that the controller can be stepped (simply returns inputs)
+    test_hercules_v1_dict["time"] = 20
+    test_hercules_dict_out = test_estimator.step(input_dict=test_hercules_v1_dict)
+
+    assert np.allclose(
+        test_hercules_dict_out["hercules_comms"]["amr_wind"]["test_farm"][
+            "turbine_wind_directions"
+        ],
+        test_hercules_v1_dict["hercules_comms"]["amr_wind"]["test_farm"]["turbine_wind_directions"],
+    )
+
+    # Test that estimates are also computed (for passthrough, these are simply a match)
+    estimates_dict = test_estimator.compute_estimates(test_estimator._measurements_dict)
+
+    assert np.allclose(
+        estimates_dict["wind_directions"],
+        test_estimator._measurements_dict["wind_farm"]["wind_directions"],
+    )