# -*- coding: utf-8 -*- import numpy as np from httpx import AsyncClient from loguru import logger from app.controllers.equipment.controller import EquipmentController from app.models.equipment import AirValveSpeed from app.models.equipment import FCU from app.models.space import Space from app.services.transfer import SpaceInfoService, EquipmentInfoService class FCUController(EquipmentController): def __init__(self, equipment: FCU): super(FCUController, self).__init__() self._equipment = equipment async def get_temperature_target(self) -> float: target = self._equipment.space.temperature_target if target is not np.NAN and target > 0: 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 and temperature_target > 0: self._equipment.air_valve_speed = AirValveSpeed.auto else: self._equipment.air_valve_speed = AirValveSpeed.off async def run(self): await self.get_temperature_target() await self.get_air_valve_speed() self._equipment.running_status = True def get_results(self): return self._equipment @logger.catch() async def get_fcu_control_result(project_id: str, equipment_id: str) -> FCU: async with AsyncClient() as client: duo_duo = EquipmentInfoService(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() temp_space_params = { 'id': sp.get('id'), 'temperature_target': current_target } space = Space(**temp_space_params) break fcu_temp_params = { 'id': equipment_id, 'space': space } fcu = FCU(**fcu_temp_params) fcu_controller = FCUController(fcu) await fcu_controller.run() regulated_fcu = fcu_controller.get_results() return regulated_fcu