targets.py 15 KB

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