Spaces:
Sleeping
Sleeping
File size: 3,716 Bytes
49e9f9d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | import json
import math
from typing import Dict, List, Optional, Tuple
from fastapi import WebSocket
# Map category → preferred skill tags. Volunteers tagged with any listed
# skill get the alert even if they're outside the normal radius (up to the
# extended radius). Keeps useful helpers aware of alerts that match them.
CATEGORY_PREFERRED_SKILLS: Dict[str, List[str]] = {
"medical": ["medical", "cpr", "elderly_care", "child_care"],
"flood": ["swim", "driver"],
"fire": ["medical", "driver"],
"missing": ["driver"],
"power": ["electrician"],
"other": [],
}
# "Skill match" extends the broadcast radius so a swimmer 10 km away still
# gets the flood alert, but someone 50 km away doesn't get spammed.
DEFAULT_RADIUS_KM = 5.0
SKILL_RADIUS_KM = 15.0
class ConnectionManager:
def __init__(self):
# volunteer_id -> (websocket, [lng, lat], skills, has_vehicle)
self._active: Dict[
str, Tuple[WebSocket, List[float], List[str], bool]
] = {}
def register(
self,
volunteer_id: str,
ws: WebSocket,
coordinates: List[float],
skills: Optional[List[str]] = None,
has_vehicle: bool = False,
):
self._active[volunteer_id] = (
ws,
coordinates,
list(skills or []),
bool(has_vehicle),
)
def disconnect(self, volunteer_id: str):
self._active.pop(volunteer_id, None)
def count(self) -> int:
return len(self._active)
def coords_for(self, volunteer_id: str) -> Optional[List[float]]:
"""Last-known [lng, lat] for a connected volunteer, or None when
they're offline. Used by the live-tracking endpoint so a reporter
can see "is my volunteer on the way?" without polling them
directly."""
entry = self._active.get(volunteer_id)
if entry is None:
return None
return list(entry[1])
async def broadcast_nearby(
self,
alert_dict: dict,
radius_km: float = DEFAULT_RADIUS_KM,
):
"""Broadcast an alert to volunteers within `radius_km`, plus volunteers
whose skills match the alert category within SKILL_RADIUS_KM.
Adds an `is_skill_match` flag so the client can render a stronger
notification when the alert is a near-perfect fit."""
a_lng, a_lat = alert_dict["location"]["coordinates"]
category = alert_dict.get("category", "other")
preferred = set(CATEGORY_PREFERRED_SKILLS.get(category, []))
for vid, (ws, coords, skills, has_vehicle) in list(self._active.items()):
v_lng, v_lat = coords
distance = _haversine(a_lat, a_lng, v_lat, v_lng)
skill_match = bool(preferred.intersection(set(skills)))
effective_radius = SKILL_RADIUS_KM if skill_match else radius_km
if distance > effective_radius:
continue
try:
payload = dict(alert_dict)
payload["is_skill_match"] = skill_match
payload["your_distance_km"] = round(distance, 2)
payload["your_has_vehicle"] = has_vehicle
await ws.send_text(json.dumps(payload, default=str))
except Exception:
self.disconnect(vid)
def _haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
R = 6371
dlat = math.radians(lat2 - lat1)
dlon = math.radians(lon2 - lon1)
a = (
math.sin(dlat / 2) ** 2
+ math.cos(math.radians(lat1))
* math.cos(math.radians(lat2))
* math.sin(dlon / 2) ** 2
)
return R * 2 * math.asin(math.sqrt(a))
manager = ConnectionManager()
|