Browse Source

add a new logic for ACATFU's freq set

highing666 3 năm trước cách đây
mục cha
commit
18308a9e53

+ 1 - 1
app/api/routers/devices.py

@@ -137,7 +137,7 @@ async def get_acatfu_supply_air_temperature_set_v2(
     "/instructions/acatfu/freq-set", response_model=domain_devices.ACATFUFreqSetResponse
 )
 async def get_acatfu_freq_set(params: domain_devices.ACATFUFreqSetRequest):
-    freq_set = build_acatfu_freq_set(params)
+    freq_set = await build_acatfu_freq_set(params)
 
     logger.info(f"{params.device_id} - {freq_set}")
 

+ 170 - 23
app/controllers/equipment/pau/freq_set.py

@@ -1,13 +1,14 @@
 # -*- coding: utf-8 -*-
 
-from typing import Tuple
+from typing import List, Tuple
 
-from httpx import AsyncClient
 from loguru import logger
+import numpy as np
 
 from app.models.domain.devices import ACATFUFreqSetRequest
-from app.services.platform import DataPlatformService, InfoCode
-from app.services.transfer import Duoduo
+from app.schemas.season import Season
+from app.schemas.space import SpaceATFU
+from app.utils.helpers import is_off_to_on
 
 
 class ACATFUFanFreqController:
@@ -45,31 +46,177 @@ class ACATFUFanFreqController:
         return next_freq_set
 
 
-async def fetch_params(project_id: str, device_id: str) -> Tuple[float, float, float]:
-    async with AsyncClient() as client:
-        platform = DataPlatformService(client, project_id)
-        duoduo = Duoduo(client, project_id)
-
-        current_freq = await platform.get_realtime_data(InfoCode.fan_freq, device_id)
-        hot_rate, cold_rate = await duoduo.query_fill_rate_by_device(device_id)
-
-        return current_freq, hot_rate, cold_rate
+class ACATFUFanFreqControllerV2:
+    """
+    Logic writen by Wuxu.
+    """
 
