Weather (pyobs.modules.weather)

Weather modules. TODO: write doc

MockWeather

class MockWeather(good: bool = True, sensors: dict[str, float] | None = None, **kwargs: Any)

Bases: Module, IWeather, IFitsHeaderBefore

A mock weather station for testing and simulations.

Creates a new mock weather station.

Parameters:

• good – Initial weather-good state.

• sensors – Initial sensor values, keyed by sensor value (e.g. “temp”, “humid”), overriding the defaults.

async get_fits_header_before(namespaces: list[str] | None = None, **kwargs: Any) → dict[str, FitsHeaderEntry]

Returns FITS header for the current status of this module.

Parameters:

namespaces – If given, only return FITS headers for the given namespaces.

Returns:

Dictionary containing FITS headers.

async get_sensor_value(station: str, sensor: WeatherSensors, **kwargs: Any) → WeatherSensorReading

Return value for given sensor.

Parameters:

• 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.

async is_running(**kwargs: Any) → bool

Whether a service is running.

async open() → None

Open module.

async set_good(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.

Parameters:

good – New weather-good state.

set_sensor_value(sensor: WeatherSensors, value: float) → None

Set a simulated sensor’s value, for use in tests and simulations.

Parameters:

• sensor – Sensor to set.

• value – New value.

async start(**kwargs: Any) → None

Starts a service.

async stop(**kwargs: Any) → None

Stops a service.

Weather

class Weather(url: str, system_init_time: int = 300, **kwargs: Any)

Bases: Module, IWeather, IFitsHeaderBefore

Connection to pyobs-weather.

Initialize a new pyobs-weather connector.

Parameters:

• url – URL to weather station

• system_init_time – Time in seconds the full system needs to initialize

async get_fits_header_before(namespaces: list[str] | None = None, **kwargs: Any) → dict[str, FitsHeaderEntry]

Returns FITS header for the current status of this module.

Parameters:

namespaces – If given, only return FITS headers for the given namespaces.

Returns:

Dictionary containing FITS headers.

async get_sensor_value(station: str, sensor: WeatherSensors, **kwargs: Any) → WeatherSensorReading

Return value for given sensor.

Parameters:

• station – Name of weather station to get value from.

• sensor – Name of sensor to get value from.

Returns:

Current reading for the given sensor.

async is_running(**kwargs: Any) → bool

Whether a service is running.

async open() → None

Open module.

async start(**kwargs: Any) → None

Starts a service.

async stop(**kwargs: Any) → None

Stops a service.