import logging import xml.etree.ElementTree as ET from typing import Optional from urllib.parse import urlencode import requests from django.utils import timezone from scrobbles.models import Scrobble logger = logging.getLogger(__name__) # Speed thresholds in meters per second SPEED_THRESHOLDS = { "at_rest": 0.5, "walking": 1, "running": 2.5, "bicycling": 5, "driving": 10, } # Activity mapping from Android activity recognition ACTIVITY_MAPPING = { "still": "at_rest", "on_foot": "walking", "walking": "walking", "running": "running", "on_bicycle": "bicycling", "in_vehicle": "driving", "unknown": "unknown", } def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """Calculate the great-circle distance between two points in meters.""" from math import radians, cos, sin, asin, sqrt lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2]) dlat = lat2 - lat1 dlon = lon2 - lon1 a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2 c = 2 * asin(sqrt(a)) r = 6371000 # Radius of earth in meters return c * r def detect_movement( current_scrobble: Scrobble, previous_scrobble: Optional[Scrobble] = None, ) -> dict: """ Analyze location scrobble to determine if moving and mode of transport. Returns a dict shaped like GeoLocationLogData that can be stored in scrobble.log: { "is_moving": bool, "movement_type": str, "confidence": str, "speed_mps": float, "speed_accuracy": float, "detection_method": str, "activity": str, "detected_at": str, } """ result = { "is_moving": False, "movement_type": "unknown", "confidence": "low", "speed_mps": 0.0, "speed_accuracy": 0.0, "detection_method": "unknown", "activity": "", "detected_at": timezone.now().isoformat(), } if current_scrobble.media_type != Scrobble.MediaType.GEO_LOCATION: return result # Get GPS data from log log_data = current_scrobble.log or {} gps_data = log_data.get("gps_data", {}) # Method 1: Check Android activity recognition (highest confidence) activity = gps_data.get("activity", "") if activity and activity in ACTIVITY_MAPPING: movement_type = ACTIVITY_MAPPING.get(activity, "unknown") result.update( { "is_moving": movement_type != "at_rest", "movement_type": movement_type, "confidence": "high", "detection_method": "activity_recognition", "activity": activity, } ) return result # Method 2: Use speed from GPS speed = gps_data.get("speed") accuracy = gps_data.get("accuracy", 100.0) if speed is not None: try: speed = float(speed) result["speed_mps"] = speed result["speed_accuracy"] = float(accuracy) # Determine movement type from speed movement_type = "at_rest" is_moving = False if speed >= SPEED_THRESHOLDS["walking"]: movement_type = "walking" is_moving = True if speed >= SPEED_THRESHOLDS["running"]: movement_type = "running" is_moving = True if speed >= SPEED_THRESHOLDS["bicycling"]: movement_type = "bicycling" is_moving = True if speed >= SPEED_THRESHOLDS["driving"]: movement_type = "driving" is_moving = True # Confidence based on GPS accuracy confidence = "low" if accuracy <= 10: confidence = "high" elif accuracy <= 30: confidence = "medium" result.update( { "is_moving": is_moving, "movement_type": movement_type, "confidence": confidence, "detection_method": "gps_speed", } ) return result except (ValueError, TypeError): pass # Method 3: Calculate speed from distance/time between scrobbles if not previous_scrobble: # Try to get previous scrobble from the same user previous_scrobble = ( Scrobble.objects.filter( user=current_scrobble.user, media_type=Scrobble.MediaType.GEO_LOCATION, timestamp__lt=current_scrobble.timestamp, ) .order_by("-timestamp") .first() ) if previous_scrobble and previous_scrobble.geo_location: time_diff = ( current_scrobble.timestamp - previous_scrobble.timestamp ).total_seconds() if time_diff > 0: distance = haversine_distance( previous_scrobble.geo_location.lat, previous_scrobble.geo_location.lon, current_scrobble.geo_location.lat, current_scrobble.geo_location.lon, ) speed = distance / time_diff result["speed_mps"] = round(speed, 2) # Determine movement type from calculated speed movement_type = "at_rest" is_moving = False if speed >= SPEED_THRESHOLDS["walking"]: movement_type = "walking" is_moving = True if speed >= SPEED_THRESHOLDS["running"]: movement_type = "running" is_moving = True if speed >= SPEED_THRESHOLDS["bicycling"]: movement_type = "bicycling" is_moving = True if speed >= SPEED_THRESHOLDS["driving"]: movement_type = "driving" is_moving = True # Lower confidence for calculated speed result.update( { "is_moving": is_moving, "movement_type": movement_type, "confidence": "low", "detection_method": "calculated_speed", } ) return result return result NWS_URL = "https://forecast.weather.gov/MapClick.php" def fetch_current_weather(lat: float, lon: float) -> Optional[dict]: """Fetch current weather from NWS DWML feed for the given lat/lon. Returns dict with 'temp' and 'description' keys, or None on failure. """ params = { "lat": lat, "lon": lon, "unit": 0, "lg": "english", "FcstType": "dwml", } url = f"{NWS_URL}?{urlencode(params)}" try: resp = requests.get(url, timeout=10) resp.raise_for_status() except requests.RequestException as e: logger.warning("Failed to fetch NWS weather data: %s", e) return None try: root = ET.fromstring(resp.content) except ET.ParseError as e: logger.warning("Failed to parse NWS XML: %s", e) return None # Find current observations data block data_blocks = root.findall("data") obs_block = None for block in data_blocks: if block.get("type") == "current observations": obs_block = block break if obs_block is None: logger.warning("No current observations block in NWS response") return None params_el = obs_block.find("parameters") if params_el is None: return None # Temperature (apparent / feels-like) temp = None for temp_el in params_el.findall("temperature"): if temp_el.get("type") == "apparent": val_el = temp_el.find("value") if val_el is not None and val_el.text: try: temp = int(round(float(val_el.text))) except (ValueError, TypeError): pass break # Weather description description = None weather_el = params_el.find("weather") if weather_el is not None: cond_el = weather_el.find("weather-conditions") if cond_el is not None: description = cond_el.get("weather-summary") if temp is None and description is None: logger.warning("No weather data found in NWS response") return None return {"temp": temp, "description": description}