From af6ae1651badf48146b816d9189a29a9429c70b7 Mon Sep 17 00:00:00 2001 From: Bjarne Riis Date: Wed, 27 Dec 2023 19:17:08 +0100 Subject: [PATCH] Fixing mypy recommendations --- async_test_module.py | 3 +- pyweatherflow_forecast/data.py | 40 +++++++++-------- pyweatherflow_forecast/wffcst_lib.py | 67 ++++++++++++++-------------- 3 files changed, 56 insertions(+), 54 deletions(-) diff --git a/async_test_module.py b/async_test_module.py index 77486e8..2c86add 100644 --- a/async_test_module.py +++ b/async_test_module.py @@ -1,3 +1,4 @@ +# ruff: noqa: F401 """This module is only used to run some realtime data tests using the async functions, while developing the module. Create a .env file and add STATION_ID with the id of your station and API_TOKEN with the personal Token. @@ -13,9 +14,9 @@ from pyweatherflow_forecast import ( WeatherFlow, + WeatherFlowStationData, WeatherFlowForecastData, WeatherFlowSensorData, - WeatherFlowStationData, ) _LOGGER = logging.getLogger(__name__) diff --git a/pyweatherflow_forecast/data.py b/pyweatherflow_forecast/data.py index b8c6f33..b1f92b6 100644 --- a/pyweatherflow_forecast/data.py +++ b/pyweatherflow_forecast/data.py @@ -26,8 +26,8 @@ def __init__( wind_bearing: int, wind_gust_speed: float, wind_speed: float, - forecast_daily: WeatherFlowForecastDaily = None, - forecast_hourly: WeatherFlowForecastHourly = None, + forecast_daily: WeatherFlowForecastDaily | None = None, + forecast_hourly: WeatherFlowForecastHourly | None = None, ) -> None: """Dataset constructor.""" self._datetime = datetime @@ -119,7 +119,7 @@ def datetimestamptime(self) -> int: return self._timestamp @property - def forecast_daily(self) -> WeatherFlowForecastDaily: + def forecast_daily(self) -> WeatherFlowForecastDaily | None: """Forecast List.""" return self._forecast_daily @@ -129,7 +129,7 @@ def forecast_daily(self, new_forecast): self._forecast_daily = new_forecast @property - def forecast_hourly(self) -> WeatherFlowForecastHourly: + def forecast_hourly(self) -> WeatherFlowForecastHourly | None: """Forecast List.""" return self._forecast_hourly @@ -329,7 +329,7 @@ def timestamp(self) -> int: class WeatherFlowDeviceData: """Class to hold device data.""" - # pylint: disable=R0913, R0902, R0914 + # pylint: disable=R0913, R0902, R0914 def __init__( self, device_id: int, @@ -357,11 +357,12 @@ def precipitation_type(self) -> int: return self._precipitation_type @property - def battery(self) -> int: + def battery(self) -> float: """Battery (%).""" if self._voltage is None: return None + _percent: float if self._voltage > 2.80: _percent = 100 elif self._voltage < 1.80: @@ -369,7 +370,7 @@ def battery(self) -> int: else: _percent = (self._voltage - 1.8) * 100 - return _percent + return float(_percent) class WeatherFlowStationData: """Class to hold station data.""" @@ -447,14 +448,14 @@ def __init__( lightning_strike_count_last_1hr: int, lightning_strike_count_last_3hr: int, lightning_strike_last_distance: int, - lightning_strike_last_epoch: datetime.timestamp, + lightning_strike_last_epoch: float, precip: float, precip_accum_last_1hr: float, precip_accum_local_day: float, precip_accum_local_yesterday: float, precip_minutes_local_day: int, precip_minutes_local_yesterday: int, - precipitation_type: int, + precipitation_type: int | None, pressure_trend: str, relative_humidity: int, sea_level_pressure: float, @@ -462,7 +463,7 @@ def __init__( station_pressure: float, timestamp: int, uv: float, - voltage: float, + voltage: float | None, wet_bulb_globe_temperature: float, wet_bulb_temperature: float, wind_avg: float, @@ -546,11 +547,12 @@ def barometric_pressure(self) -> float: return self._barometric_pressure @property - def battery(self) -> int: + def battery(self) -> float | None: """Battery (%).""" if self._voltage is None: return None + _percent: float if self._voltage > 2.80: _percent = 100 elif self._voltage < 1.80: @@ -561,7 +563,7 @@ def battery(self) -> int: return _percent @property - def beaufort(self) -> int: + def beaufort(self) -> int | None: """Beaufort Value.""" if self._wind_avg is None: return None @@ -588,7 +590,7 @@ def beaufort(self) -> int: return None @property - def beaufort_description(self) -> str: + def beaufort_description(self) -> str | None: """Beaufort Textual Description.""" if self._wind_avg is None: @@ -699,12 +701,12 @@ def lightning_strike_last_distance(self) -> int: return self._lightning_strike_last_distance @property - def lightning_strike_last_epoch(self) -> datetime.timestamp: + def lightning_strike_last_epoch(self) -> float: """Last lightning strike epoch time.""" return self._lightning_strike_last_epoch @property - def power_save_mode(self) -> int: + def power_save_mode(self) -> int | None: """Power Save Mode (Tempest devices).""" if self._voltage is None or self._solar_radiation is None: return None @@ -763,7 +765,7 @@ def precip_accum_local_yesterday(self) -> float: return self._precip_accum_local_yesterday @property - def precip_intensity(self) -> str: + def precip_intensity(self) -> str | None: """Return a string with precipitation intensity.""" if self._precip is None: return None @@ -814,7 +816,7 @@ def precip_minutes_local_yesterday_final(self) -> int: return self._precip_minutes_local_yesterday_final @property - def precip_type(self) -> str: + def precip_type(self) -> int | None: """Return precipitation type.""" return self._precipitation_type @@ -859,7 +861,7 @@ def uv(self) -> float: return self._uv @property - def uv_description(self) -> str: + def uv_description(self) -> str | None: """UV value description.""" if self._uv is None: return None @@ -900,7 +902,7 @@ def visibility(self) -> float: return float(_max_visibility * _percent_reduction) @property - def voltage(self) -> float: + def voltage(self) -> float | None: """Return voltage of device.""" return self._voltage diff --git a/pyweatherflow_forecast/wffcst_lib.py b/pyweatherflow_forecast/wffcst_lib.py index 989de89..5963908 100644 --- a/pyweatherflow_forecast/wffcst_lib.py +++ b/pyweatherflow_forecast/wffcst_lib.py @@ -68,7 +68,7 @@ class WeatherFlowAPI(WeatherFlowAPIBase): def __init__(self) -> None: """Init the API with or without session.""" - self.session = None + self.session: aiohttp.ClientSession | None = None def api_request(self, url: str) -> dict[str, Any]: """Return data from API.""" @@ -127,7 +127,7 @@ def __init__( self, station_id: int, api_token: str, - session: aiohttp.ClientSession = None, + session: aiohttp.ClientSession | None = None, elevation = None, api: WeatherFlowAPIBase = WeatherFlowAPI(), ) -> None: @@ -136,35 +136,35 @@ def __init__( self._api_token = api_token self._elevation = elevation self._api = api - self._device_id = None + self._device_id: int | None = None self._tempest_device = False - self._json_data = None - self._station_data: WeatherFlowStationData = None - self._device_data: WeatherFlowDeviceData = None - self._voltage: float = None + self._json_data: dict[Any, Any] | None = None + self._station_data: WeatherFlowStationData | None = None + self._device_data: WeatherFlowDeviceData | None = None + self._voltage: float | None = None if session: self._api.session = session - def get_forecast(self) -> list[WeatherFlowForecastData]: + def get_forecast(self) -> WeatherFlowForecastData: """Return list of forecasts. The first in list are the current one.""" api_url = f"{WEATHERFLOW_FORECAST_URL}{self._station_id}&token={self._api_token}" self._json_data = self._api.api_request(api_url) return _get_forecast(self._json_data) - def get_station(self) -> list[WeatherFlowStationData]: + def get_station(self) -> WeatherFlowStationData: """Return list of station information.""" station_url = f"{WEATHERFLOW_STATION_URL}{self._station_id}?token={self._api_token}" json_data = self._api.api_request(station_url) return _get_station(json_data) - def fetch_sensor_data(self, voltage: float = None) -> list[WeatherFlowSensorData]: + def fetch_sensor_data(self) -> WeatherFlowSensorData: """Return list of sensor data.""" - device_data = None - sensor_data = None + device_data: WeatherFlowDeviceData | None = None + sensor_data: WeatherFlowSensorData | None = None if self._device_id is None and not self._tempest_device: station_url = f"{WEATHERFLOW_STATION_URL}{self._station_id}?token={self._api_token}" @@ -177,7 +177,7 @@ def fetch_sensor_data(self, voltage: float = None) -> list[WeatherFlowSensorData _device_id = station_data.device_id device_url = f"{WEATHERFLOW_DEVICE_URL}{_device_id}?token={self._api_token}" json_device_data = self._api.api_request(device_url) - device_data: WeatherFlowDeviceData = _get_device_data(json_device_data, _device_id) + device_data = _get_device_data(json_device_data, _device_id) if device_data is not None or not self._tempest_device: _precipitation_type = device_data.precipitation_type if self._tempest_device else None @@ -188,10 +188,9 @@ def fetch_sensor_data(self, voltage: float = None) -> list[WeatherFlowSensorData return sensor_data - async def async_fetch_sensor_data(self) -> list[WeatherFlowSensorData]: + async def async_fetch_sensor_data(self) -> WeatherFlowSensorData: """Return sensor data from API.""" - device_data = None - sensor_data = None + device_data: WeatherFlowDeviceData | None = None if self._device_id is None and not self._tempest_device: station_url = f"{WEATHERFLOW_STATION_URL}{self._station_id}?token={self._api_token}" @@ -204,25 +203,25 @@ async def async_fetch_sensor_data(self) -> list[WeatherFlowSensorData]: if self._device_id is not None: device_url = f"{WEATHERFLOW_DEVICE_URL}{self._device_id}?token={self._api_token}" json_device_data = await self._api.async_api_request(device_url) - device_data: WeatherFlowDeviceData = _get_device_data(json_device_data, self._device_id) + device_data = _get_device_data(json_device_data, self._device_id) if device_data is not None or not self._tempest_device: - _precipitation_type = device_data.precipitation_type if self._tempest_device else None - _voltage = device_data.voltage if self._tempest_device else None + _precipitation_type: int | None = device_data.precipitation_type if self._tempest_device else None + _voltage: float | None = device_data.voltage if self._tempest_device else None api_url = f"{WEATHERFLOW_SENSOR_URL}{self._station_id}?token={self._api_token}" json_data = await self._api.async_api_request(api_url) - sensor_data = _get_sensor_data(json_data, self._elevation, _voltage, _precipitation_type, self._station_name) + sensor_data: WeatherFlowSensorData = _get_sensor_data(json_data, self._elevation, _voltage, _precipitation_type, self._station_name) return sensor_data - async def async_get_forecast(self) -> list[WeatherFlowForecastData]: + async def async_get_forecast(self) -> WeatherFlowForecastData: """Return list of forecasts. The first in list are the current one.""" api_url = f"{WEATHERFLOW_FORECAST_URL}{self._station_id}&token={self._api_token}" self._json_data = await self._api.async_api_request(api_url) return _get_forecast(self._json_data) - async def async_get_station(self) -> list[WeatherFlowStationData]: + async def async_get_station(self) -> WeatherFlowStationData: """Return list with Station information.""" api_url = f"{WEATHERFLOW_STATION_URL}{self._station_id}?token={self._api_token}" @@ -255,7 +254,7 @@ def _calced_day_values(day_number, hourly_data) -> dict[str, Any]: # pylint: disable=R0914, R0912, W0212, R0915 -def _get_forecast(api_result: dict) -> list[WeatherFlowForecastData]: +def _get_forecast(api_result: dict) -> WeatherFlowForecastData: """Return WeatherFlowForecast list from API.""" # Get Current Conditions @@ -278,7 +277,7 @@ def _get_forecast(api_result: dict) -> list[WeatherFlowForecastData]: wind_bearing = _calc_values["wind_bearing"] wind_speed = _calc_values["wind_speed"] - forecast = WeatherFlowForecastDaily( + forecast_d = WeatherFlowForecastDaily( valid_time, timestamp, temperature, @@ -290,7 +289,7 @@ def _get_forecast(api_result: dict) -> list[WeatherFlowForecastData]: wind_bearing, wind_speed, ) - forecasts_daily.append(forecast) + forecasts_daily.append(forecast_d) current_conditions.forecast_daily = forecasts_daily @@ -311,7 +310,7 @@ def _get_forecast(api_result: dict) -> list[WeatherFlowForecastData]: wind_gust_speed = item.get("wind_gust", None) wind_bearing = item.get("wind_direction", None) - forecast = WeatherFlowForecastHourly( + forecast_h = WeatherFlowForecastHourly( valid_time, timestamp, temperature, @@ -327,7 +326,7 @@ def _get_forecast(api_result: dict) -> list[WeatherFlowForecastData]: wind_speed, uv_index, ) - forecasts_hourly.append(forecast) + forecasts_hourly.append(forecast_h) current_conditions.forecast_hourly = forecasts_hourly @@ -335,7 +334,7 @@ def _get_forecast(api_result: dict) -> list[WeatherFlowForecastData]: # pylint: disable=R0914, R0912, W0212, R0915 -def _get_forecast_current(api_result: dict) -> list[WeatherFlowForecastData]: +def _get_forecast_current(api_result: dict) -> WeatherFlowForecastData: """Return WeatherFlowForecast list from API.""" item = api_result["current_conditions"] @@ -375,7 +374,7 @@ def _get_forecast_current(api_result: dict) -> list[WeatherFlowForecastData]: # pylint: disable=R0914, R0912, W0212, R0915 -def _get_station(api_result: dict) -> list[WeatherFlowStationData]: +def _get_station(api_result: dict) -> WeatherFlowStationData: """Return WeatherFlowForecast list from API.""" item = api_result["stations"][0] @@ -384,9 +383,9 @@ def _get_station(api_result: dict) -> list[WeatherFlowStationData]: latitude = item.get("latitude", None) longitude = item.get("longitude", None) timezone = item.get("timezone", None) - device_id = None - firmware_revision = None - serial_number = None + device_id: int + firmware_revision: str + serial_number: str for device in item["devices"]: if device.get("device_type", None) == "ST": device_id = device.get("device_id", None) @@ -408,7 +407,7 @@ def _get_station(api_result: dict) -> list[WeatherFlowStationData]: # pylint: disable=R0914, R0912, W0212, R0915 -def _get_sensor_data(api_result: dict, elevation: float, voltage: float, precipitation_type: int, station_name: str) -> list[WeatherFlowSensorData]: +def _get_sensor_data(api_result: dict, elevation: float, voltage: float | None, precipitation_type: int | None, station_name: str) -> WeatherFlowSensorData: """Return WeatherFlowSensorData list from API.""" _LOGGER.debug("ELEVATION: %s", elevation) @@ -500,7 +499,7 @@ def _get_sensor_data(api_result: dict, elevation: float, voltage: float, precipi return sensor_data # pylint: disable=R0914, R0912, W0212, R0915 -def _get_device_data(api_result: dict, device_id: int) -> float: +def _get_device_data(api_result: dict, device_id: int) -> WeatherFlowDeviceData: """Return WeatherFlow Device Voltage from API.""" precipitation_type = None if api_result is None else api_result["obs"][0][13]