targets.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371
  1. # -*- coding: utf-8 -*-
  2. from abc import abstractmethod
  3. import time
  4. from typing import Dict, Tuple, Optional
  5. import arrow
  6. import numpy as np
  7. import pandas as pd
  8. from httpx import AsyncClient
  9. from loguru import logger
  10. from app.controllers.controller import Controller
  11. from app.resources.params import (
  12. TEMPERATURE_RELATED_FEEDBACK_WEIGHT,
  13. TEMPERATURE_RELATED_FEEDBACK,
  14. CO2_RELATED_FEEDBACK_WEIGHT,
  15. SWITCH_RELATED_FEEDBACK
  16. )
  17. from app.services.platform import DataPlatformService
  18. from app.services.transfer import TransferService, Season
  19. from app.utils.date import get_time_str, get_quarter_minutes, TIME_FMT
  20. class TargetController(Controller):
  21. def __init__(
  22. self,
  23. realtime_data: float,
  24. feedback: Dict,
  25. is_customized: bool,
  26. is_temporary: bool,
  27. current_targets: pd.DataFrame,
  28. ) -> None:
  29. super(TargetController, self).__init__()
  30. self._realtime_data = realtime_data
  31. self._feedback = feedback
  32. self._is_customized = is_customized
  33. self._is_temporary = is_temporary
  34. self._current_targets = current_targets
  35. self._now_time = arrow.get(get_time_str(), TIME_FMT).time().strftime('%H%M%S')
  36. self._quarter_time = get_quarter_minutes(get_time_str())
  37. async def calculate_diff(self, weight: Dict) -> float:
  38. related_feedback = [v for k, v in self._feedback.items() if k in weight]
  39. related_feedback = np.array(related_feedback)
  40. weight = np.array(list(weight.values()))
  41. feedback_count = related_feedback.sum()
  42. diff = 0
  43. if feedback_count > 0:
  44. diff = np.dot(related_feedback, weight) / feedback_count
  45. return diff
  46. @abstractmethod
  47. async def init_temporary(self):
  48. pass
  49. @abstractmethod
  50. async def get_targets(self):
  51. pass
  52. async def generate_temporary(self, lower, upper):
  53. now_str = get_time_str()
  54. time_index = arrow.get(arrow.get(now_str, TIME_FMT).shift(minutes=15).timestamp
  55. // (15 * 60) * (15 * 60)).time().strftime('%H%M%S')
  56. result = {time_index: [lower, upper]}
  57. self._results.update({'temporary_targets': result})
  58. async def readjust_global(self, latest_change: float, previous_changes: pd.DataFrame):
  59. previous_changes = pd.concat([
  60. pd.DataFrame({'timestamp': [self._now_time], 'value': [latest_change]}),
  61. previous_changes,
  62. ])
  63. previous_changes.reset_index(inplace=True)
  64. previous_changes['weight1'] = previous_changes['index'].apply(lambda x: (1 / (x + 1)) ** 3)
  65. new_targets = []
  66. time_index = self._current_targets.reset_index()['time']
  67. for item in time_index:
  68. previous_changes['delta'] = previous_changes['timestamp'].apply(
  69. lambda x: abs(arrow.get(str(x), 'HHmmss') - arrow.get(item, 'HHmmss')).seconds // (15 * 60)
  70. )
  71. previous_changes['weight2'] = previous_changes['delta'].apply(lambda x: 0.5 ** x)
  72. previous_changes['weight'] = previous_changes['weight1'] * previous_changes['weight2']
  73. new_targets.append(
  74. (previous_changes['value'] * previous_changes['weight']).sum() / previous_changes['weight'].sum()
  75. )
  76. self._current_targets['new_targets'] = new_targets
  77. @abstractmethod
  78. async def run(self):
  79. pass
  80. class TemperatureTargetController(TargetController):
  81. def __init__(
  82. self,
  83. realtime_data: float,
  84. feedback: Dict,
  85. is_customized: bool,
  86. is_temporary: bool,
  87. current_targets: pd.DataFrame,
  88. season: Season,
  89. previous_changes: Optional[pd.DataFrame] = None
  90. ) -> None:
  91. super(TemperatureTargetController, self).__init__(
  92. realtime_data,
  93. feedback,
  94. is_customized,
  95. is_temporary,
  96. current_targets
  97. )
  98. self._season = season
  99. self._previous_changes = previous_changes
  100. @staticmethod
  101. def _cut(value: float) -> float:
  102. _LOWER_LIMIT = 21.0
  103. _UPPER_LIMIT = 29.0
  104. value = min(value, _UPPER_LIMIT)
  105. value = max(value, _LOWER_LIMIT)
  106. return value
  107. async def init_temporary(self) -> Tuple[float, float]:
  108. _VAR = 2
  109. _RANGE = 1
  110. new_lower_bound, new_upper_bound = 23.0, 25.0
  111. if self._realtime_data is not np.NAN:
  112. if self._season == Season.cooling:
  113. if ('a little hot' in self._feedback
  114. or 'so hot' in self._feedback
  115. or 'switch on' in self._feedback):
  116. mid = self._realtime_data - _VAR
  117. new_lower_bound = mid - _RANGE
  118. new_upper_bound = mid + _RANGE
  119. elif self._season == Season.heating:
  120. if ('a little cold' in self._feedback
  121. or 'so cold' in self._feedback
  122. or 'switch on' in self._feedback):
  123. mid = self._realtime_data + _VAR
  124. new_lower_bound = mid - _RANGE
  125. new_upper_bound = mid + _RANGE
  126. return self._cut(new_lower_bound), self._cut(new_upper_bound)
  127. async def get_targets(self):
  128. current_lower_target = self._current_targets['temperatureMin'].loc[self._quarter_time]
  129. current_upper_target = self._current_targets['temperatureMax'].loc[self._quarter_time]
  130. if current_lower_target is np.NAN:
  131. current_lower_target = 23.0
  132. if current_upper_target is np.NAN:
  133. current_upper_target = 25.0
  134. return current_lower_target, current_upper_target
  135. async def readjust_current(self, lower: float, upper: float, diff: float) -> Tuple[float, float]:
  136. _RANGE = 2
  137. new_lower_bound, new_upper_bound = lower, upper
  138. if self._realtime_data is np.NAN:
  139. new_lower_bound += lower + diff
  140. new_upper_bound += upper + diff
  141. else:
  142. if self._season == Season.cooling:
  143. if (diff > 0 and self._realtime_data + _RANGE > upper
  144. or diff < 0 and self._realtime_data - _RANGE < upper):
  145. new_upper_bound = self._realtime_data + diff
  146. new_lower_bound = new_upper_bound - (upper - lower)
  147. else:
  148. if (diff > 0 and self._realtime_data + _RANGE > lower
  149. or diff < 0 and self._realtime_data - _RANGE < lower):
  150. new_lower_bound = self._realtime_data + diff
  151. new_upper_bound = new_lower_bound + (upper - lower)
  152. return self._cut(new_lower_bound), self._cut(new_upper_bound)
  153. async def generate_global(self, lower: float, upper: float):
  154. extent = upper - lower
  155. if self._season == Season.cooling:
  156. new_targets = self._current_targets['new_targets'].apply(lambda x: [x - extent, x])
  157. else:
  158. new_targets = self._current_targets['new_targets'].apply(lambda x: [x, x + extent])
  159. time_index = self._current_targets.reset_index()['time']
  160. result = {}
  161. for i in range(len(time_index)):
  162. result.update({time_index[i]: new_targets[i]})
  163. self._results.update({'global_targets': result})
  164. async def run(self):
  165. diff = await self.calculate_diff(TEMPERATURE_RELATED_FEEDBACK_WEIGHT)
  166. if diff != 0:
  167. if not self._is_customized:
  168. lower_bound, upper_bound = await self.init_temporary()
  169. await self.generate_temporary(lower_bound, upper_bound)
  170. else:
  171. current_lower, current_upper = await self.get_targets()
  172. lower_bound, upper_bound = await self.readjust_current(current_lower, current_upper, diff)
  173. if self._season == Season.cooling:
  174. temp_target = upper_bound
  175. else:
  176. temp_target = lower_bound
  177. if not self._is_temporary:
  178. self._results.update({'latest_change': temp_target})
  179. await self.readjust_global(temp_target, self._previous_changes)
  180. await self.generate_global(current_lower, current_upper)
  181. else:
  182. await self.generate_temporary(lower_bound, upper_bound)
  183. else:
  184. return
  185. class Co2TargetController(TargetController):
  186. def __init__(
  187. self,
  188. realtime_data: float,
  189. feedback: Dict,
  190. is_customized: bool,
  191. is_temporary: bool,
  192. current_targets: pd.DataFrame,
  193. previous_changes: Optional[pd.DataFrame] = None
  194. ) -> None:
  195. super(Co2TargetController, self).__init__(
  196. realtime_data,
  197. feedback,
  198. is_customized,
  199. is_temporary,
  200. current_targets
  201. )
  202. self._previous_changes = previous_changes
  203. @staticmethod
  204. def _cut(value: float) -> float:
  205. _UPPER_LIMIT = 1000.0
  206. value = min(value, _UPPER_LIMIT)
  207. return value
  208. async def init_temporary(self) -> float:
  209. new_target = 1000
  210. diff = await self.calculate_diff(CO2_RELATED_FEEDBACK_WEIGHT)
  211. if self._realtime_data is not np.NAN:
  212. new_target += diff
  213. return self._cut(new_target)
  214. async def get_targets(self):
  215. current_upper_target = self._current_targets['co2Max'].loc[self._quarter_time]
  216. if current_upper_target is np.NAN:
  217. current_upper_target = 500.0
  218. return current_upper_target
  219. async def readjust_current(self, lower: float, upper: float, diff: float) -> float:
  220. new_target = upper - lower
  221. if self._realtime_data is np.NAN:
  222. new_target += diff
  223. else:
  224. if (diff > 50 and self._realtime_data + 100 > upper
  225. or diff < -50 and self._realtime_data - 100 < upper):
  226. new_target = self._realtime_data + diff
  227. return self._cut(new_target)
  228. async def generate_global(self):
  229. new_targets = self._current_targets['new_targets'].apply(lambda x: [0, x])
  230. time_index = self._current_targets.reset_index()['time']
  231. result = {}
  232. for i in range(len(time_index)):
  233. result.update({time_index[i]: new_targets[i]})
  234. self._results.update({'global_targets': result})
  235. async def run(self):
  236. diff = await self.calculate_diff(CO2_RELATED_FEEDBACK_WEIGHT)
  237. if diff != 0:
  238. if not self._is_customized:
  239. upper_bound = await self.init_temporary()
  240. await self.generate_temporary(0, upper_bound)
  241. else:
  242. current_upper = await self.get_targets()
  243. upper_bound = await self.readjust_current(0, current_upper, diff)
  244. if not self._is_temporary:
  245. self._results.update({'latest_change': upper_bound})
  246. await self.readjust_global(upper_bound, self._previous_changes)
  247. await self.generate_global()
  248. else:
  249. await self.generate_temporary(0, upper_bound)
  250. else:
  251. return
  252. @logger.catch()
  253. async def readjust_all_target(project_id: str, space_id: str, wechat_time: str):
  254. start = time.perf_counter()
  255. async with AsyncClient() as client:
  256. transfer = TransferService(client, project_id, space_id)
  257. platform = DataPlatformService(client, project_id, space_id)
  258. nl = '\n'
  259. realtime_temperature = await platform.get_realtime_temperature()
  260. logger.debug(f'realtime temperature: {realtime_temperature}')
  261. current_targets = await transfer.get_custom_target()
  262. logger.debug(f'current targets: {nl}{current_targets}')
  263. feedback = await transfer.get_feedback(wechat_time)
  264. feedback_for_log = {k: v for k, v in feedback.items() if v > 0}
  265. logger.debug(f'feedback: {feedback_for_log}')
  266. is_customized = await transfer.is_customized()
  267. logger.debug(f'is customized: {is_customized}')
  268. is_temporary = await transfer.is_temporary()
  269. logger.debug(f'is temporary: {is_temporary}')
  270. season = await transfer.get_season()
  271. logger.debug(f'season: {season}')
  272. previous_changes = await transfer.env_database_get()
  273. temperature_changes = previous_changes.get('temperature')
  274. logger.debug(f'temperature previous changes: {nl}{temperature_changes}')
  275. logger.debug(f'request web service elapsed: {time.perf_counter() - start}')
  276. if feedback.get('switch off') and feedback.get('switch off') > 0:
  277. need_switch_off = True
  278. for item in SWITCH_RELATED_FEEDBACK:
  279. if feedback.get(item) and feedback.get(item) > 0:
  280. need_switch_off = False
  281. break
  282. else:
  283. need_switch_off = False
  284. need_run_room_control = False
  285. if need_switch_off:
  286. async with AsyncClient() as client:
  287. transfer = TransferService(client, project_id, space_id)
  288. await transfer.set_temporary_custom()
  289. return need_run_room_control
  290. temperature_results = {}
  291. for item in TEMPERATURE_RELATED_FEEDBACK:
  292. if feedback.get(item) and feedback.get(item) > 0:
  293. temperature_controller = TemperatureTargetController(
  294. realtime_temperature,
  295. feedback,
  296. is_customized,
  297. is_temporary,
  298. current_targets[['temperatureMin', 'temperatureMax']].copy(),
  299. season,
  300. previous_changes['temperature']
  301. )
  302. await temperature_controller.run()
  303. temperature_results = temperature_controller.get_results()
  304. break
  305. if temperature_results:
  306. need_run_room_control = True
  307. async with AsyncClient() as client:
  308. transfer = TransferService(client, project_id, space_id)
  309. if temperature_results.get('temporary_targets'):
  310. temporary_targets = temperature_results.get('temporary_targets')
  311. logger.debug(f'temperature temporary targets: {nl}{temporary_targets}')
  312. await transfer.set_custom_target('temperature', temperature_results.get('temporary_targets'), '0')
  313. if temperature_results.get('global_targets'):
  314. global_targets = temperature_results.get('global_targets')
  315. logger.debug(f'temperature global targets: {nl}{global_targets}')
  316. await transfer.set_custom_target('temperature', temperature_results.get('global_targets'), '1')
  317. if temperature_results.get('latest_change'):
  318. latest_change = temperature_results.get('latest_change')
  319. logger.debug(f'temperature latest change: {latest_change}')
  320. await transfer.env_database_set('temperature', temperature_results.get('latest_change'))
  321. return need_run_room_control