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

final cleanup #27

Merged
merged 1 commit into from
May 3, 2023
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
67 changes: 31 additions & 36 deletions pytrafikverket/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,59 +1,54 @@
"""Exceptions for Trafikverket."""

from typing import Any
from __future__ import annotations


class NoCameraFound(Exception):
"""Error from Sensibo api."""

def __init__(self, *args: Any) -> None:
"""Initialize the exception."""
Exception.__init__(self, *args)
"""Error found no camera."""


class MultipleCamerasFound(Exception):
"""Authentication issue from Sensibo api."""

def __init__(self, *args: Any) -> None:
"""Initialize the exception."""
Exception.__init__(self, *args)
"""Error found multiple cameras."""


class NoRouteFound(Exception):
"""Error from Sensibo api."""

def __init__(self, *args: Any) -> None:
"""Initialize the exception."""
Exception.__init__(self, *args)
"""Error found no ferry route."""


class MultipleRoutesFound(Exception):
"""Authentication issue from Sensibo api."""

def __init__(self, *args: Any) -> None:
"""Initialize the exception."""
Exception.__init__(self, *args)
"""Error found multiple ferry routes."""


class NoFerryFound(Exception):
"""Error from Sensibo api."""

def __init__(self, *args: Any) -> None:
"""Initialize the exception."""
Exception.__init__(self, *args)
"""Error found no ferry."""


class NoDeviationFound(Exception):
"""Error from Sensibo api."""

def __init__(self, *args: Any) -> None:
"""Initialize the exception."""
Exception.__init__(self, *args)
"""Error found no deviation."""


class MultipleDeviationsFound(Exception):
"""Authentication issue from Sensibo api."""
"""Error found multiple deviations."""


class NoWeatherStationFound(Exception):
"""Error found no weather station."""


class MultipleWeatherStationsFound(Exception):
"""Error found multiple weather stations."""


class NoTrainStationFound(Exception):
"""Error found no train station."""


class MultipleTrainStationsFound(Exception):
"""Error found multiple train stations."""


class NoTrainAnnouncementFound(Exception):
"""Error found no train announcement."""


def __init__(self, *args: Any) -> None:
"""Initialize the exception."""
Exception.__init__(self, *args)
class MultipleTrainAnnouncementFound(Exception):
"""Error found multiple train announcements."""
8 changes: 4 additions & 4 deletions pytrafikverket/trafikverket.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ class SortOrder(Enum):
class FieldSort:
"""What field and how to sort on it."""

def __init__(self, field: str, sort_order: SortOrder):
def __init__(self, field: str, sort_order: SortOrder) -> None:
"""Initialize the class."""
self._field = field
self._sort_order = sort_order
Expand Down Expand Up @@ -77,7 +77,7 @@ def generate_node(self, parent_node: Any) -> Any:
class OrFilter(Filter):
"""Used to create a Or filter."""

def __init__(self, filters: list[Filter]):
def __init__(self, filters: list[Filter]) -> None:
"""Initialize the class."""
self.filters = filters

Expand All @@ -92,7 +92,7 @@ def generate_node(self, parent_node: Any) -> Any:
class AndFilter(Filter):
"""Used to create a And filter."""

def __init__(self, filters: list[Filter]):
def __init__(self, filters: list[Filter]) -> None:
"""Initialize the class."""
self.filters = filters

Expand All @@ -111,7 +111,7 @@ class Trafikverket:
date_time_format = "%Y-%m-%dT%H:%M:%S.%f%z"
date_time_format_for_modified = "%Y-%m-%dT%H:%M:%S.%fZ"

def __init__(self, client_session: aiohttp.ClientSession, api_key: str):
def __init__(self, client_session: aiohttp.ClientSession, api_key: str) -> None:
"""Initialize TrafikInfo object."""
self._client_session = client_session
self._api_key = api_key
Expand Down
49 changes: 32 additions & 17 deletions pytrafikverket/trafikverket_train.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,13 @@
Trafikverket,
)

from .exceptions import (
NoTrainStationFound,
MultipleTrainStationsFound,
NoTrainAnnouncementFound,
MultipleTrainAnnouncementFound,
)

# pylint: disable=W0622, C0103


Expand All @@ -25,7 +32,9 @@ class StationInfo:

_required_fields = ["LocationSignature", "AdvertisedLocationName"]

def __init__(self, signature: str | None, name: str | None, advertised: str | None):
def __init__(
self, signature: str | None, name: str | None, advertised: str | None
) -> None:
"""Initialize StationInfo class."""
self.signature = signature
self.name = name
Expand All @@ -44,12 +53,12 @@ def from_xml_node(cls, node: Any) -> StationInfo:
class TrainStopStatus(Enum):
"""Contain the different train stop statuses."""

