Stop service

`StopNotFoundError (ValueError)`

Custom exception for stop not found.

Source code in now8_api/service/stop_service.py

class StopNotFoundError(ValueError):
    """Custom exception for stop not found."""

    def __init__(self, stop_id: str) -> None:
        """Initialize ValueError with custom message.

        Arguments:
            stop_id: Stop ID that was not found.
        """
        super().__init__(f'Stop "{stop_id}" not found.')

`init(self, stop_id)` `special`

Initialize ValueError with custom message.

Parameters:

Name	Type	Description	Default
`stop_id`	`str`	Stop ID that was not found.	required

Source code in now8_api/service/stop_service.py

def __init__(self, stop_id: str) -> None:
    """Initialize ValueError with custom message.

    Arguments:
        stop_id: Stop ID that was not found.
    """
    super().__init__(f'Stop "{stop_id}" not found.')

`StopService (Service)` `pydantic-model`

Service base class.

Attributes:

Name	Type	Description
`city_data`	`CityData`	CityData instance for the city.

Source code in now8_api/service/stop_service.py

class StopService(Service):
    """Service base class.

    Attributes:
        city_data: CityData instance for the city.
    """

    city_data: CityData = CITY_DATA_DICT[CITY]

    async def all_stops(
        self,
    ) -> Dict[str, Dict[str, Union[str, float, dict]]]:
        """Return all the stops of the city.

        Returns:
            List of dictionaries with the stop ID, transport type, way,
            name, coordinates and zone of each stop.
        """
        query_result: List[tuple] = await self.sql_engine.execute_query(
            str(_all_stops_query)
        )

        result: Dict[str, Dict[str, Any]] = {}
        for row in query_result:
            stop_id: str = row[0]
            stop_code: str = row[1]
            stop_name: str = row[2]
            stop_lat: float = row[3]
            stop_lon: float = row[4]
            stop_zone: str = row[5]

            if stop_id not in result:
                stop = Stop(
                    id=stop_id,
                    code=stop_code,
                    name=stop_name,
                    coordinates=Coordinates(
                        latitude=stop_lat, longitude=stop_lon
                    ),
                    zone=stop_zone,
                )

                result[stop.id] = {
                    "id": stop.id,
                    "code": stop.code,
                    "name": stop.name,
                    "longitude": stop.coordinates.longitude,
                    "latitude": stop.coordinates.latitude,
                    "zone": stop.zone,
                    "route_ways": [],
                }

            route_id: str = row[6]
            route_way: int = row[7]

            result[stop_id]["route_ways"].append(
                {"id": route_id, "way": route_way}
            )

        return result

    async def stop_info(
        self, stop_id: str
    ) -> Dict[str, Union[str, float, list]]:
        """Return the stop information.

        Arguments:
            stop_id: Stop identifier.

        Returns:
            Dictionary with the stop ID, transport type, way, name,
                coordinates and zone.

        Raises:
            StopNotFoundError: If the `stop_id` does not match any stop.
        """
        query_result: List[tuple] = await self.sql_engine.execute_query(
            str(_all_stops_query.where(_table_stops.stop_id == stop_id))
        )

        if len(query_result) < 1:
            raise StopNotFoundError(stop_id=stop_id)

        result: Dict[str, Any] = {}
        for row in query_result:
            stop_code: str = row[1]
            stop_name: str = row[2]
            stop_lat: float = row[3]
            stop_lon: float = row[4]
            stop_zone: str = row[5]

            if result == {}:
                stop = Stop(
                    id=stop_id,
                    code=stop_code,
                    name=stop_name,
                    coordinates=Coordinates(
                        latitude=stop_lat, longitude=stop_lon
                    ),
                    zone=stop_zone,
                )

                result = {
                    "id": stop.id,
                    "code": stop.code,
                    "name": stop.name,
                    "longitude": stop.coordinates.longitude,
                    "latitude": stop.coordinates.latitude,
                    "zone": stop.zone,
                    "route_ways": [],
                }

            route_id: str = row[6]
            route_way: int = row[7]

            result["route_ways"].append({"id": route_id, "way": route_way})

        return result

    async def stop_estimation(self, stop_id: str) -> List[Dict[str, dict]]:
        """Return ETA for the next vehicles to the stop.

        Arguments:
            stop_id: Stop identifier.

        Returns:
            ETA for the next vehicles to the stop.
        """
        stop = Stop(id=stop_id, transport_type=TransportType.INTERCITY_BUS)

        estimations = await self.city_data.get_estimations(stop)

        result: List[Dict[str, dict]] = [
            {
                "vehicle": {
                    "id": v_e.vehicle.id,
                    "route_way": {
                        "id": v_e.vehicle.route_id,
                        "way": v_e.vehicle.route_way.value
                        if v_e.vehicle.route_way is not None
                        else None,
                    },
                    "destination_stop": {
                        "id": v_e.vehicle.destination_stop.id,
                        "code": v_e.vehicle.destination_stop.code,
                        "name": v_e.vehicle.destination_stop.name,
                    },
                },
                "estimation": {
                    "estimation": v_e.estimation.estimation,
                    "time": v_e.estimation.time,
                },
            }
            for v_e in estimations
        ]

        return result

`all_stops(self)` `async`

Return all the stops of the city.

Returns:

