targets.py 15 KB

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