targets.py 16 KB

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