Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use StrEnum #6

Merged
merged 2 commits into from
Apr 23, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 31 additions & 55 deletions imgw_pib/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,36 +7,9 @@

from aiohttp import ClientSession

from .const import (
API_HUMIDITY,
API_HYDROLOGICAL_ENDPOINT,
API_MEASUREMENT_DATE,
API_MEASUREMENT_TIME,
API_PRECIPITATION,
API_PRESSURE,
API_RIVER,
API_STATION,
API_STATION_ID,
API_TEMPERATURE,
API_WATER_LEVEL,
API_WATER_LEVEL_MEASUREMENT_DATE,
API_WATER_TEMPERATURE,
API_WATER_TEMPERATURE_MEASUREMENT_DATE,
API_WEATHER_ENDPOINT,
API_WIND_DIRECTION,
API_WIND_SPEED,
HEADERS,
TIMEOUT,
UNIT_CELSIUS,
UNIT_CENTIMETERS,
UNIT_DEGREE,
UNIT_HPA,
UNIT_METERS_PER_SECOND,
UNIT_MILLIMETERS,
UNIT_PERCENT,
)
from .const import API_HYDROLOGICAL_ENDPOINT, API_WEATHER_ENDPOINT, HEADERS, TIMEOUT
from .exceptions import ApiError
from .model import HydrologicalData, SensorData, WeatherData
from .model import ApiNames, HydrologicalData, SensorData, Units, WeatherData

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -106,7 +79,8 @@ async def update_weather_stations(self: Self) -> None:
stations_data = await self._http_request(url)

self._weather_station_list = {
station[API_STATION_ID]: station[API_STATION] for station in stations_data
station[ApiNames.STATION_ID]: station[ApiNames.STATION]
for station in stations_data
}

async def get_weather_data(self: Self) -> WeatherData:
Expand All @@ -128,7 +102,9 @@ async def update_hydrological_stations(self: Self) -> None:
stations_data = await self._http_request(url)

self._hydrological_station_list = {
station[API_STATION_ID]: f"{station[API_STATION]} ({station[API_RIVER]})"
station[
ApiNames.STATION_ID
]: f"{station[ApiNames.STATION]} ({station[ApiNames.RIVER]})"
for station in stations_data
}

Expand Down Expand Up @@ -162,44 +138,44 @@ async def _http_request(self: Self, url: str) -> Any: # noqa: ANN401

def _parse_weather_data(self: Self, data: dict[str, Any]) -> WeatherData:
"""Parse weather data."""
temperature = data[API_TEMPERATURE]
temperature = data[ApiNames.TEMPERATURE]
temperature_sensor = SensorData(
name="Temperature",
value=float(temperature) if temperature is not None else None,
unit=UNIT_CELSIUS if temperature is not None else None,
unit=Units.CELSIUS.value if temperature is not None else None,
)
humidity = data[API_HUMIDITY]
humidity = data[ApiNames.HUMIDITY]
humidity_sensor = SensorData(
name="Humidity",
value=float(humidity) if humidity is not None else None,
unit=UNIT_PERCENT if humidity is not None else None,
unit=Units.PERCENT.value if humidity is not None else None,
)
wind_speed = data[API_WIND_SPEED]
wind_speed = data[ApiNames.WIND_SPEED]
wind_speed_sensor = SensorData(
name="Wind Speed",
value=float(wind_speed) if wind_speed is not None else None,
unit=UNIT_METERS_PER_SECOND if wind_speed is not None else None,
unit=Units.METERS_PER_SECOND.value if wind_speed is not None else None,
)
wind_direction = data[API_WIND_DIRECTION]
wind_direction = data[ApiNames.WIND_DIRECTION]
wind_direction_sensor = SensorData(
name="Wind Direction",
value=float(wind_direction) if wind_direction is not None else None,
unit=UNIT_DEGREE if wind_direction is not None else None,
unit=Units.DEGREE.value if wind_direction is not None else None,
)
precipitation = data[API_PRECIPITATION]
precipitation = data[ApiNames.PRECIPITATION]
precipitation_sensor = SensorData(
name="Precipitation",
value=float(precipitation) if precipitation is not None else None,
unit=UNIT_MILLIMETERS if precipitation is not None else None,
unit=Units.MILLIMETERS.value if precipitation is not None else None,
)
pressure = data[API_PRESSURE]
pressure = data[ApiNames.PRESSURE]
pressure_sensor = SensorData(
name="Pressure",
value=float(pressure) if pressure is not None else None,
unit=UNIT_HPA if pressure is not None else None,
unit=Units.HPA.value if pressure is not None else None,
)
measurement_date = self._get_datetime(
f"{data[API_MEASUREMENT_DATE]} {data[API_MEASUREMENT_TIME]}",
f"{data[ApiNames.MEASUREMENT_DATE]} {data[ApiNames.MEASUREMENT_TIME]}",
"%Y-%m-%d %H",
)

