# -*- coding: utf-8 -*- from abc import ABCMeta, abstractmethod from typing import Dict, Optional, Tuple import arrow import numpy as np from httpx import AsyncClient from loguru import logger from app.resources.params import TEMPERATURE_TARGET_WEIGHT from app.schemas.feedback import Feedback from app.schemas.target import TemperatureTarget from app.services.platform import DataPlatformService from app.services.transfer import SpaceInfoService, Season from app.utils.date import get_time_str, TIME_FMT class StepSizeCalculator: """ Calculate adjustment step size of environment target when a user send feedback. """ def __init__(self, weight: Dict): self.weight = weight def run( self, realtime_temperature: float, comfortable_temperature: float, feedback: Feedback ) -> float: base_step_size = 1.8 / (1 + np.exp((comfortable_temperature - realtime_temperature) / 2)) return self.weight.get(feedback.value) * base_step_size class NewTargetBuilder(metaclass=ABCMeta): """ Calculate a new target value. """ @abstractmethod def build(self) -> float: raise NotImplementedError class NewTemperatureTargetBuilder(NewTargetBuilder): """ Calculate a new temperature target value. """ def __init__(self, realtime_temperature: float, actual_target: float, step_size: float): self.realtime_temperature = realtime_temperature self.actual_target = actual_target self.step_size = step_size def build(self) -> float: new_actual_target = self.actual_target if self.step_size > 0: if self.realtime_temperature + self.step_size > self.actual_target: new_actual_target = self.realtime_temperature + self.step_size elif self.step_size < 0: if self.realtime_temperature + self.step_size < self.actual_target: new_actual_target = self.realtime_temperature + self.step_size return new_actual_target class Clipper: """ Return a number which is in the range of [min, max]. """ def __init__(self, upper_limit: float = 28.0, lower_limit: float = 22.0): self.upper_limit = upper_limit self.lower_limit = lower_limit def cut(self, num: float) -> float: num = min(num, self.upper_limit) num = max(num, self.lower_limit) return num class TemporaryTargetInit: """ Initialize temporary temperature target. """ def __init__(self, step_size: float, default_target: float = 24.0): self.step_size = step_size self.default_target = default_target def build( self, extent: float, season: Season, realtime_temperature: float, ) -> Tuple[float, float]: if np.isnan(realtime_temperature): upper_bound, lower_bound = self.default_target + 1.0, self.default_target - 1.0 else: actual_target = np.NAN if season == Season.cooling: actual_target = realtime_temperature - self.step_size elif season == Season.heating: actual_target = realtime_temperature + self.step_size clipper = Clipper() actual_target = clipper.cut(actual_target) upper_bound, lower_bound = actual_target + (extent / 2), actual_target - (extent / 2) return lower_bound, upper_bound class GlobalTargetBaseBuilder(metaclass=ABCMeta): """ Generate global target and format it for sending to TransferServer. """ @abstractmethod def build(self, new_actual_target: float) -> Dict: raise NotImplementedError class SimpleGlobalTemperatureTargetBuilder(GlobalTargetBaseBuilder): """ Set all day temperature target same. """ def __init__(self, current_global_target: TemperatureTarget): self.current_global_target = current_global_target def build(self, new_actual_target: float) -> Dict: result = {} half_extent = self.current_global_target.extent / 2 for time_index in self.current_global_target.target_schedule['temperatureMin'].keys(): result.update({time_index: [new_actual_target - half_extent, new_actual_target + half_extent]}) return result class TemporaryTargetBuilder: """ Generate global target and format it for sending to TransferServer. """ def __init__(self, lower_target: float, upper_target: float): self.lower_target = lower_target self.upper_target = upper_target def build(self) -> Dict: now_str = get_time_str() time_index = arrow.get(arrow.get(now_str, TIME_FMT).shift(minutes=15).timestamp // (15 * 60) * (15 * 60)).time().strftime('%H%M%S') return {time_index: [self.lower_target, self.upper_target]} class AdjustmentController(metaclass=ABCMeta): """ Fetch some data, assemble target adjustment related functions and classes, send the new target to transfer server, and return a flag which denote whether transfer server need to request room/control. """ @abstractmethod async def run(self) -> bool: raise NotImplementedError class TemperatureAdjustmentController(AdjustmentController): """ Test strategy for Sequoia. """ def __init__( self, project_id: str, space_id: str, wechat_time: Optional[str] = None, raw_feedback: Optional[Dict] = None ): self.project_id = project_id self.space_id = space_id self.wechat_time = wechat_time self.raw_feedback = raw_feedback @logger.catch() async def run(self) -> bool: # Fetch all data from data platform and transfer server. async with AsyncClient() as client: transfer = SpaceInfoService(client, self.project_id, self.space_id) platform = DataPlatformService(client, self.project_id) realtime_temperature = await platform.get_realtime_temperature(self.space_id) logger.debug(f'realtime temperature: {realtime_temperature}') all_day_targets = await transfer.get_custom_target() current_target = await transfer.get_current_temperature_target() logger.debug(f'current target: {current_target}') if self.wechat_time: self.raw_feedback = await transfer.get_feedback(self.wechat_time) logger.debug(f'raw feedback: {self.raw_feedback}') is_customized = await transfer.is_customized() logger.debug(f'is customized? {is_customized}') is_temporary = await transfer.is_temporary() logger.debug(f'is temporary? {is_temporary}') season = await transfer.get_season() # Arrange data. feedback = Feedback('nice') for k, v in self.raw_feedback.items(): if v > 0: feedback = Feedback(k) break logger.debug(feedback) if len(all_day_targets) > 0: extent = all_day_targets['temperatureMax'].iloc[0] - all_day_targets['temperatureMin'].iloc[0] else: extent = 2.0 temperature_all_day_targets = all_day_targets[['temperatureMin', 'temperatureMax']].copy() target_params = { 'is_customized': is_customized, 'is_temporary': is_temporary, 'target_schedule': temperature_all_day_targets.to_dict(), 'extent': extent } target = TemperatureTarget(**target_params) # Primary flow need_switch_off = False new_temporary_target = {} new_global_target = {} new_actual_target = 0 if feedback == Feedback.switch_off: need_switch_off = True need_run_room_control = True elif feedback == Feedback.switch_on: need_run_room_control = True if not is_customized: new_lower, new_upper = TemporaryTargetInit(1, 24).build(extent, season, realtime_temperature) new_temporary_target = TemporaryTargetBuilder(new_lower, new_upper) elif (feedback == Feedback.a_little_hot or feedback == Feedback.a_little_cold or feedback == Feedback.so_hot or feedback == Feedback.so_cold): step_size = StepSizeCalculator(TEMPERATURE_TARGET_WEIGHT).run(realtime_temperature, 25.0, feedback) logger.debug(f'step size: {step_size}') new_actual_target = NewTemperatureTargetBuilder(realtime_temperature, current_target, step_size).build() if new_actual_target != current_target: new_global_target = SimpleGlobalTemperatureTargetBuilder(target).build(new_actual_target) need_run_room_control = True else: need_run_room_control = False else: need_run_room_control = False # Send adjustment result to transfer server async with AsyncClient() as client: transfer = SpaceInfoService(client, self.project_id, self.space_id) if need_switch_off: await transfer.set_temporary_custom() logger.debug(f'turn off') if new_temporary_target: await transfer.set_custom_target('temperature', new_temporary_target, '0') logger.debug(f'new temporary targets: {new_temporary_target}') if new_global_target: await transfer.set_custom_target('temperature', new_global_target, '1') logger.debug(f'new global targets: {new_global_target}') if new_actual_target != current_target: await transfer.env_database_set('temperature', new_actual_target) logger.debug(f'new actual target: {new_actual_target}') return need_run_room_control