# -*- 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.models.domain.feedback import FeedbackValue from app.resources.params import TEMPERATURE_TARGET_WEIGHT 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: FeedbackValue ) -> 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, feedback: Optional[FeedbackValue] = None ): self.project_id = project_id self.space_id = space_id self.feedback = 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}') 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 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 self.feedback == FeedbackValue.switch_off: need_switch_off = True need_run_room_control = True elif self.feedback == FeedbackValue.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 (self.feedback == FeedbackValue.a_little_hot or self.feedback == FeedbackValue.a_little_cold or self.feedback == FeedbackValue.so_hot or self.feedback == FeedbackValue.so_cold): step_size = StepSizeCalculator(TEMPERATURE_TARGET_WEIGHT).run(realtime_temperature, 25.0, self.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