Expand All @@ -210,14 +186,14 @@ def _parse_weather_data(self: Self, data: dict[str, Any]) -> WeatherData:
wind_speed=wind_speed_sensor,
wind_direction=wind_direction_sensor,
precipitation=precipitation_sensor,
station=data[API_STATION],
station_id=data[API_STATION_ID],
station=data[ApiNames.STATION],
station_id=data[ApiNames.STATION_ID],
measurement_date=measurement_date,
)

def _parse_hydrological_data(self: Self, data: dict[str, Any]) -> HydrologicalData:
"""Parse hydrological data."""
water_level = data[API_WATER_LEVEL]
water_level = data[ApiNames.WATER_LEVEL]

if water_level is None:
msg = "Invalid water level value"
Expand All @@ -226,29 +202,29 @@ def _parse_hydrological_data(self: Self, data: dict[str, Any]) -> HydrologicalDa
water_level_sensor = SensorData(
name="Water Level",
value=float(water_level) if water_level is not None else None,
unit=UNIT_CENTIMETERS if water_level is not None else None,
unit=Units.CENTIMETERS.value if water_level is not None else None,
)
water_level_measurement_date = self._get_datetime(
data[API_WATER_LEVEL_MEASUREMENT_DATE],
data[ApiNames.WATER_LEVEL_MEASUREMENT_DATE],
"%Y-%m-%d %H:%M:%S",
)
water_temperature = data[API_WATER_TEMPERATURE]
water_temperature = data[ApiNames.WATER_TEMPERATURE]
water_temperature_sensor = SensorData(
name="Water Temperature",
value=float(water_temperature) if water_temperature is not None else None,
unit=UNIT_CELSIUS if water_temperature is not None else None,
unit=Units.CELSIUS.value if water_temperature is not None else None,
)
water_temperature_measurement_date = self._get_datetime(
data[API_WATER_TEMPERATURE_MEASUREMENT_DATE],
data[ApiNames.WATER_TEMPERATURE_MEASUREMENT_DATE],
"%Y-%m-%d %H:%M:%S",
)

return HydrologicalData(
water_level=water_level_sensor,
water_temperature=water_temperature_sensor,
station=data[API_STATION],
river=data[API_RIVER],
station_id=data[API_STATION_ID],
station=data[ApiNames.STATION],
river=data[ApiNames.RIVER],
station_id=data[ApiNames.STATION_ID],
water_level_measurement_date=water_level_measurement_date,
water_temperature_measurement_date=water_temperature_measurement_date,
)
Expand Down
24 changes: 0 additions & 24 deletions imgw_pib/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,27 +8,3 @@

HEADERS = {"Content-Type": "application/json"}
TIMEOUT = ClientTimeout(total=20)

