IoT-Systems/MapView/main.py

import asyncio
from kivy.app import App
from kivy_garden.mapview import MapMarker, MapView
from kivy.clock import Clock
from lineMapLayer import LineMapLayer
from datasource import Datasource
import config

line_layer_colors = [
    [1, 0, 0, 1],
    [1, 0.5, 0, 1],
    [0, 1, 0, 1],
    [0, 1, 1, 1],
    [0, 0, 1, 1],
    [1, 0, 1, 1],
]


def get_lat_lon(point: dict[str, float] | list[float] | tuple[float, float]) -> tuple[float, float] | None:
    if isinstance(point, dict):
        lat = point.get("lat")
        lon = point.get("lon")
    else:
        lat, lon = point

    if lat is None or lon is None:
        return None
    return lat, lon


class MapViewApp(App):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

        self.mapview: MapView | None = None
        self.datasource = Datasource(user_id=1)
        self.line_layers = dict()
        self.car_markers = dict()

        # додати необхідні змінні
        self.bump_markers: list[MapMarker] = []
        self.pothole_markers: list[MapMarker] = []

    def on_start(self):
        """
        Встановлює необхідні маркери, викликає функцію для оновлення мапи
        """
        self.update()
        Clock.schedule_interval(self.update, 0.1)

    def update(self, *args):
        """
        Викликається регулярно для оновлення мапи
        """
        new_points = self.datasource.get_new_points()

        if not new_points:
            return

        for point in new_points:

            lat, lon, road_state, user_id = point

            # Оновлює лінію маршрута
            if user_id not in self.line_layers:
                self.line_layers[user_id] = LineMapLayer(color = line_layer_colors[user_id % len(line_layer_colors)])
                self.mapview.add_layer(self.line_layers[user_id])

            self.line_layers[user_id].add_point((lat, lon))

            # Оновлює маркер маниши
            self.update_car_marker(lat, lon, user_id)

            # Перевіряємо стан дороги
            self.check_road_quality(point)

    def check_road_quality(self, point):
        """
        Аналізує дані акселерометра для подальшого визначення
        та відображення ям та лежачих поліцейських
        """
        if len(point) < 3:
            return

        lat, lon, road_state, user_id = point

        if road_state == "pothole":
            self.set_pothole_marker((lat, lon))
        elif road_state == "bump":
            self.set_bump_marker((lat, lon))

    def update_car_marker(self, lat, lon, user_id):
        """
        Оновлює відображення маркера машини на мапі
        :param point: GPS координати
        """
        if user_id not in self.car_markers:
            self.car_markers[user_id] = MapMarker(lat=lat, lon=lon, source='./images/car.png')
            self.mapview.add_marker(self.car_markers[user_id])
        else:
            self.car_markers[user_id].lat = lat
            self.car_markers[user_id].lon = lon

        if user_id == config.TRACK_ID:
            self.mapview.center_on(lat, lon)

    def map_lat_lon_to_marker(self, lat: float, lon: float) -> MapMarker | None:
        flt = filter(lambda marker: lon == marker.lat and lat == marker.lon, self.pothole_markers + self.bump_markers)

        try:
            return next(flt)
        except StopIteration as e:
            return None

    def set_pothole_marker(self, point):
        lat, lon = get_lat_lon(point)
        if lat is None or lon is None:
            return

        marker = MapMarker(
            lat=lat,
            lon=lon,
            source="images/pothole.png"
        )

        self.mapview.add_marker(marker)
        self.pothole_markers.append(marker)

    def set_bump_marker(self, point):
        lat, lon = get_lat_lon(point)
        if lat is None or lon is None:
            return

        marker = MapMarker(
            lat=lat,
            lon=lon,
            source="images/bump.png"
        )

        self.mapview.add_marker(marker)
        self.bump_markers.append(marker)

    def delete_pothole_marker(self, point):
        lat, lon = get_lat_lon(point)
        if lat is None or lon is None:
            return

        clicked_marker_data = self.datasource.map_lat_lon_to_ProcessedAgentData(lat, lon)

        if not clicked_marker_data:
            return

        clicked_marker = self.map_lat_lon_to_marker(clicked_marker_data.latitude, clicked_marker_data.longitude)

        if clicked_marker == None:
            return

        self.mapview.remove_marker(clicked_marker)

        if clicked_marker in self.pothole_markers:
            self.pothole_markers.pop(self.pothole_markers.index(clicked_marker))
        elif clicked_marker in self.bump_markers:
            self.bump_markers.pop(self.bump_markers.index(clicked_marker))

    def on_touch_down(self, widget, touch):
        if touch.button == "right":
            coordinate = self.mapview.get_latlon_at(touch.x, touch.y, self.mapview.zoom)
            self.delete_pothole_marker(coordinate)
            return True


    def build(self):
        """
        Ініціалізує мапу MapView(zoom, lat, lon)
        :return: мапу
        """
        self.mapview = MapView(
            zoom=15,
            lat=50.4501,
            lon=30.5234
        )

        self.mapview.bind(on_touch_down = self.on_touch_down)

        return self.mapview


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(MapViewApp().async_run(async_lib="asyncio"))
    loop.close()