from __future__ import annotations
from typing import Any
from pyobs.events import BadWeatherEvent, GoodWeatherEvent
from pyobs.interfaces import FitsHeaderEntry, IFitsHeaderBefore, IWeather, WeatherSensorReading, WeatherState
from pyobs.modules import Module
from pyobs.modules.weather.weather import FITS_HEADERS, SENSOR_UNITS
from pyobs.utils.enums import WeatherSensors
DEFAULT_SENSOR_VALUES: dict[WeatherSensors, float] = {
WeatherSensors.TEMPERATURE: 15.0,
WeatherSensors.HUMIDITY: 40.0,
WeatherSensors.PRESSURE: 1013.0,
WeatherSensors.WINDDIR: 180.0,
WeatherSensors.WINDSPEED: 5.0,
WeatherSensors.RAIN: 0.0,
WeatherSensors.SKYTEMP: -20.0,
WeatherSensors.DEWPOINT: 2.0,
WeatherSensors.PARTICLES: 0.0,
WeatherSensors.SKYMAG: 21.5,
}
class MockWeather(Module, IWeather, IFitsHeaderBefore):
"""A mock weather station for testing and simulations."""
__module__ = "pyobs.modules.weather"
def __init__(self, good: bool = True, sensors: dict[str, float] | None = None, **kwargs: Any):
"""Creates a new mock weather station.
Args:
good: Initial weather-good state.
sensors: Initial sensor values, keyed by sensor value (e.g. "temp", "humid"), overriding the defaults.
"""
Module.__init__(self, **kwargs)
self._good = good
self._active = True
self._sensors: dict[WeatherSensors, float] = dict(DEFAULT_SENSOR_VALUES)
if sensors:
for key, value in sensors.items():
self._sensors[WeatherSensors(key)] = value
[docs]
async def open(self) -> None:
"""Open module."""
await Module.open(self)
if self._comm:
await self.comm.register_event(BadWeatherEvent)
await self.comm.register_event(GoodWeatherEvent)
await self._publish_state()
[docs]
async def start(self, **kwargs: Any) -> None:
"""Starts a service."""
# did status change and weather is now bad?
if not self._active and not self._good:
await self.comm.send_event(BadWeatherEvent())
self._active = True
await self._publish_state()
[docs]
async def stop(self, **kwargs: Any) -> None:
"""Stops a service."""
self._active = False
await self._publish_state()
[docs]
async def is_running(self, **kwargs: Any) -> bool:
"""Whether a service is running."""
return self._active
[docs]
async def set_good(self, good: bool) -> None:
"""Set the simulated weather-good state, for use in tests and simulations.
Fires a GoodWeatherEvent/BadWeatherEvent on change, and publishes the updated state.
Args:
good: New weather-good state.
"""
if good == self._good:
return
self._good = good
if self._active:
await self.comm.send_event(GoodWeatherEvent() if good else BadWeatherEvent())
await self._publish_state()
[docs]
def set_sensor_value(self, sensor: WeatherSensors, value: float) -> None:
"""Set a simulated sensor's value, for use in tests and simulations.
Args:
sensor: Sensor to set.
value: New value.
"""
self._sensors[sensor] = value
async def _publish_state(self) -> None:
is_good = True if not self._active else self._good
await self.comm.set_state(IWeather, WeatherState(good=is_good, readings=self._get_readings()))
def _get_readings(self) -> list[WeatherSensorReading]:
return [
WeatherSensorReading(sensor=sensor, value=value, unit=SENSOR_UNITS[sensor])
for sensor, value in self._sensors.items()
]
[docs]
async def get_sensor_value(self, station: str, sensor: WeatherSensors, **kwargs: Any) -> WeatherSensorReading:
"""Return value for given sensor.
Args:
station: Name of weather station to get value from (ignored, there is only one simulated station).
sensor: Name of sensor to get value from.
Returns:
Current reading for the given sensor.
"""
if sensor not in self._sensors:
raise ValueError(f"Unknown sensor: {sensor}")
return WeatherSensorReading(sensor=sensor, value=self._sensors[sensor], unit=SENSOR_UNITS[sensor])
__all__ = ["MockWeather"]