Compare commits
8 Commits
project/hu
...
4c594bb7cc
| Author | SHA1 | Date | |
|---|---|---|---|
| 4c594bb7cc | |||
| 9dcde8ef2a | |||
| 2d10d13214 | |||
| eeb3266a32 | |||
| de9ecda77f | |||
| 030c6e5fe9 | |||
| e455fc41de | |||
| f12562eb02 |
@@ -37,6 +37,7 @@ class Datasource:
|
|||||||
self.user_id = user_id
|
self.user_id = user_id
|
||||||
self.connection_status = None
|
self.connection_status = None
|
||||||
self._new_points = []
|
self._new_points = []
|
||||||
|
self._active_markers = []
|
||||||
asyncio.ensure_future(self.connect_to_server())
|
asyncio.ensure_future(self.connect_to_server())
|
||||||
|
|
||||||
def get_new_points(self):
|
def get_new_points(self):
|
||||||
@@ -60,6 +61,20 @@ class Datasource:
|
|||||||
self.connection_status = "Disconnected"
|
self.connection_status = "Disconnected"
|
||||||
Logger.debug("SERVER DISCONNECT")
|
Logger.debug("SERVER DISCONNECT")
|
||||||
|
|
||||||
|
def map_lat_lon_to_ProcessedAgentData(self, lat: float, lon: float) -> ProcessedAgentData | None:
|
||||||
|
distances = tuple((abs(lon - marker.latitude) ** 2 + abs(lat - marker.longitude) ** 2) ** 0.5 for marker in self._active_markers)
|
||||||
|
|
||||||
|
if len(distances) == 0:
|
||||||
|
return None
|
||||||
|
|
||||||
|
min_distance = min(distances)
|
||||||
|
marker = self._active_markers[distances.index(min_distance)]
|
||||||
|
|
||||||
|
if min_distance < 0.005:
|
||||||
|
return marker
|
||||||
|
else:
|
||||||
|
return None
|
||||||
|
|
||||||
def handle_received_data(self, data):
|
def handle_received_data(self, data):
|
||||||
# Update your UI or perform actions with received data here
|
# Update your UI or perform actions with received data here
|
||||||
Logger.debug(f"Received data: {data}")
|
Logger.debug(f"Received data: {data}")
|
||||||
@@ -70,6 +85,9 @@ class Datasource:
|
|||||||
],
|
],
|
||||||
key=lambda v: v.timestamp,
|
key=lambda v: v.timestamp,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
self._active_markers += [i for i in processed_agent_data_list if i.road_state != 'normal']
|
||||||
|
|
||||||
new_points = [
|
new_points = [
|
||||||
(
|
(
|
||||||
processed_agent_data.longitude,
|
processed_agent_data.longitude,
|
||||||
|
|||||||
@@ -15,18 +15,31 @@ line_layer_colors = [
|
|||||||
[1, 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):
|
class MapViewApp(App):
|
||||||
def __init__(self, **kwargs):
|
def __init__(self, **kwargs):
|
||||||
super().__init__(**kwargs)
|
super().__init__(**kwargs)
|
||||||
|
|
||||||
self.mapview = None
|
self.mapview: MapView | None = None
|
||||||
self.datasource = Datasource(user_id=1)
|
self.datasource = Datasource(user_id=1)
|
||||||
self.line_layers = dict()
|
self.line_layers = dict()
|
||||||
self.car_markers = dict()
|
self.car_markers = dict()
|
||||||
|
|
||||||
# додати необхідні змінні
|
# додати необхідні змінні
|
||||||
self.bump_markers = []
|
self.bump_markers: list[MapMarker] = []
|
||||||
self.pothole_markers = []
|
self.pothole_markers: list[MapMarker] = []
|
||||||
|
|
||||||
def on_start(self):
|
def on_start(self):
|
||||||
"""
|
"""
|
||||||
@@ -91,13 +104,16 @@ class MapViewApp(App):
|
|||||||
if user_id == config.TRACK_ID:
|
if user_id == config.TRACK_ID:
|
||||||
self.mapview.center_on(lat, lon)
|
self.mapview.center_on(lat, lon)
|
||||||
|
|
||||||
def set_pothole_marker(self, point):
|
def map_lat_lon_to_marker(self, lat: float, lon: float) -> MapMarker | None:
|
||||||
if isinstance(point, dict):
|
flt = filter(lambda marker: lon == marker.lat and lat == marker.lon, self.pothole_markers + self.bump_markers)
|
||||||
lat = point.get("lat")
|
|
||||||
lon = point.get("lon")
|
|
||||||
else:
|
|
||||||
lat, lon = point
|
|
||||||
|
|
||||||
|
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:
|
if lat is None or lon is None:
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -111,24 +127,47 @@ class MapViewApp(App):
|
|||||||
self.pothole_markers.append(marker)
|
self.pothole_markers.append(marker)
|
||||||
|
|
||||||
def set_bump_marker(self, point):
|
def set_bump_marker(self, point):
|
||||||
if isinstance(point, dict):
|
lat, lon = get_lat_lon(point)
|
||||||
lat = point.get("lat")
|
|
||||||
lon = point.get("lon")
|
|
||||||
else:
|
|
||||||
lat, lon = point
|
|
||||||
|
|
||||||
if lat is None or lon is None:
|
if lat is None or lon is None:
|
||||||
return
|
return
|
||||||
|
|
||||||
marker = MapMarker(
|
marker = MapMarker(
|
||||||
lat=lat,
|
lat=lat,
|
||||||
lon=lon,
|
lon=lon,
|
||||||
source="images/bump.png"
|
source="images/bump.png"
|
||||||
)
|
)
|
||||||
|
|
||||||
self.mapview.add_marker(marker)
|
self.mapview.add_marker(marker)
|
||||||
self.bump_markers.append(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):
|
def build(self):
|
||||||
"""
|
"""
|
||||||
@@ -141,6 +180,8 @@ class MapViewApp(App):
|
|||||||
lon=30.5234
|
lon=30.5234
|
||||||
)
|
)
|
||||||
|
|
||||||
|
self.mapview.bind(on_touch_down = self.on_touch_down)
|
||||||
|
|
||||||
return self.mapview
|
return self.mapview
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user