spothole/providers/activityrefdata/toilets.py

import csv

from core.enums import ActivityRefType
from data.activity_ref import ActivityRef
from providers.activityrefdata.local_file_activity_ref_data_provider import (
    LocalFileActivityRefDataProvider,
)


class Toilets(LocalFileActivityRefDataProvider):
    """Activity ref data provider for Toilets on the Air"""

    ACTIVITY = "Toilets"
    PATH = "datafiles/toilets.csv"

    def __init__(self, provider_config):
        super().__init__(self.ACTIVITY, provider_config, self.PATH)

    def _file_to_data(self, path):
        new_data = []
        with open(path) as _f:
            csv_data = _f.read()
            dr = csv.DictReader(csv_data.splitlines())
            for row in dr:
                new_data.append(
                    ActivityRef(
                        sig=self.ACTIVITY,
                        id=row["ref"],
                        name=row["ref"],
                        ref_type=ActivityRefType.TOILET,
                        latitude=float(row["lat"]),
                        longitude=float(row["lon"]),
                    )
                )

                # Bail out if a stop has been requested, i.e. the program is shutting down - no need to parse the rest
                # of the data in this case
                if self._stop_event.is_set():
                    break

        return new_data