q_learning.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126
  1. # -*- coding: utf-8 -*-
  2. from typing import Dict, Optional
  3. import numpy as np
  4. from httpx import AsyncClient
  5. from loguru import logger
  6. from app.controllers.events import q_learning_models
  7. from app.services.platform import DataPlatformService
  8. from app.services.transfer import Duoduo, SpaceInfoService
  9. from app.services.transfer import Season
  10. class QLearningCommandBuilder:
  11. """
  12. Build FCU command by Q learning net.
  13. """
  14. def __init__(self, season: Season):
  15. self.season = season
  16. if season == Season.cooling:
  17. self.model = q_learning_models.get('summer')
  18. elif season == Season.heating:
  19. self.model = q_learning_models.get('winter')
  20. else:
  21. self.model = None
  22. def get_type(self, layer: int) -> str:
  23. return self.model[0, layer][0, 0][0][0]
  24. def get_weight(self, layer: int, idx: int) -> np.ndarray:
  25. return self.model[0, layer][0, 0][1][0, idx]
  26. @staticmethod
  27. def linear(input_v: np.ndarray, weight: np.ndarray, bias: Optional[np.ndarray] = None) -> np.ndarray:
  28. y = np.dot(weight, input_v)
  29. if bias.size > 0:
  30. y += bias
  31. return y
  32. @staticmethod
  33. def relu(x: np.ndarray) -> np.ndarray:
  34. return np.maximum(x, 0)
  35. def predict_speed(self, input_v: np.ndarray) -> int:
  36. result = [input_v]
  37. for layer in range(self.model.shape[1]):
  38. if self.get_type(layer) == 'mlp' or self.get_type(layer) == 'linear':
  39. y = self.linear(result[layer], self.get_weight(layer, 0), self.get_weight(layer, 1))
  40. result.append(y)
  41. elif self.get_type(layer) == 'relu':
  42. result.append(self.relu(result[layer]))
  43. speed = np.argmax(result[-1])
  44. return int(speed)
  45. async def get_command(self, current_temperature: float, pre_temperature: float, actual_target: float) -> Dict:
  46. input_value = np.array([
  47. [(current_temperature - actual_target) / 5],
  48. [(current_temperature - pre_temperature) / 5]
  49. ])
  50. speed = self.predict_speed(input_value)
  51. if np.isnan(current_temperature) or np.isnan(pre_temperature):
  52. speed = 2
  53. if np.isnan(actual_target):
  54. speed = 0
  55. if speed == 0:
  56. on_off = 0
  57. water_on_off = 0
  58. else:
  59. on_off = 1
  60. water_on_off = 1
  61. if self.season == Season.cooling:
  62. season = 1
  63. elif self.season == Season.heating:
  64. season = 2
  65. else:
  66. season = 0
  67. command = {
  68. 'onOff': on_off,
  69. 'mode': season,
  70. 'speed': int(speed),
  71. 'temperature': actual_target if not np.isnan(actual_target) else None,
  72. 'water': water_on_off
  73. }
  74. return command
  75. @logger.catch()
  76. async def get_fcu_q_learning_control_result(project_id: str, equipment_id: str) -> Dict:
  77. async with AsyncClient() as client:
  78. duo_duo = Duoduo(client, project_id)
  79. platform = DataPlatformService(client, project_id)
  80. spaces = await duo_duo.get_space_by_equipment(equipment_id)
  81. if not spaces:
  82. logger.error(f'FCU {equipment_id} does not have space')
  83. return {}
  84. else:
  85. if len(spaces) > 1:
  86. logger.error(f'FCU {equipment_id} control more than one spaces!')
  87. transfer = SpaceInfoService(client, project_id, spaces[0].get('id'))
  88. season = await duo_duo.get_season()
  89. current_target = await transfer.get_current_temperature_target()
  90. realtime_temperature = await platform.get_realtime_temperature(spaces[0].get('id'))
  91. past_temperature = await platform.get_past_temperature(spaces[0].get('id'), 15 * 60)
  92. logger.debug(
  93. f'{spaces[0]["id"]} - {equipment_id} - '
  94. f'realtime Tdb: {realtime_temperature} - '
  95. f'pre Tdb: {past_temperature} - '
  96. f'target: {current_target}'
  97. )
  98. if season == Season.transition:
  99. command = {}
  100. else:
  101. builder = QLearningCommandBuilder(season)
  102. command = await builder.get_command(realtime_temperature, past_temperature, current_target)
  103. return command