add fallback_locations for via_ip

This commit is contained in:
Sean Breckenridge 2023-01-27 11:46:22 -08:00
parent f2c9715dbc
commit 31981b8f34
3 changed files with 62 additions and 10 deletions

View file

@ -7,7 +7,7 @@ REQUIRES = ["git+https://github.com/seanbreckenridge/ipgeocache"]
from my.core import __NOT_HPI_MODULE__ from my.core import __NOT_HPI_MODULE__
import ipaddress import ipaddress
from typing import NamedTuple, Iterator from typing import NamedTuple, Iterator, Tuple
from datetime import datetime from datetime import datetime
import ipgeocache import ipgeocache
@ -23,6 +23,12 @@ class IP(NamedTuple):
def ipgeocache(self) -> Json: def ipgeocache(self) -> Json:
return ipgeocache.get(self.addr) return ipgeocache.get(self.addr)
@property
def latlon(self) -> Tuple[float, float]:
loc: str = self.ipgeocache()["loc"]
lat, _, lon = loc.partition(",")
return float(lat), float(lon)
@property @property
def tzname(self) -> str: def tzname(self) -> str:
tz: str = self.ipgeocache()["timezone"] tz: str = self.ipgeocache()["timezone"]

View file

@ -10,16 +10,16 @@ class FallbackLocation:
lat: float lat: float
lon: float lon: float
dt: datetime dt: datetime
duration: int # time in seconds for how long this is valid duration: float # time in seconds for how long this is valid
accuracy: Optional[float] = None accuracy: Optional[float] = None
elevation: Optional[float] = None elevation: Optional[float] = None
datasource: Optional[str] = None # which module provided this, useful for debugging datasource: Optional[str] = None # which module provided this, useful for debugging
def to_location(self, end: bool = False) -> Location: def to_location(self, end: bool = False) -> Location:
""" '''
by default the start date is used for the location by default the start date is used for the location
If end is True, the start date + duration is used If end is True, the start date + duration is used
""" '''
dt: datetime = self.dt dt: datetime = self.dt
if end: if end:
dt += timedelta(self.duration) dt += timedelta(self.duration)
@ -31,3 +31,33 @@ class FallbackLocation:
elevation=self.elevation, elevation=self.elevation,
datasource=self.datasource, datasource=self.datasource,
) )
@classmethod
def from_end_date(
cls,
lat: float,
lon: float,
dt: datetime,
end_dt: datetime,
accuracy: Optional[float] = None,
elevation: Optional[float] = None,
datasource: Optional[str] = None,
) -> 'FallbackLocation':
'''
Create FallbackLocation from a start date and an end date
'''
if end_dt < dt:
raise ValueError('end_date must be after dt')
duration = (end_dt - dt).total_seconds()
return cls(
lat=lat,
lon=lon,
dt=dt,
duration=duration,
accuracy=accuracy,
elevation=elevation,
datasource=datasource,
)
# TODO: create estimate location which uses other fallback_locations to estimate a location

View file

@ -6,6 +6,8 @@ REQUIRES = ["git+https://github.com/seanbreckenridge/ipgeocache"]
from my.core import dataclass, Stats from my.core import dataclass, Stats
from my.config import location from my.config import location
from my.core.warnings import medium
from datetime import datetime
@dataclass @dataclass
@ -13,26 +15,40 @@ class config(location.via_ip):
# no real science to this, just a guess of ~15km accuracy for IP addresses # no real science to this, just a guess of ~15km accuracy for IP addresses
accuracy: float = 15_000.0 accuracy: float = 15_000.0
for_duration: float = 60 * 10 # default to being accurate for ~10 minutes
from typing import Iterator from typing import Iterator
from ..common import Location from ..common import Location
from .common import FallbackLocation
from my.ip.all import ips from my.ip.all import ips
def locations() -> Iterator[Location]: def fallback_locations() -> Iterator[FallbackLocation]:
for ip in ips(): for ip in ips():
loc: str = ip.ipgeocache()["loc"] lat, lon = ip.latlon
lat, _, lon = loc.partition(",") yield FallbackLocation(
yield Location( lat=lat,
lat=float(lat), lon=lon,
lon=float(lon),
dt=ip.dt, dt=ip.dt,
accuracy=config.accuracy, accuracy=config.accuracy,
duration=config.for_duration,
elevation=None, elevation=None,
datasource="ip",
) )
# for compatibility with my.location.via_ip, this shouldnt be used by other modules
def locations() -> Iterator[Location]:
medium("via_ip.locations is deprecated, use via_ip.fallback_locations instead")
yield from map(FallbackLocation.to_location, fallback_locations())
def estimate_location(dt: datetime) -> Location:
raise NotImplementedError("not implemented yet")
def stats() -> Stats: def stats() -> Stats:
from my.core import stat from my.core import stat