123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272 |
- # -*- 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
|