q_learning.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124
  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 EquipmentInfoService, 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. # actual_target = np.mean(np.array(target))
  47. input_value = np.array([
  48. [(current_temperature - actual_target) / 5],
  49. [(current_temperature - pre_temperature) / 5]
  50. ])
  51. speed = self.predict_speed(input_value)
  52. if np.isnan(current_temperature) or np.isnan(pre_temperature):
  53. speed = 2
  54. if actual_target == 0.0:
  55. speed = 0
  56. if speed == 0:
  57. on_off = 0
  58. water_on_off = 0
  59. else:
  60. on_off = 1
  61. water_on_off = 1
  62. if self.season == Season.cooling:
  63. season = 1
  64. elif self.season == Season.heating:
  65. season = 2
  66. else:
  67. season = 0
  68. command = {
  69. 'onOff': on_off,
  70. 'mode': season,
  71. 'speed': int(speed),
  72. 'temperature': actual_target,
  73. 'water': water_on_off
  74. }
  75. return command
  76. @logger.catch()
  77. async def get_fcu_q_learning_control_result(project_id: str, equipment_id: str) -> Dict:
  78. async with AsyncClient() as client:
  79. duo_duo = EquipmentInfoService(client, project_id)
  80. platform = DataPlatformService(client, project_id)
  81. spaces = await duo_duo.get_space_by_equipment(equipment_id)
  82. if not spaces:
  83. logger.error(f'FCU {equipment_id} does not have space')
  84. return {}
  85. else:
  86. if len(spaces) > 1:
  87. logger.error(f'FCU {equipment_id} control more than one spaces!')
  88. transfer = SpaceInfoService(client, project_id, spaces[0].get('id'))
  89. season = await transfer.get_season()
  90. current_target = await transfer.get_current_temperature_target()
  91. realtime_temperature = await platform.get_realtime_temperature(spaces[0].get('id'))
  92. past_temperature = await platform.get_past_temperature(spaces[0].get('id'), 15 * 60)
  93. logger.debug(
  94. f'{spaces[0]["id"]} - {equipment_id} - '
  95. f'realtime Tdb: {realtime_temperature} - '
  96. f'pre Tdb: {past_temperature} - '
  97. f'target: {current_target}'
  98. )
  99. builder = QLearningCommandBuilder(season)
  100. command = await builder.get_command(realtime_temperature, past_temperature, current_target)
  101. return command