123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122 |
- 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
|