API_HUMIDITY = "wilgotnosc_wzgledna"
API_MEASUREMENT_DATE = "data_pomiaru"
API_MEASUREMENT_TIME = "godzina_pomiaru"
API_PRECIPITATION = "suma_opadu"
API_PRESSURE = "cisnienie"
API_RIVER = "rzeka"
API_STATION = "stacja"
API_STATION_ID = "id_stacji"
API_TEMPERATURE = "temperatura"
API_WATER_LEVEL = "stan_wody"
API_WATER_LEVEL_MEASUREMENT_DATE = "stan_wody_data_pomiaru"
API_WATER_TEMPERATURE = "temperatura_wody"
API_WATER_TEMPERATURE_MEASUREMENT_DATE = "temperatura_wody_data_pomiaru"
API_WIND_DIRECTION = "kierunek_wiatru"
API_WIND_SPEED = "predkosc_wiatru"

UNIT_CELSIUS = "°C"
UNIT_CENTIMETERS = "cm"
UNIT_DEGREE = "°"
UNIT_HPA = "hPa"
UNIT_METERS_PER_SECOND = "m/s"
UNIT_MILLIMETERS = "mm"
UNIT_PERCENT = "%"
33 changes: 33 additions & 0 deletions imgw_pib/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum


@dataclass
Expand Down Expand Up @@ -44,3 +45,35 @@ class HydrologicalData(ImgwPibData):
station_id: str
water_level_measurement_date: datetime | None
water_temperature_measurement_date: datetime | None


class ApiNames(StrEnum):
"""Names type for API."""

HUMIDITY = "wilgotnosc_wzgledna"
MEASUREMENT_DATE = "data_pomiaru"
MEASUREMENT_TIME = "godzina_pomiaru"
PRECIPITATION = "suma_opadu"
PRESSURE = "cisnienie"
RIVER = "rzeka"
STATION = "stacja"
STATION_ID = "id_stacji"
TEMPERATURE = "temperatura"
WATER_LEVEL = "stan_wody"
WATER_LEVEL_MEASUREMENT_DATE = "stan_wody_data_pomiaru"
WATER_TEMPERATURE = "temperatura_wody"
WATER_TEMPERATURE_MEASUREMENT_DATE = "temperatura_wody_data_pomiaru"
WIND_DIRECTION = "kierunek_wiatru"
WIND_SPEED = "predkosc_wiatru"


class Units(StrEnum):
"""Units."""

CELSIUS = "°C"
CENTIMETERS = "cm"
DEGREE = "°"
HPA = "hPa"
METERS_PER_SECOND = "m/s"
MILLIMETERS = "mm"
PERCENT = "%"
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

PROJECT_DIR = Path(__file__).parent.resolve()
README_FILE = PROJECT_DIR / "README.md"
VERSION = "0.0.3"
VERSION = "0.0.4"

setup(
name="imgw_pib",
Expand Down
12 changes: 4 additions & 8 deletions tests/test_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,9 @@
from syrupy import SnapshotAssertion

from imgw_pib import ImgwPib
from imgw_pib.const import (
API_HYDROLOGICAL_ENDPOINT,
API_WATER_LEVEL,
API_WATER_LEVEL_MEASUREMENT_DATE,
API_WEATHER_ENDPOINT,
)
from imgw_pib.const import API_HYDROLOGICAL_ENDPOINT, API_WEATHER_ENDPOINT
from imgw_pib.exceptions import ApiError
from imgw_pib.model import ApiNames


@pytest.mark.asyncio()
Expand Down Expand Up @@ -198,7 +194,7 @@ async def test_invalid_water_level_value(
"""Test invalid water level value."""
session = aiohttp.ClientSession()

hydrological_station[API_WATER_LEVEL] = None
hydrological_station[ApiNames.WATER_LEVEL] = None

with aioresponses() as session_mock:
session_mock.get(API_HYDROLOGICAL_ENDPOINT, payload=hydrological_stations)
Expand All @@ -224,7 +220,7 @@ async def test_invalid_date(
"""Test invalid water level value."""
session = aiohttp.ClientSession()

hydrological_station[API_WATER_LEVEL_MEASUREMENT_DATE] = None
hydrological_station[ApiNames.WATER_LEVEL_MEASUREMENT_DATE] = None

with aioresponses() as session_mock:
session_mock.get(API_HYDROLOGICAL_ENDPOINT, payload=hydrological_stations)
Expand Down