+    def __init__(
+        self,
+        freq: float,
+        fresh_air_temperature: float,
+        season: Season,
+    ) -> None:
+        self._freq = freq
+        self._fresh_air_temperature = fresh_air_temperature
+        self._season = season
+
+    def get_next_set(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
+        if self._season == Season.transition:
+            next_freq_set = self.get_transition_logic(spaces_params, on_flag)
+        elif self._season == Season.cooling:
+            next_freq_set = self.get_cooling_logic(spaces_params, on_flag)
+        else:
+            next_freq_set = self.get_heating_logic(spaces_params, on_flag)
+
+        next_freq_set = max(20.0, next_freq_set)
+        next_freq_set = min(50.0, next_freq_set)
 
-@logger.catch()
-async def get_next_acatfu_freq_set(project_id: str, device_id: str) -> float:
-    current_freq, hot_rate, cold_rate = await fetch_params(project_id, device_id)
-    controller = ACATFUFanFreqController(current_freq, hot_rate, cold_rate)
-    next_freq_set = controller.get_next_set()
+        return next_freq_set
 
-    return next_freq_set
+    def get_transition_logic(
+        self, spaces_params: List[SpaceATFU], on_flag: bool
+    ) -> float:
+        temp_avg, co2_avg = self.get_avg(spaces_params)
+        if on_flag:
+            if self._fresh_air_temperature <= 16.0:
+                freq_set = 20.0
+            elif self._fresh_air_temperature <= 18.0:
+                if temp_avg <= 23.0:
+                    freq_set = 20.0
+                else:
+                    freq_set = 25.0
+            elif self._fresh_air_temperature <= 21.0:
+                if temp_avg <= 23.0:
+                    freq_set = 25.0
+                elif temp_avg <= 25.5:
+                    freq_set = 30.0
+                else:
+                    freq_set = 35.0
+            elif self._fresh_air_temperature <= 24.0:
+                if temp_avg <= 23.0:
+                    freq_set = 30.0
+                elif temp_avg <= 25.5:
+                    freq_set = 35.0
+                else:
+                    freq_set = 40.0
+            else:
+                if temp_avg <= 23.0:
+                    freq_set = 35.0
+                elif temp_avg <= 25.5:
+                    freq_set = 30.0
+                else:
+                    freq_set = 25.0
+
+            if co2_avg > 750.0:
+                freq_set = max(freq_set, 25.0)
+        else:
+            if self._fresh_air_temperature <= 18.0:
+                if temp_avg <= 23.0:
+                    freq_set = 20.0
+                elif temp_avg <= 25.5:
+                    freq_set = self._freq - 2.0
+                else:
+                    freq_set = self._freq + 3.0
+            elif self._fresh_air_temperature <= 23.0:
+                if temp_avg <= 23.0:
+                    freq_set = self._freq - 2.0
+                elif temp_avg <= 25.5:
+                    freq_set = self._freq
+                else:
+                    freq_set = self._freq + 2.0
+            elif self._fresh_air_temperature <= 25.0:
+                if temp_avg <= 23.0:
+                    freq_set = self._freq + 2.0
+                elif temp_avg <= 25.5:
+                    freq_set = self._freq
+                else:
+                    freq_set = self._freq + 1.0
+            else:
+                if temp_avg <= 23.0:
+                    freq_set = self._freq + 5.0
+                elif temp_avg <= 25.5:
+                    freq_set = self._freq - 2.0
+                else:
+                    freq_set = 20.0
+
+        return freq_set
+
+    def get_cooling_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
+        _, co2_avg = self.get_avg(spaces_params)
+        if on_flag:
+            freq_set = 30.0
+        else:
+            high_co2_ratio = self.get_high_co2_ratio(spaces_params)
+            if high_co2_ratio == 0.0:
+                if co2_avg < 600.0:
+                    freq_set = self._freq - 2.0
+                else:
+                    freq_set = self._freq
+                freq_set = max(25.0, freq_set)
+            elif high_co2_ratio < 0.05:
+                freq_set = self._freq + 2.0
+                freq_set = min(35.0, freq_set)
+            elif high_co2_ratio < 0.1:
+                freq_set = self._freq + 3.0
+                freq_set = min(40.0, freq_set)
+            elif high_co2_ratio < 0.2:
+                freq_set = self._freq + 5.0
+            else:
+                freq_set = 50.0
+
+        return freq_set
+
+    def get_heating_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
+
+        return self._freq
+
+    @staticmethod
+    def get_avg(spaces_params: List[SpaceATFU]) -> Tuple[float, float]:
+        valid_temperature_list, valid_co2_list = list(), list()
+        for space in spaces_params:
+            if 0.0 < space.realtime_temperature < 40.0:
+                valid_temperature_list.append(space.realtime_temperature)
+            if 0.0 < space.realtime_co2 < 5000.0:
+                valid_co2_list.append(space.realtime_co2)
+
+        if valid_temperature_list:
+            temp_avg = np.mean(valid_temperature_list)
+        else:
+            temp_avg = np.NAN
+
+        if valid_co2_list:
+            co2_avg = np.mean(valid_co2_list)
+        else:
+            co2_avg = np.NAN
+
+        return temp_avg, co2_avg
+
+    @staticmethod
+    def get_high_co2_ratio(spaces_params: List[SpaceATFU]) -> float:
+        valid_co2_count, high_co2_count = 0, 0
+        for space in spaces_params:
+            if 0.0 < space.realtime_co2 < 5000.0:
+                valid_co2_count += 1
+                if space.realtime_co2 > 900.0:
+                    high_co2_count += 1
+
+        if valid_co2_count == 0:
+            ratio = 0.0
+        else:
+            ratio = high_co2_count / valid_co2_count
+
+        return ratio
 
 
 @logger.catch()
-def build_acatfu_freq_set(params: ACATFUFreqSetRequest) -> float:
-    controller = ACATFUFanFreqController(
-        params.freq, params.hot_ratio, params.cold_ratio
+async def build_acatfu_freq_set(params: ACATFUFreqSetRequest) -> float:
+    controller = ACATFUFanFreqControllerV2(
+        params.freq,
+        params.fresh_air_temperature,
+        params.season,
     )
-    freq_set = controller.get_next_set()
+    on_flag = is_off_to_on(params.running_status_list)
+    freq_set = controller.get_next_set(params.spaces, on_flag)
 
     return freq_set

+ 7 - 8
app/models/domain/devices.py

@@ -3,7 +3,10 @@ from typing import Dict, List, Optional
 
 from pydantic import BaseModel, Field
 
+from app.controllers.equipment.switch import SwitchSet
 from app.models.domain.feedback import FeedbackValue
+from app.schemas.season import Season
+from app.schemas.space import SpaceATFU
 
 
 class ThermalMode(str, Enum):
@@ -112,12 +115,6 @@ class ACATAHFreqSetResponse(BaseModel):
     freq_set: float
 
 
-class SwitchSet(str, Enum):
-    on = "on"
-    off = "off"
-    hold = "hold"
-
-
 class SwitchSetRequestBase(BaseModel):
     device_id: str
     running_status: bool
@@ -208,8 +205,10 @@ class ACATFUSupplyAirTempSetResponse(BaseModel):
 class ACATFUFreqSetRequest(BaseModel):
     device_id: str
     freq: float
-    hot_ratio: float
-    cold_ratio: float
+    fresh_air_temperature: float
+    spaces: List[SpaceATFU]
+    season: Season
+    running_status_list: List[float]
 
 
 class ACATFUFreqSetResponse(BaseModel):

+ 12 - 0
app/utils/helpers.py

@@ -0,0 +1,12 @@
+from typing import List
+
+
+def is_off_to_on(running_status_list: List[float]) -> bool:
+    flag = False
+    if running_status_list:
+        if running_status_list[-1] == 1.0:
+            for item in running_status_list:
+                if item == 0.0:
+                    flag = True
+    
+    return flag