|
@@ -1,6 +1,6 @@
|
|
# -*- coding: utf-8 -*-
|
|
# -*- coding: utf-8 -*-
|
|
|
|
|
|
-from typing import Dict
|
|
|
|
|
|
+from typing import Dict, Tuple
|
|
|
|
|
|
import numpy as np
|
|
import numpy as np
|
|
from httpx import AsyncClient
|
|
from httpx import AsyncClient
|
|
@@ -38,10 +38,13 @@ class FCUController(EquipmentController):
|
|
async def get_mode(self) -> int:
|
|
async def get_mode(self) -> int:
|
|
temperature_target = self._equipment.space.temperature_target
|
|
temperature_target = self._equipment.space.temperature_target
|
|
if self.season == Season.heating:
|
|
if self.season == Season.heating:
|
|
- if temperature_target >= self._equipment.space.realtime_temperature:
|
|
|
|
- mode = 2
|
|
|
|
- else:
|
|
|
|
- mode = 1
|
|
|
|
|
|
+ mode = 2
|
|
|
|
+ if (
|
|
|
|
+ temperature_target > 0
|
|
|
|
+ and self._equipment.space.realtime_temperature > 0
|
|
|
|
+ ):
|
|
|
|
+ if temperature_target < self._equipment.space.realtime_temperature:
|
|
|
|
+ mode = 1
|
|
elif self.season == Season.cooling:
|
|
elif self.season == Season.cooling:
|
|
mode = 1
|
|
mode = 1
|
|
else:
|
|
else:
|
|
@@ -179,6 +182,124 @@ class FCUControllerV2(EquipmentController):
|
|
return self.equipment
|
|
return self.equipment
|
|
|
|
|
|
|
|
|
|
|
|
+class ATFC2Controller:
|
|
|
|
+ """Two-wipe ATFC logic"""
|
|
|
|
+
|
|
|
|
+ def __init__(self, space_target: float, space_realtime: float, season: Season):
|
|
|
|
+ self.space_target = space_target
|
|
|
|
+ self.space_realtime = space_realtime
|
|
|
|
+ self.season = season
|
|
|
|
+
|
|
|
|
+ def build_speed_set(self) -> str:
|
|
|
|
+ if self.space_realtime > 0.0 and self.space_target > 0.0:
|
|
|
|
+ if self.season == Season.cooling:
|
|
|
|
+ diff = self.space_realtime - self.space_target
|
|
|
|
+ elif self.season == Season.heating:
|
|
|
|
+ diff = self.space_target - self.space_realtime
|
|
|
|
+ else:
|
|
|
|
+ diff = 0.0
|
|
|
|
+
|
|
|
|
+ if diff <= 0.0:
|
|
|
|
+ speed = "off"
|
|
|
|
+ elif diff < 0.5:
|
|
|
|
+ speed = "low"
|
|
|
|
+ elif diff < 1.5:
|
|
|
|
+ speed = "medium"
|
|
|
|
+ else:
|
|
|
|
+ speed = "high"
|
|
|
|
+ else:
|
|
|
|
+ speed = "hold"
|
|
|
|
+ if self.space_target < 0.0:
|
|
|
|
+ speed = "off"
|
|
|
|
+
|
|
|
|
+ return speed
|
|
|
|
+
|
|
|
|
+ def build_switch_set(self, new_speed_set: str) -> str:
|
|
|
|
+ if new_speed_set == "off":
|
|
|
|
+ switch_set = "off"
|
|
|
|
+ else:
|
|
|
|
+ switch_set = "on"
|
|
|
|
+
|
|
|
|
+ return switch_set
|
|
|
|
+
|
|
|
|
+ def build_water_valve_set(self, new_switch_set: str) -> str:
|
|
|
|
+ return new_switch_set
|
|
|
|
+
|
|
|
|
+ def run(self) -> dict:
|
|
|
|
+ speed = self.build_speed_set()
|
|
|
|
+ switch_set = self.build_switch_set(speed)
|
|
|
|
+ water_switch_set = self.build_water_valve_set(switch_set)
|
|
|
|
+
|
|
|
|
+ res = {
|
|
|
|
+ "speed_set": speed,
|
|
|
|
+ "switch_set": switch_set,
|
|
|
|
+ "water_valve_set": water_switch_set,
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ return res
|
|
|
|
+
|
|
|
|
+
|
|
|
|
+class ATFC4Controller(ATFC2Controller):
|
|
|
|
+ """Four-wipe ATFC logic"""
|
|
|
|
+
|
|
|
|
+ def __init__(self, space_target: float, space_realtime: float, season: Season):
|
|
|
|
+ super().__init__(space_target, space_realtime, season)
|
|
|
|
+
|
|
|
|
+ def build_speed_set(self) -> str:
|
|
|
|
+ speed = super().build_speed_set()
|
|
|
|
+ if self.season == Season.heating:
|
|
|
|
+ if self.space_target > 0.0 and self.space_realtime > 0.0:
|
|
|
|
+ diff = abs(self.space_realtime - self.space_target)
|
|
|
|
+ if diff < 0.2:
|
|
|
|
+ speed = "off"
|
|
|
|
+ elif diff < 0.8:
|
|
|
|
+ speed = "low"
|
|
|
|
+ elif diff < 1.5:
|
|
|
|
+ speed = "medium"
|
|
|
|
+ else:
|
|
|
|
+ speed = "high"
|
|
|
|
+
|
|
|
|
+ return speed
|
|
|
|
+
|
|
|
|
+ def build_water_valve_set(self, new_switch_set: str) -> Tuple[str, str]:
|
|
|
|
+ if new_switch_set == "off":
|
|
|
|
+ chill_water_valve_set = "off"
|
|
|
|
+ hot_water_valve_set = "off"
|
|
|
|
+ else:
|
|
|
|
+ if self.season == Season.heating:
|
|
|
|
+ if self.space_realtime > 0.0:
|
|
|
|
+ if self.space_realtime > self.space_target:
|
|
|
|
+ chill_water_valve_set = "on"
|
|
|
|
+ hot_water_valve_set = "off"
|
|
|
|
+ else:
|
|
|
|
+ chill_water_valve_set = "off"
|
|
|
|
+ hot_water_valve_set = "on"
|
|
|
|
+ else:
|
|
|
|
+ chill_water_valve_set = "hold"
|
|
|
|
+ hot_water_valve_set = "hold"
|
|
|
|
+ else:
|
|
|
|
+ chill_water_valve_set = "on"
|
|
|
|
+ hot_water_valve_set = "off"
|
|
|
|
+
|
|
|
|
+ return chill_water_valve_set, hot_water_valve_set
|
|
|
|
+
|
|
|
|
+ def run(self) -> dict:
|
|
|
|
+ speed = self.build_speed_set()
|
|
|
|
+ switch_set = self.build_switch_set(speed)
|
|
|
|
+ chill_water_valve_set, hot_water_valve_set = self.build_water_valve_set(
|
|
|
|
+ switch_set
|
|
|
|
+ )
|
|
|
|
+
|
|
|
|
+ res = {
|
|
|
|
+ "speed_set": speed,
|
|
|
|
+ "switch_set": switch_set,
|
|
|
|
+ "chill_water_valve_set": chill_water_valve_set,
|
|
|
|
+ "hot_water_valve_set": hot_water_valve_set,
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ return res
|
|
|
|
+
|
|
|
|
+
|
|
@logger.catch()
|
|
@logger.catch()
|
|
async def get_fcu_control_result(project_id: str, equipment_id: str) -> Dict:
|
|
async def get_fcu_control_result(project_id: str, equipment_id: str) -> Dict:
|
|
async with AsyncClient() as client:
|
|
async with AsyncClient() as client:
|
|
@@ -282,3 +403,29 @@ async def build_acatfc4_instructions(
|
|
)
|
|
)
|
|
|
|
|
|
return instructions
|
|
return instructions
|
|
|
|
+
|
|
|
|
+
|
|
|
|
+@logger.catch()
|
|
|
|
+async def build_acatfc2_instructions_v2(params: ACATFC2InstructionsRequest) -> dict:
|
|
|
|
+ space_realtime = (
|
|
|
|
+ params.space_realtime_temperature if params.space_realtime_temperature else -1
|
|
|
|
+ )
|
|
|
|
+ controller = ATFC2Controller(
|
|
|
|
+ params.space_temperature_target, space_realtime, Season(params.season)
|
|
|
|
+ )
|
|
|
|
+ instructions = controller.run()
|
|
|
|
+
|
|
|
|
+ return instructions
|
|
|
|
+
|
|
|
|
+
|
|
|
|
+@logger.catch()
|
|
|
|
+async def build_acatfc4_instructions_v2(params: ACATFC4InstructionsRequest) -> dict:
|
|
|
|
+ space_realtime = (
|
|
|
|
+ params.space_realtime_temperature if params.space_realtime_temperature else -1
|
|
|
|
+ )
|
|
|
|
+ controller = ATFC4Controller(
|
|
|
|
+ params.space_temperature_target, space_realtime, Season(params.season)
|
|
|
|
+ )
|
|
|
|
+ instructions = controller.run()
|
|
|
|
+
|
|
|
|
+ return instructions
|