q_learning.py 4.1 KB

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