123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520 |
- # -*- coding: utf-8 -*-
- from abc import ABCMeta, abstractmethod
- from typing import Dict, Tuple
- import arrow
- import numpy as np
- import pandas as pd
- from httpx import AsyncClient
- from loguru import logger
- from sqlalchemy.orm import Session
- from app.crud.space.weight import get_weights_by_space, update_weight
- from app.models.domain.feedback import FeedbackValue
- from app.resources.params import TEMPERATURE_TARGET_WEIGHT
- from app.schemas.sapce_weight import SpaceWeightUpdate
- from app.schemas.target import TemperatureTarget
- from app.services.platform import DataPlatformService
- from app.services.transfer import SpaceInfoService, Duoduo, 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
- clipper = Clipper()
- return clipper.cut(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 ExpSmoothingTemperatureTargetBuilder(GlobalTargetBaseBuilder):
- """
- Exponential smooth previous changes and set them as new global target.
- """
- def __init__(self, current_global_target: TemperatureTarget, previous_changes: pd.DataFrame):
- self.current_global_target = current_global_target
- self.previous_changes = previous_changes
- def build(self, new_actual_target: float) -> Dict:
- now_time = arrow.get(get_time_str(), TIME_FMT).time().strftime('%H%M%S')
- half_extent = self.current_global_target.extent / 2
- previous_changes = pd.concat([
- pd.DataFrame({'timestamp': [now_time], 'value': [new_actual_target]}),
- self.previous_changes,
- ])
- previous_changes.reset_index(inplace=True)
- previous_changes['weight1'] = previous_changes['index'].apply(lambda x: (1 / (x + 1)) ** 3)
- new_targets = {}
- time_index = self.current_global_target.target_schedule['temperatureMin'].keys()
- for item in time_index:
- previous_changes['delta'] = previous_changes['timestamp'].apply(
- lambda x: abs(arrow.get(str(x), 'HHmmss') - arrow.get(item, 'HHmmss')).seconds // (15 * 60)
- )
- previous_changes['weight2'] = previous_changes['delta'].apply(lambda x: 0.5 ** x)
- previous_changes['weight'] = previous_changes['weight1'] * previous_changes['weight2']
- temp_target = ((previous_changes['value'] * previous_changes['weight']).sum()
- / previous_changes['weight'].sum())
- new_targets.update({item: [temp_target - half_extent, temp_target + half_extent]})
- return new_targets
- 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 Carrier(metaclass=ABCMeta):
- """
- Fetch all you need data by one http client.
- """
- @abstractmethod
- async def fetch_all(self) -> None:
- raise NotImplementedError
- class Packer(metaclass=ABCMeta):
- """
- Arrange raw data for using.
- """
- @abstractmethod
- def run(self) -> Dict:
- raise NotImplementedError
- 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 TemperatureTargetCarrier(Carrier):
- """
- Fetch all the data that temperature target adjustment will use.
- """
- def __init__(self, project_id: str, object_id: str):
- self.project_id = project_id
- self.object_id = object_id
- self.result = {}
- async def fetch_all(self) -> None:
- async with AsyncClient() as client:
- transfer = SpaceInfoService(client, self.project_id, self.object_id)
- duoduo = Duoduo(client, self.project_id)
- platform = DataPlatformService(client, self.project_id)
- realtime_temperature = await platform.get_realtime_temperature(self.object_id)
- all_day_targets = await transfer.get_custom_target()
- current_target = await transfer.get_current_temperature_target()
- is_customized = await transfer.is_customized()
- is_temporary = await transfer.is_temporary()
- season = await duoduo.get_season()
- self.result = {
- 'realtime_temperature': realtime_temperature,
- 'all_day_targets': all_day_targets,
- 'current_target': current_target,
- 'is_customized': is_customized,
- 'is_temporary': is_temporary,
- 'season': season
- }
- async def get_result(self) -> Dict:
- await self.fetch_all()
- return self.result
- class TemperatureTargetV2Carrier(TemperatureTargetCarrier):
- """
- Add previous adjustment result to result.
- """
- async def fetch_previous_changes(self) -> None:
- async with AsyncClient() as client:
- transfer = SpaceInfoService(client, self.project_id, self.object_id)
- previous_changes = await transfer.env_database_get()
- self.result.update({'previous_changes': previous_changes['temperature']})
- async def get_result(self) -> Dict:
- await self.fetch_all()
- await self.fetch_previous_changes()
- return self.result
- class TemperatureTargetPacker:
- """
- Arrange raw data for temperature target adjustment.
- """
- def __init__(self, data):
- self.result = data
- def get_temperature_target(self):
- all_day_targets = self.result['all_day_targets']
- 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().to_dict()
- target_params = {
- 'is_customized': self.result['is_customized'],
- 'is_temporary': self.result['is_temporary'],
- 'target_schedule': temperature_all_day_targets,
- 'extent': extent
- }
- target = TemperatureTarget(**target_params)
- self.result.update({'target': target})
- def get_result(self) -> Dict:
- self.get_temperature_target()
- return self.result
- class TargetDeliver:
- """
- Send target adjustment result to transfer.
- """
- def __init__(self, project_id: str, space_id: str):
- self.project_id = project_id
- self.space_id = space_id
- async def send(self, controlled_result: Dict):
- async with AsyncClient() as client:
- transfer = SpaceInfoService(client, self.project_id, self.space_id)
- if controlled_result['need_switch_off']:
- await transfer.set_temporary_custom()
- if controlled_result['new_temporary_target']:
- await transfer.set_custom_target('temperature', controlled_result['new_temporary_target'], '0')
- if controlled_result['new_global_target']:
- await transfer.set_custom_target('temperature', controlled_result['new_global_target'], '1')
- if controlled_result['new_actual_target'] > 0 and controlled_result['need_run_room_control']:
- await transfer.env_database_set('temperature', controlled_result['new_actual_target'])
- class WeightFlagDeliver:
- """
- Change a space temporary weight when the space receives a feedback about temperature.
- """
- def __init__(self, db: Session, feedback: FeedbackValue):
- self.db = db
- self.feedback = feedback
- def is_temperature_feedback(self) -> bool:
- if (self.feedback == FeedbackValue.a_little_hot
- or self.feedback == FeedbackValue.so_hot
- or self.feedback == FeedbackValue.a_little_cold
- or self.feedback == FeedbackValue.so_cold):
- flag = True
- else:
- flag = False
- return flag
- def save(self, space: str):
- if self.is_temperature_feedback():
- weights = get_weights_by_space(self.db, space_id=space)
- for weight in weights:
- weight_in = SpaceWeightUpdate(temporary_weight=1.0)
- update_weight(self.db, db_weight=weight, weight_in=weight_in)
- class TemperatureTargetController:
- """
- Primary flow of temperature target adjustment for Sequoia.
- """
- def __init__(self, data: Dict):
- self.data = data
- self.result = {}
- def run(self, feedback: FeedbackValue):
- need_switch_off = False
- new_temporary_target = {}
- new_global_target = {}
- new_actual_target = 0
- if feedback == FeedbackValue.switch_off:
- need_switch_off = True
- need_run_room_control = True
- elif feedback == FeedbackValue.switch_on:
- need_run_room_control = True
- if not self.data['is_customized']:
- new_lower, new_upper = TemporaryTargetInit(1, 24).build(
- self.data['extent'],
- self.data['season'],
- self.data['realtime_temperature']
- )
- new_temporary_target = TemporaryTargetBuilder(new_lower, new_upper).build()
- elif (feedback == FeedbackValue.a_little_hot
- or feedback == FeedbackValue.a_little_cold
- or feedback == FeedbackValue.so_hot
- or feedback == FeedbackValue.so_cold):
- step_size = StepSizeCalculator(TEMPERATURE_TARGET_WEIGHT).run(
- self.data['realtime_temperature'],
- 25.0,
- feedback
- )
- new_actual_target = NewTemperatureTargetBuilder(
- self.data['realtime_temperature'],
- self.data['current_target'],
- step_size
- ).build()
- need_run_room_control = True
- if new_actual_target != self.data['current_target']:
- new_global_target = SimpleGlobalTemperatureTargetBuilder(self.data['target']).build(new_actual_target)
- else:
- need_run_room_control = False
- self.result.update({
- 'need_switch_off': need_switch_off,
- 'new_temporary_target': new_temporary_target,
- 'new_global_target': new_global_target,
- 'new_actual_target': new_actual_target,
- 'need_run_room_control': need_run_room_control
- })
- def get_result(self) -> Dict:
- return self.result
- class TemperatureTargetControllerV2:
- """
- Primary flow of temperature target adjustment for Zhonghai.
- """
- def __init__(self, data: Dict):
- self.data = data
- self.result = {}
- def run(self, feedback: FeedbackValue):
- need_switch_off = False
- new_temporary_target = {}
- new_global_target = {}
- new_actual_target = 0
- if feedback == FeedbackValue.switch_off:
- need_switch_off = True
- need_run_room_control = True
- elif feedback == FeedbackValue.switch_on:
- need_run_room_control = True
- if not self.data['target'].is_customized:
- new_lower, new_upper = TemporaryTargetInit(1, 24).build(
- self.data['target'].extent,
- self.data['season'],
- self.data['realtime_temperature']
- )
- new_temporary_target = TemporaryTargetBuilder(new_lower, new_upper).build()
- elif (feedback == FeedbackValue.a_little_hot
- or feedback == FeedbackValue.a_little_cold
- or feedback == FeedbackValue.so_hot
- or feedback == FeedbackValue.so_cold):
- step_size = StepSizeCalculator(TEMPERATURE_TARGET_WEIGHT).run(
- self.data['realtime_temperature'],
- 25.0,
- feedback
- )
- new_actual_target = NewTemperatureTargetBuilder(
- self.data['realtime_temperature'],
- self.data['current_target'],
- step_size
- ).build()
- need_run_room_control = True
- if new_actual_target != self.data['current_target']:
- new_global_target = ExpSmoothingTemperatureTargetBuilder(
- self.data['target'],
- self.data['previous_changes']
- ).build(new_actual_target)
- else:
- need_run_room_control = False
- self.result.update({
- 'need_switch_off': need_switch_off,
- 'new_temporary_target': new_temporary_target,
- 'new_global_target': new_global_target,
- 'new_actual_target': new_actual_target,
- 'need_run_room_control': need_run_room_control
- })
- def get_result(self) -> Dict:
- return self.result
- @logger.catch()
- async def temperature_target_control_v1(project_id: str, space_id: str, feedback: FeedbackValue, db: Session) -> bool:
- temperature_target_raw_data = await TemperatureTargetCarrier(project_id, space_id).get_result()
- temperature_target_data = TemperatureTargetPacker(temperature_target_raw_data).get_result()
- controller = TemperatureTargetController(temperature_target_data)
- controller.run(feedback)
- controlled_result = controller.get_result()
- await TargetDeliver(project_id, space_id).send(controlled_result)
- WeightFlagDeliver(db, feedback).save(space_id)
- return controlled_result['need_run_room_control']
- @logger.catch()
- async def temperature_target_control_v2(project_id: str, space_id: str, feedback: FeedbackValue) -> bool:
- temperature_target_raw_data = await TemperatureTargetV2Carrier(project_id, space_id).get_result()
- temperature_target_data = TemperatureTargetPacker(temperature_target_raw_data).get_result()
- controller = TemperatureTargetControllerV2(temperature_target_data)
- controller.run(feedback)
- controlled_result = controller.get_result()
- await TargetDeliver(project_id, space_id).send(controlled_result)
- return controlled_result['need_run_room_control']
- @logger.catch()
- async def get_target_after_feedback(project_id: str, space_id: str, feedback: FeedbackValue) -> float:
- if project_id == 'Pj1101050030':
- temperature_target_raw_data = await TemperatureTargetCarrier(project_id, space_id).get_result()
- else:
- temperature_target_raw_data = await TemperatureTargetV2Carrier(project_id, space_id).get_result()
- temperature_target_data = TemperatureTargetPacker(temperature_target_raw_data).get_result()
- if project_id == 'Pj1101050030':
- controller = TemperatureTargetController(temperature_target_data)
- else:
- controller = TemperatureTargetControllerV2(temperature_target_data)
- controller.run(feedback)
- controlled_result = controller.get_result()
- return controlled_result.get('new_actual_target')
|