import math import random from datetime import datetime, timezone from bots.base import BaseBot class EnvSensorBot(BaseBot): """Temperature + humidity sensor for a single rack.""" interval = 30 def __init__(self, site_id: str, room_id: str, rack_id: str) -> None: super().__init__() self.site_id = site_id self.room_id = room_id self.rack_id = rack_id # Racks at the far end of a row run slightly warmer rack_num = int(rack_id[-2:]) if rack_id[-2:].isdigit() else 1 self._base_temp = 21.5 + (rack_num * 0.15) self._base_humidity = 44.0 + random.uniform(-2, 2) def get_topic(self) -> str: return f"bms/{self.site_id}/{self.room_id}/{self.rack_id}/env" def get_payload(self) -> dict: hour = datetime.now(timezone.utc).hour # Natural day/night cycle — slightly warmer during working hours day_factor = 1.0 + 0.04 * math.sin(math.pi * (hour - 6) / 12) temp = ( self._base_temp * day_factor + math.sin(self._scenario_step * 0.3) * 0.2 + random.gauss(0, 0.15) ) humidity = self._base_humidity + math.sin(self._scenario_step * 0.2) * 1.5 + random.gauss(0, 0.3) # Scenarios if self._scenario == "COOLING_FAILURE": # CRAC offline — racks heat up rapidly temp += min(self._scenario_step * 0.4, 12.0) elif self._scenario == "HIGH_TEMPERATURE": # Gradual ambient rise (e.g. summer overload) — slower, caps lower temp += min(self._scenario_step * 0.2, 6.0) elif self._scenario == "HUMIDITY_SPIKE": # Condensation / humidifier fault humidity += min(self._scenario_step * 1.5, 28.0) elif self._scenario == "SLOW_BURN": # Dirty filter starving airflow — temp and humidity both climb together temp += min(self._scenario_step * 0.15, 7.0) humidity += min(self._scenario_step * 0.8, 22.0) return { "temperature": round(max(18.0, temp), 2), "humidity": round(max(30.0, min(80.0, humidity)), 1), }