on_time = "scheduled to arrive on schedule"
delayed = "delayed"
canceled = "canceled"
ON_TIME = "scheduled to arrive on schedule"
DELAYED = "delayed"
CANCELED = "canceled"


class TrainStop(object):
class TrainStop:
"""Contain information about a train stop."""

_required_fields = [
Expand Down Expand Up @@ -90,20 +99,20 @@ def __init__(
def get_state(self) -> TrainStopStatus:
"""Retrieve the state of the departure."""
if self.canceled:
return TrainStopStatus.canceled
return TrainStopStatus.CANCELED
if (
self.advertised_time_at_location is not None
and self.time_at_location is not None
and self.advertised_time_at_location != self.time_at_location
):
return TrainStopStatus.delayed
return TrainStopStatus.DELAYED
if (
self.advertised_time_at_location is not None
and self.estimated_time_at_location is not None
and self.advertised_time_at_location != self.estimated_time_at_location
):
return TrainStopStatus.delayed
return TrainStopStatus.on_time
return TrainStopStatus.DELAYED
return TrainStopStatus.ON_TIME

def get_delay_time(self) -> timedelta | None:
"""Calculate the delay of a departure."""
Expand Down Expand Up @@ -154,7 +163,7 @@ def from_xml_node(cls, node: Any) -> TrainStop:
class TrafikverketTrain:
"""Class used to communicate with trafikverket's train api."""

def __init__(self, client_session: aiohttp.ClientSession, api_key: str):
def __init__(self, client_session: aiohttp.ClientSession, api_key: str) -> None:
"""Initialize TrainInfo object."""
self._api = Trafikverket(client_session, api_key)

Expand All @@ -172,9 +181,13 @@ async def async_get_train_station(self, location_name: str) -> StationInfo:
],
)
if len(train_stations) == 0:
raise ValueError("Could not find a station with the specified name")
raise NoTrainStationFound(
"Could not find a station with the specified name"
)
if len(train_stations) > 1:
raise ValueError("Found multiple stations with the specified name")
raise MultipleTrainStationsFound(
"Found multiple stations with the specified name"
)

return StationInfo.from_xml_node(train_stations[0])

Expand All @@ -194,7 +207,9 @@ async def async_search_train_stations(
],
)
if len(train_stations) == 0:
raise ValueError("Could not find a station with the specified name")
raise NoTrainStationFound(
"Could not find a station with the specified name"
)

result = []

Expand Down Expand Up @@ -258,10 +273,10 @@ async def async_get_train_stop(
)

if len(train_announcements) == 0:
raise ValueError("No TrainAnnouncement found")
raise NoTrainAnnouncementFound("No TrainAnnouncement found")

if len(train_announcements) > 1:
raise ValueError("Multiple TrainAnnouncements found")
raise MultipleTrainAnnouncementFound("Multiple TrainAnnouncements found")

train_announcement = train_announcements[0]
return TrainStop.from_xml_node(train_announcement)
Expand Down Expand Up @@ -326,10 +341,10 @@ async def async_get_next_train_stop(
)

if len(train_announcements) == 0:
raise ValueError("No TrainAnnouncement found")
raise NoTrainAnnouncementFound("No TrainAnnouncement found")

if len(train_announcements) > 1:
raise ValueError("Multiple TrainAnnouncements found")
raise MultipleTrainAnnouncementFound("Multiple TrainAnnouncements found")

train_announcement = train_announcements[0]

Expand Down
10 changes: 8 additions & 2 deletions pytrafikverket/trafikverket_weather.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
Trafikverket,
)

from .exceptions import NoWeatherStationFound, MultipleWeatherStationsFound

WIND_DIRECTION_TRANSLATION = {
"Öst": "east",
"Nordöst": "north_east",
Expand Down Expand Up @@ -186,8 +188,12 @@ async def async_get_weather(self, location_name: str) -> WeatherStationInfo:
[FieldFilter(FilterOperation.EQUAL, "Name", location_name)],
)
if len(weather_stations) == 0:
raise ValueError("Could not find a weather station with the specified name")
raise NoWeatherStationFound(
"Could not find a weather station with the specified name"
)
if len(weather_stations) > 1:
raise ValueError("Found multiple weather stations with the specified name")
raise MultipleWeatherStationsFound(
"Found multiple weather stations with the specified name"
)

return WeatherStationInfo.from_xml_node(weather_stations[0])