[locations] Add weather fetching to birds and moods
This commit is contained in:
@ -16,6 +16,7 @@ class BirdingLocationAdmin(admin.ModelAdmin):
|
||||
date_hierarchy = "created"
|
||||
list_display = ("uuid", "title")
|
||||
ordering = ("-created",)
|
||||
raw_id_fields = ("geo_location",)
|
||||
search_fields = ("title",)
|
||||
inlines = [
|
||||
ScrobbleInline,
|
||||
|
||||
@ -136,6 +136,25 @@ def import_birding_csv(file_path, user_id):
|
||||
|
||||
log_dict = logdata.asdict
|
||||
|
||||
weather_loc = location.geo_location
|
||||
if not weather_loc:
|
||||
last_loc = (
|
||||
Scrobble.objects.filter(
|
||||
user=user,
|
||||
media_type=Scrobble.MediaType.GEO_LOCATION,
|
||||
geo_location__isnull=False,
|
||||
)
|
||||
.order_by("-timestamp")
|
||||
.first()
|
||||
)
|
||||
if last_loc:
|
||||
weather_loc = last_loc.geo_location
|
||||
if weather_loc:
|
||||
weather = weather_loc.current_weather
|
||||
if weather:
|
||||
log_dict["weather"] = weather["description"]
|
||||
log_dict["temperature"] = weather["temp"]
|
||||
|
||||
stop_timestamp = timestamp + timedelta(minutes=duration_minutes) if duration_minutes else None
|
||||
|
||||
scrobble = Scrobble(
|
||||
|
||||
@ -16,6 +16,7 @@ class GeoLocationAdmin(admin.ModelAdmin):
|
||||
"altitude",
|
||||
)
|
||||
ordering = ("-created",)
|
||||
search_fields = ("title",)
|
||||
inlines = [
|
||||
ScrobbleInline,
|
||||
]
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
from typing import Dict
|
||||
from typing import Dict, Optional
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth import get_user_model
|
||||
@ -107,6 +107,12 @@ class GeoLocation(ScrobblableMixin):
|
||||
def strings(self) -> ScrobblableConstants:
|
||||
return ScrobblableConstants(verb="Going", tags="world_map", priority="low")
|
||||
|
||||
@property
|
||||
def current_weather(self) -> Optional[dict]:
|
||||
from locations.utils import fetch_current_weather
|
||||
|
||||
return fetch_current_weather(self.lat, self.lon)
|
||||
|
||||
def loc_diff(self, old_lat_lon: tuple) -> tuple:
|
||||
return (
|
||||
abs(Decimal(old_lat_lon[0]) - Decimal(self.lat)),
|
||||
|
||||
@ -1,6 +1,9 @@
|
||||
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
|
||||
|
||||
@ -196,3 +199,76 @@ def detect_movement(
|
||||
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}
|
||||
|
||||
@ -1,80 +0,0 @@
|
||||
import logging
|
||||
import xml.etree.ElementTree as ET
|
||||
from typing import Optional
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import requests
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
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}
|
||||
@ -5,7 +5,6 @@ from django.shortcuts import redirect, render
|
||||
from django.urls import reverse
|
||||
from django.utils import timezone
|
||||
from moods.models import Mood, MoodLogData, MoodQuality, MoodReason
|
||||
from moods.utils import fetch_current_weather
|
||||
from scrobbles.models import Scrobble
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -78,10 +77,7 @@ def checkin(request):
|
||||
.first()
|
||||
)
|
||||
if last_loc and last_loc.geo_location:
|
||||
weather = fetch_current_weather(
|
||||
last_loc.geo_location.lat,
|
||||
last_loc.geo_location.lon,
|
||||
)
|
||||
weather = last_loc.geo_location.current_weather
|
||||
if weather:
|
||||
log_data.weather_temp = weather["temp"]
|
||||
log_data.weather_description = weather["description"]
|
||||
|
||||
Reference in New Issue
Block a user