Type	Description
`Dict[str, Dict[str, Union[str, float, dict]]]`	List of dictionaries with the stop ID, transport type, way, name, coordinates and zone of each stop.

Source code in now8_api/service/stop_service.py

async def all_stops(
    self,
) -> Dict[str, Dict[str, Union[str, float, dict]]]:
    """Return all the stops of the city.

    Returns:
        List of dictionaries with the stop ID, transport type, way,
        name, coordinates and zone of each stop.
    """
    query_result: List[tuple] = await self.sql_engine.execute_query(
        str(_all_stops_query)
    )

    result: Dict[str, Dict[str, Any]] = {}
    for row in query_result:
        stop_id: str = row[0]
        stop_code: str = row[1]
        stop_name: str = row[2]
        stop_lat: float = row[3]
        stop_lon: float = row[4]
        stop_zone: str = row[5]

        if stop_id not in result:
            stop = Stop(
                id=stop_id,
                code=stop_code,
                name=stop_name,
                coordinates=Coordinates(
                    latitude=stop_lat, longitude=stop_lon
                ),
                zone=stop_zone,
            )

            result[stop.id] = {
                "id": stop.id,
                "code": stop.code,
                "name": stop.name,
                "longitude": stop.coordinates.longitude,
                "latitude": stop.coordinates.latitude,
                "zone": stop.zone,
                "route_ways": [],
            }

        route_id: str = row[6]
        route_way: int = row[7]

        result[stop_id]["route_ways"].append(
            {"id": route_id, "way": route_way}
        )

    return result

`stop_estimation(self, stop_id)` `async`

Return ETA for the next vehicles to the stop.

Parameters:

Name	Type	Description	Default
`stop_id`	`str`	Stop identifier.	required

Returns:

Type	Description
`List[Dict[str, dict]]`	ETA for the next vehicles to the stop.

Source code in now8_api/service/stop_service.py

async def stop_estimation(self, stop_id: str) -> List[Dict[str, dict]]:
    """Return ETA for the next vehicles to the stop.

    Arguments:
        stop_id: Stop identifier.

    Returns:
        ETA for the next vehicles to the stop.
    """
    stop = Stop(id=stop_id, transport_type=TransportType.INTERCITY_BUS)

    estimations = await self.city_data.get_estimations(stop)

    result: List[Dict[str, dict]] = [
        {
            "vehicle": {
                "id": v_e.vehicle.id,
                "route_way": {
                    "id": v_e.vehicle.route_id,
                    "way": v_e.vehicle.route_way.value
                    if v_e.vehicle.route_way is not None
                    else None,
                },
                "destination_stop": {
                    "id": v_e.vehicle.destination_stop.id,
                    "code": v_e.vehicle.destination_stop.code,
                    "name": v_e.vehicle.destination_stop.name,
                },
            },
            "estimation": {
                "estimation": v_e.estimation.estimation,
                "time": v_e.estimation.time,
            },
        }
        for v_e in estimations
    ]

    return result

`stop_info(self, stop_id)` `async`

Return the stop information.

Parameters:

Name	Type	Description	Default
`stop_id`	`str`	Stop identifier.	required

Returns:

Type	Description
`Dict[str, Union[str, float, list]]`	Dictionary with the stop ID, transport type, way, name, coordinates and zone.

Exceptions:

Type	Description
`StopNotFoundError`	If the `stop_id` does not match any stop.

Source code in now8_api/service/stop_service.py

async def stop_info(
    self, stop_id: str
) -> Dict[str, Union[str, float, list]]:
    """Return the stop information.

    Arguments:
        stop_id: Stop identifier.

    Returns:
        Dictionary with the stop ID, transport type, way, name,
            coordinates and zone.

    Raises:
        StopNotFoundError: If the `stop_id` does not match any stop.
    """
    query_result: List[tuple] = await self.sql_engine.execute_query(
        str(_all_stops_query.where(_table_stops.stop_id == stop_id))
    )

    if len(query_result) < 1:
        raise StopNotFoundError(stop_id=stop_id)

    result: Dict[str, Any] = {}
    for row in query_result:
        stop_code: str = row[1]
        stop_name: str = row[2]
        stop_lat: float = row[3]
        stop_lon: float = row[4]
        stop_zone: str = row[5]

        if result == {}:
            stop = Stop(
                id=stop_id,
                code=stop_code,
                name=stop_name,
                coordinates=Coordinates(
                    latitude=stop_lat, longitude=stop_lon
                ),
                zone=stop_zone,
            )

            result = {
                "id": stop.id,
                "code": stop.code,
                "name": stop.name,
                "longitude": stop.coordinates.longitude,
                "latitude": stop.coordinates.latitude,
                "zone": stop.zone,
                "route_ways": [],
            }

        route_id: str = row[6]
        route_way: int = row[7]

        result["route_ways"].append({"id": route_id, "way": route_way})

    return result

Last update: 2022-08-05

Stop service

StopNotFoundError (ValueError)

__init__(self, stop_id) special

StopService (Service) pydantic-model

all_stops(self) async

stop_estimation(self, stop_id) async

stop_info(self, stop_id) async

`StopNotFoundError (ValueError)`

`init(self, stop_id)` `special`

`StopService (Service)` `pydantic-model`

`all_stops(self)` `async`

`stop_estimation(self, stop_id)` `async`

`stop_info(self, stop_id)` `async`