275 lines
8.2 KiB
Python
275 lines
8.2 KiB
Python
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}
|