import time from typing import Dict, Tuple import arrow from httpx import AsyncClient from loguru import logger from app.utils.math import round_half_up from app.utils.date import get_time_str, TIME_FMT 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: mode = 'normal' else: mode = 'on_ratio' return mode def select_speed(self, last_return_air: float) -> str: mode = self.select_mode() if mode == 'off': speed = 'off' elif mode == 'normal': if (self.return_air - last_return_air) > 0.8: speed = 'medium' else: speed = 'high' else: speed = 'medium' return speed def select_water_valve(self) -> bool: mode = self.select_mode() if mode == 'off': switch = True elif mode == 'normal': switch = False else: switch = True return switch def calculate_on_ratio(self, delta_on: float, delta_off: float) -> float: if self.period_num == 0: ratio = 0.5 else: if delta_on <= 0: ratio = 1.0 else: try: ratio = (0.5 * (self.return_air - self.target) + delta_off) / (delta_off - delta_on) except ZeroDivisionError: ratio = 0.0 return ratio async def send_instructions(switch: bool, speed: str, water_valve: bool) -> None: pass async def fetch_params(device_id: str, period_time: float, last_ratio: float) -> Tuple[float, float, float, float]: pass @logger.catch() async def on_ratio_experiment(device_id): _TARGET = 26.0 _DAILY_ON_TIME = '060000' _DAILY_OFF_TIME = '220000' period_time = 10 * 60 period_num = 0 last_on_ratio = 1.0 life_count = 0 while life_count < 1000: 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, last_return_air, 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(last_return_air) switch = False if speed == 'off' else True water_valve = controller.select_water_valve() if mode == 'on_ratio': on_ratio = controller.calculate_on_ratio(delta_on, delta_off) on_range = round_half_up(period_time * on_ratio) off_range = period_time - on_range await send_instructions(switch, speed, water_valve) time.sleep(on_range) await send_instructions(switch, speed, False) time.sleep(off_range) period_num += 1 last_on_ratio = on_ratio else: await send_instructions(switch, speed, water_valve) time.sleep(period_time) period_num = 0 last_on_ratio = 1.0 life_count += 1 else: await send_instructions(False, 'off', False) period_num = 0 last_on_ratio = 0.0