162 lines
5.2 KiB
Python
162 lines
5.2 KiB
Python
import logging
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal
|
|
from typing import Dict, Optional
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.db import models
|
|
from django.urls import reverse
|
|
from scrobbles.dataclasses import BaseLogData, WithPeopleLogData
|
|
from scrobbles.mixins import ScrobblableConstants, ScrobblableMixin
|
|
|
|
logger = logging.getLogger(__name__)
|
|
BNULL = {"blank": True, "null": True}
|
|
User = get_user_model()
|
|
GEOLOC_ACCURACY = int(getattr(settings, "GEOLOC_ACCURACY", 4))
|
|
GEOLOC_PROXIMITY = Decimal(getattr(settings, "GEOLOC_PROXIMITY", "0.0001"))
|
|
|
|
|
|
@dataclass
|
|
class GeoLocationLogData(BaseLogData, WithPeopleLogData):
|
|
is_moving: bool = False
|
|
movement_type: str = "unknown"
|
|
confidence: str = "low"
|
|
speed_mps: float = 0.0
|
|
speed_accuracy: float = 0.0
|
|
detection_method: str = "unknown"
|
|
activity: str = ""
|
|
detected_at: str = ""
|
|
|
|
@classmethod
|
|
def from_log_dict(cls, log_dict: dict) -> dict:
|
|
instance_data = log_dict.get("movement_detection", {}).copy()
|
|
for field_name in ["description", "notes", "with_people_ids"]:
|
|
if field_name in log_dict:
|
|
instance_data[field_name] = log_dict[field_name]
|
|
return instance_data
|
|
|
|
|
|
class GeoLocation(ScrobblableMixin):
|
|
COMPLETION_PERCENT = getattr(settings, "LOCATION_COMPLETION_PERCENT", 100)
|
|
|
|
lat = models.FloatField()
|
|
lon = models.FloatField()
|
|
truncated_lat = models.FloatField(**BNULL)
|
|
truncated_lon = models.FloatField(**BNULL)
|
|
altitude = models.FloatField(**BNULL)
|
|
|
|
class Meta:
|
|
unique_together = [["lat", "lon", "altitude"]]
|
|
|
|
def __str__(self):
|
|
if self.title:
|
|
return self.title
|
|
|
|
return f"{self.lat} x {self.lon}"
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("locations:geolocation_detail", kwargs={"slug": self.uuid})
|
|
|
|
@property
|
|
def logdata_cls(self):
|
|
return GeoLocationLogData
|
|
|
|
@classmethod
|
|
def find_or_create(cls, data_dict: Dict) -> "GeoLocation":
|
|
"""Given a data dict from GPSLogger, does the heavy lifting of looking up
|
|
the location, creating if if doesn't exist yet.
|
|
|
|
"""
|
|
# TODO Add constants for all these data keys
|
|
if "lat" not in data_dict.keys() or "lon" not in data_dict.keys():
|
|
logger.error("No lat or lon keys in data dict")
|
|
return
|
|
|
|
int_lat, r_lat = str(data_dict.get("lat", "")).split(".")
|
|
int_lon, r_lon = str(data_dict.get("lon", "")).split(".")
|
|
|
|
try:
|
|
trunc_lat = r_lat[0:GEOLOC_ACCURACY]
|
|
except IndexError:
|
|
trunc_lat = r_lat
|
|
try:
|
|
trunc_lon = r_lon[0:GEOLOC_ACCURACY]
|
|
except IndexError:
|
|
trunc_lon = r_lon
|
|
|
|
data_dict["lat"] = float(f"{int_lat}.{trunc_lat}")
|
|
data_dict["lon"] = float(f"{int_lon}.{trunc_lon}")
|
|
|
|
int_alt, r_alt = str(data_dict.get("alt", "")).split(".")
|
|
|
|
data_dict["altitude"] = float(int_alt)
|
|
|
|
location = cls.objects.filter(
|
|
lat=data_dict.get("lat"),
|
|
lon=data_dict.get("lon"),
|
|
).first()
|
|
|
|
if not location:
|
|
location = cls.objects.create(
|
|
lat=data_dict.get("lat"),
|
|
lon=data_dict.get("lon"),
|
|
altitude=data_dict.get("altitude"),
|
|
)
|
|
return location
|
|
|
|
@property
|
|
def subtitle(self) -> str:
|
|
if self.title:
|
|
return f"{self.lat} x {self.lon}"
|
|
return ""
|
|
|
|
@property
|
|
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)),
|
|
abs(Decimal(old_lat_lon[1]) - Decimal(self.lon)),
|
|
)
|
|
|
|
def has_moved(self, previous_location: "GeoLocation") -> bool:
|
|
has_moved = False
|
|
|
|
loc_diff = self.loc_diff((previous_location.lat, previous_location.lon))
|
|
if loc_diff[0] > GEOLOC_PROXIMITY or loc_diff[1] > GEOLOC_PROXIMITY:
|
|
has_moved = True
|
|
logger.debug(
|
|
f"[locations] checked whether location has moved against proximity setting",
|
|
extra={
|
|
"location_id": self.id,
|
|
"loc_diff": loc_diff,
|
|
"has_moved": has_moved,
|
|
"previous_location_id": previous_location.id,
|
|
"geoloc_proximity": GEOLOC_PROXIMITY,
|
|
},
|
|
)
|
|
return has_moved
|
|
|
|
def in_proximity(self, named=False) -> models.QuerySet:
|
|
lat_min = Decimal(self.lat) - GEOLOC_PROXIMITY
|
|
lat_max = Decimal(self.lat) + GEOLOC_PROXIMITY
|
|
lon_min = Decimal(self.lon) - GEOLOC_PROXIMITY
|
|
lon_max = Decimal(self.lon) + GEOLOC_PROXIMITY
|
|
is_title_null = not named
|
|
close_locations = GeoLocation.objects.filter(
|
|
title__isnull=is_title_null,
|
|
lat__lte=lat_max,
|
|
lat__gte=lat_min,
|
|
lon__lte=lon_max,
|
|
lon__gte=lon_min,
|
|
).exclude(id=self.id)
|
|
return close_locations
|