import asyncio from typing import Tuple import arrow from fastapi import HTTPException from httpx import AsyncClient from loguru import logger from app.controllers.equipment.fcu.basic import FCUControllerV2 from app.schemas.equipment import FCU from app.schemas.space import Space from app.services.platform import DataPlatformService, InfoCode from app.services.transfer import Season from app.utils.date import get_time_str, TIME_FMT from app.utils.math import round_half_up class OnRatioController: def __init__(self, target: float, return_air: float, period_num: int): super(OnRatioController, self).__init__() self.target = target self.return_air = return_air self.period_num = period_num def select_mode(self) -> str: if self.target <= self.return_air: mode = 'off' else: if self.target - self.return_air >= 1.5: mode = 'normal' else: mode = 'on_ratio' return mode def select_speed(self, delta_all: float) -> str: mode = self.select_mode() if mode == 'off': speed = 'medium' elif mode == 'normal': if delta_all > 0.00088889: speed = 'medium' else: speed = 'high' else: speed = 'medium' return speed def select_water_valve(self) -> bool: mode = self.select_mode() if mode == 'off': switch = False elif mode == 'normal': switch = True else: switch = True return switch def calculate_on_ratio(self, period_time: float, delta_on: float, delta_off: float, last_mode: str) -> float: if self.period_num == 0: if last_mode == 'normal': ratio = 0.8 elif last_mode == 'off': ratio = 0.2 else: if self.return_air > self.target - 0.75: ratio = 0.2 else: ratio = 0.8 else: if delta_on <= 0: ratio = 0.9 else: try: if delta_off > 0: delta_off = 0.0 ratio = (0.5 * (self.target - self.return_air) / period_time - delta_off) / (delta_on - delta_off) if ratio > 0.9: ratio = 0.9 if ratio < 0.1: ratio = 0.1 if delta_on <= 0: ratio = 0.5 logger.debug(f'delta target: {0.5 * (self.target - self.return_air)}') except ZeroDivisionError: ratio = 0.1 return ratio async def send_instructions(device_id: str, switch: bool, speed: str, water_valve: bool) -> None: switch_value = 1.0 if switch else 0.0 water_valve_value = 1.0 if water_valve else 0.0 speed_value_dict = { 'off': 0.0, 'low': 1.0, 'medium': 2.0, 'high': 3.0 } speed_value = speed_value_dict.get(speed) async with AsyncClient() as client: platform = DataPlatformService(client, 'Pj1101080259') await platform.set_code_value(device_id, code=InfoCode.equip_switch_set, value=switch_value) await platform.set_code_value(device_id, code=InfoCode.water_valve_switch_set, value=water_valve_value) await platform.set_code_value(device_id, code=InfoCode.fan_speed_set, value=speed_value) await platform.set_code_value(device_id, code=InfoCode.work_mode_set, value=2.0) # await platform.set_code_value(device_id, code=InfoCode.in_cloud_set, value=1.0) async def fetch_params(device_id: str, period_time: int, last_ratio: float) -> Tuple[float, float, float, float]: async with AsyncClient() as client: middle_time = int(round_half_up(period_time * (1 - last_ratio))) platform = DataPlatformService(client, 'Pj1101080259') return_air_c = await platform.get_realtime_data(InfoCode.return_air_temperature, device_id) return_air_a = await platform.get_past_data(InfoCode.return_air_temperature, device_id, period_time) return_air_b = await platform.get_past_data(InfoCode.return_air_temperature, device_id, middle_time) delta_all = (return_air_c - return_air_a) / period_time if 0 < last_ratio < 1: delta_on = (return_air_b - return_air_a) / (period_time - middle_time) delta_off = (return_air_c - return_air_b) / middle_time elif last_ratio == 0.0: delta_on = 0.0 delta_off = (return_air_c - return_air_b) / middle_time else: delta_on = (return_air_b - return_air_a) / (period_time - middle_time) delta_off = 0.0 return return_air_c, delta_all, delta_on, delta_off @logger.catch() async def start_on_ratio_mode(device_id: str, target: float, period_time: int): _DAILY_ON_TIME = '060000' _DAILY_OFF_TIME = '220000' period_num = 0 last_on_ratio = 1.0 last_mode = 'default' life_count = 0 while life_count < 2000: try: time_str = get_time_str() if _DAILY_ON_TIME <= arrow.get(time_str, TIME_FMT).time().strftime('%H%M%S') < _DAILY_OFF_TIME: return_air, delta_all, delta_on, delta_off = await fetch_params(device_id, period_time, last_on_ratio) controller = OnRatioController(target, return_air, period_num) mode = controller.select_mode() speed = controller.select_speed(delta_all) switch = False if speed == 'off' else True water_valve = controller.select_water_valve() if mode == 'on_ratio': on_ratio = controller.calculate_on_ratio(period_time, delta_on, delta_off, last_mode) on_range = round_half_up(period_time * on_ratio) off_range = period_time - on_range logger.debug(f'life count: {life_count}, {device_id}, on time: {on_range}, off time: {off_range}, ' f'on ratio: {on_ratio}, delta_on: {delta_on * 900}, delta_off: {delta_off * 900}') await send_instructions(device_id, switch, speed, water_valve) logger.debug(f'{device_id}, {switch}, {speed}, {water_valve}') await asyncio.sleep(on_range) await send_instructions(device_id, switch, speed, False) logger.debug(f'{device_id}, {switch}, {speed}, off') await asyncio.sleep(off_range) period_num += 1 last_on_ratio = on_ratio last_mode = mode else: await send_instructions(device_id, switch, speed, water_valve) logger.debug(f'life count: {life_count}, {device_id}, {switch}, {speed}, {water_valve}') await asyncio.sleep(period_time) period_num = 0 last_on_ratio = 1.0 last_mode = mode life_count += 1 else: await send_instructions(device_id, False, 'off', False) period_num = 0 last_on_ratio = 0.0 last_mode = 'off' await asyncio.sleep(period_time) except (KeyError, IndexError, TypeError, HTTPException): await asyncio.sleep(period_time) continue @logger.catch() async def start_control_group_mode(device_id: str, target: float): async with AsyncClient() as client: platform = DataPlatformService(client, 'Pj1101080259') return_air = await platform.get_realtime_data(InfoCode.return_air_temperature, device_id) space_params = { 'id': device_id, 'temperature_target': target, 'realtime_temperature': return_air } space = Space(**space_params) fcu_params = { 'id': device_id, 'space': space } fcu = FCU(**fcu_params) controller = FCUControllerV2(fcu, Season.heating) await controller.run() regulated_fcu = controller.get_results() speed_value_dict = { 0.0: 'medium', 1.0: 'low', 2.0: 'medium', 3.0: 'high' } await send_instructions( device_id, True, speed_value_dict.get(regulated_fcu.air_valve_speed), regulated_fcu.running_status ) logger.info(f'{device_id} - {speed_value_dict.get(regulated_fcu.air_valve_speed)}')