123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112 |
- # -*- coding: utf-8 -*-
- from typing import Dict
- import numpy as np
- from httpx import AsyncClient
- from loguru import logger
- from app.controllers.equipment.controller import EquipmentController
- from app.schemas.equipment import AirValveSpeed, FCU
- from app.schemas.space import Space
- from app.services.platform import DataPlatformService
- from app.services.transfer import SpaceInfoService, Duoduo, Season
- from app.utils.math import round_half_up
- class FCUController(EquipmentController):
- def __init__(self, equipment: FCU, season: Season):
- super(FCUController, self).__init__()
- self._equipment = equipment
- self.season = season
- async def get_temperature_target(self) -> float:
- target = self._equipment.space.temperature_target
- if not np.isnan(target):
- setting_target = target
- else:
- setting_target = 25.0
- self._equipment.setting_temperature = setting_target
- return setting_target
- async def get_air_valve_speed(self):
- temperature_target = await self.get_temperature_target()
- if temperature_target > 0:
- diff = self._equipment.space.realtime_temperature - temperature_target
- if diff >= 1.0:
- self._equipment.air_valve_speed = AirValveSpeed.high
- elif diff >= 0.5:
- self._equipment.air_valve_speed = AirValveSpeed.medium
- elif diff >= 0.0:
- self._equipment.air_valve_speed = AirValveSpeed.low
- else:
- self._equipment.air_valve_speed = AirValveSpeed.off
- else:
- self._equipment.air_valve_speed = AirValveSpeed.off
- async def get_running_status(self):
- if self._equipment.air_valve_speed == AirValveSpeed.off:
- self._equipment.running_status = False
- else:
- self._equipment.running_status = True
- async def get_mode(self) -> None:
- if self.season == Season.heating:
- mode = 2
- elif self.season == Season.cooling:
- mode = 1
- else:
- mode = 0
- self._equipment.work_mode = mode
- async def run(self):
- await self.get_temperature_target()
- await self.get_air_valve_speed()
- await self.get_running_status()
- await self.get_mode()
- def get_results(self):
- return self._equipment
- @logger.catch()
- async def get_fcu_control_result(project_id: str, equipment_id: str) -> Dict:
- async with AsyncClient() as client:
- duo_duo = Duoduo(client, project_id)
- platform = DataPlatformService(client, project_id)
- spaces = await duo_duo.get_space_by_equipment(equipment_id)
- for sp in spaces:
- transfer = SpaceInfoService(client, project_id, sp.get('id'))
- current_target = await transfer.get_current_temperature_target()
- realtime_temperature = await platform.get_realtime_temperature(sp.get('id'))
- season = await transfer.get_season()
- temp_space_params = {
- 'id': sp.get('id'),
- 'temperature_target': current_target,
- 'realtime_temperature': realtime_temperature
- }
- space = Space(**temp_space_params)
- break
- fcu_temp_params = {
- 'id': equipment_id,
- 'space': space
- }
- fcu = FCU(**fcu_temp_params)
- fcu_controller = FCUController(fcu, season)
- await fcu_controller.run()
- regulated_fcu = fcu_controller.get_results()
- output = {
- 'onOff': 1 if regulated_fcu.running_status else 0,
- 'mode': regulated_fcu.work_mode,
- 'speed': regulated_fcu.air_valve_speed.value,
- 'temperature': int(round_half_up(regulated_fcu.setting_temperature, 1)),
- 'water': 1 if regulated_fcu.running_status else 0
- }
- logger.debug(output)
- return output
|