sij/sijapi
1
1
Fork 0
Code Issues 3 Pull requests Projects Releases Packages Wiki Activity Actions

Auto-update: Fri Jun 28 23:26:17 PDT 2024

Browse source
This commit is contained in:
sanj 2024-06-28 23:26:17 -07:00
parent 0660455eea
commit bc1924dd4b
4 changed files with 102 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
sijapi/__init__.py
View file

@ -12,7 +12,7 @@ from typing import List, Optional
import traceback
import logging
from .logs import Logger
from .classes import AutoResponder, IMAPConfig, SMTPConfig, EmailAccount, EmailContact, IncomingEmail, TimezoneTracker, Database, Geocoder
from .classes import AutoResponder, IMAPConfig, SMTPConfig, EmailAccount, EmailContact, IncomingEmail, Database, Geocoder
# from sijapi.config.config import load_config
# cfg = load_config()

114
sijapi/classes.py
View file

@ -3,6 +3,8 @@ from typing import List, Optional, Any, Tuple, Dict, Union, Tuple
from datetime import datetime, timedelta, timezone
import asyncio
import json
import yaml
import math
from timezonefinder import TimezoneFinder
from pathlib import Path
import asyncpg
@ -46,6 +48,7 @@ class Location(BaseModel):
}
class Geocoder:
def __init__(self, named_locs: Union[str, Path] = None, cache_file: Union[str, Path] = 'timezone_cache.json'):
self.tf = TimezoneFinder()
@ -56,10 +59,52 @@ class Geocoder:
self.last_update: Optional[datetime] = None
self.last_location: Optional[Tuple[float, float]] = None
self.executor = ThreadPoolExecutor()
self.override_locations = self.load_override_locations()
def load_override_locations(self):
if self.named_locs and self.named_locs.exists():
with open(self.named_locs, 'r') as file:
return yaml.safe_load(file)
return []
def haversine(self, lat1, lon1, lat2, lon2):
R = 6371
lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2])
dlat = lat2 - lat1
dlon = lon2 - lon1
a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
return R * c
def find_override_location(self, lat: float, lon: float) -> Optional[str]:
closest_location = None
closest_distance = float('inf')
for location in self.override_locations:
loc_name = location.get("name")
loc_lat = location.get("latitude")
loc_lon = location.get("longitude")
loc_radius = location.get("radius")
distance = self.haversine(lat, lon, loc_lat, loc_lon)
if distance <= loc_radius:
if distance < closest_distance:
closest_distance = distance
closest_location = loc_name
return closest_location
async def location(self, lat: float, lon: float):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(self.executor, rg.search, [(lat, lon)])
result = await loop.run_in_executor(self.executor, rg.search, [(lat, lon)])
override = self.find_override_location(lat, lon)
if override:
result[0]['override_name'] = override
return result
async def elevation(self, latitude: float, longitude: float, unit: str = "m") -> float:
loop = asyncio.get_running_loop()
@ -107,12 +152,14 @@ class Geocoder:
coordinates = [(location.latitude, location.longitude) for location in processed_locations]
geocode_results = await self.location(*zip(*coordinates))
geocode_results = await asyncio.gather(*[self.location(lat, lon) for lat, lon in coordinates])
elevations = await asyncio.gather(*[self.elevation(lat, lon) for lat, lon in coordinates])
timezones = await asyncio.gather(*[self.timezone(lat, lon) for lat, lon in coordinates])
geocoded_locations = []
for location, result, elevation, timezone in zip(processed_locations, geocode_results, elevations, timezones):
result = result[0] # Unpack the first result
override_name = result.get('override_name')
geocoded_location = Location(
latitude=location.latitude,
longitude=location.longitude,
@ -123,8 +170,8 @@ class Geocoder:
state=result.get("admin1"),
country=result.get("cc"),
context=location.context or {},
name=result.get("name"),
display_name=f"{result.get('name')}, {result.get('admin1')}, {result.get('cc')}",
name=override_name or result.get("name"),
display_name=f"{override_name or result.get('name')}, {result.get('admin1')}, {result.get('cc')}",
country_code=result.get("cc"),
timezone=timezone
)
@ -176,44 +223,12 @@ class Geocoder:
timezone=await self.timezone(latitude, longitude)
)
def load_override_locations(self):
if self.named_locs and self.named_locs.exists():
with open(self.named_locs, 'r') as file:
return yaml.safe_load(file)
return []
def haversine(self, lat1, lon1, lat2, lon2):
R = 6371 # Earth's radius in kilometers
def round_coords(self, lat: float, lon: float, decimal_places: int = 2) -> Tuple[float, float]:
return (round(lat, decimal_places), round(lon, decimal_places))
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 * atan2(sqrt(a), sqrt(1-a))
return R * c
async def find_override_location(self, lat: float, lon: float) -> Optional[str]:
closest_location = None
closest_distance = float('inf')
for location in self.override_locations:
loc_name = location.get("name")
loc_lat = location.get("latitude")
loc_lon = location.get("longitude")
loc_radius = location.get("radius")
distance = self.haversine(lat, lon, loc_lat, loc_lon)
if distance <= loc_radius:
if distance < closest_distance:
closest_distance = distance
closest_location = loc_name
return closest_location
def coords_equal(self, coord1: Tuple[float, float], coord2: Tuple[float, float], tolerance: float = 1e-5) -> bool:
return math.isclose(coord1[0], coord2[0], abs_tol=tolerance) and math.isclose(coord1[1], coord2[1], abs_tol=tolerance)
async def refresh_timezone(self, location: Union[Location, Tuple[float, float]], force: bool = False) -> str:
if isinstance(location, Location):
@ -221,16 +236,20 @@ class Geocoder:
else:
lat, lon = location
rounded_location = self.round_coords(lat, lon)
current_time = datetime.now()
if (force or
not self.last_update or
current_time - self.last_update > timedelta(hours=1) or
self.last_location != (lat, lon)):
not self.coords_equal(rounded_location, self.round_coords(*self.last_location) if self.last_location else (None, None))):
new_timezone = await self.timezone(lat, lon)
self.last_timezone = new_timezone
self.last_update = current_time
self.last_location = (lat, lon)
self.last_location = (lat, lon) # Store the original, non-rounded coordinates
await self.tz_save()
return self.last_timezone
async def tz_save(self):
@ -260,6 +279,17 @@ class Geocoder:
async def tz_last(self) -> Optional[str]:
await self.tz_cached()
return self.last_timezone
async def tz_at(self, lat: float, lon: float) -> str:
"""
Get the timezone at a specific latitude and longitude without affecting the cache.
:param lat: Latitude
:param lon: Longitude
:return: Timezone string
"""
return await self.timezone(lat, lon)
def __del__(self):
self.executor.shutdown()

36
sijapi/routers/loc.py
View file

@ -13,7 +13,7 @@ from zoneinfo import ZoneInfo
from dateutil.parser import parse as dateutil_parse
from typing import Optional, List, Union
from datetime import datetime
from sijapi import L, DB, TZ, NAMED_LOCATIONS, DynamicTZ, GEO
from sijapi import L, DB, TZ, NAMED_LOCATIONS, GEO
from sijapi.classes import Location
from sijapi.utilities import haversine
@ -35,8 +35,8 @@ async def dt(
# Handle provided timezone
if tz is not None:
if tz == "local":
last_loc = await get_last_location(date_time)
tz_str = DynamicTZ.find(last_loc.latitude, last_loc.longitude)
last_loc = await get_timezone_without_timezone(date_time)
tz_str = GEO.tz(last_loc.latitude, last_loc.longitude)
try:
tz = ZoneInfo(tz_str)
except Exception as e:
@ -50,7 +50,7 @@ async def dt(
L.ERR(f"Invalid timezone string '{tz}'. Error: {e}")
raise ValueError(f"Invalid timezone string: {tz}")
elif isinstance(tz, ZoneInfo):
pass # tz is already a ZoneInfo object
tz = tz # tz is already a ZoneInfo object
else:
raise ValueError("tz must be 'local', a string, or a ZoneInfo object.")
@ -64,18 +64,16 @@ async def dt(
# If no timezone provided, only fill in missing timezone info
elif date_time.tzinfo is None:
last_loc = get_last_location(date_time)
tz_str = DynamicTZ.find(last_loc.latitude, last_loc.longitude)
tz_str = await get_timezone_without_timezone(date_time)
try:
tz = ZoneInfo(tz_str)
except Exception as e:
L.WARN(f"Invalid timezone string '{tz_str}' from DynamicTZ. Falling back to UTC. Error: {e}")
L.WARN(f"Invalid timezone string '{tz_str}' from Geocoder. Falling back to UTC. Error: {e}")
tz = ZoneInfo('UTC')
date_time = date_time.replace(tzinfo=tz)
L.DEBUG(f"Filled in missing timezone info: {tz}")
# If datetime already has timezone and no new timezone provided, do nothing
else:
L.DEBUG(f"Datetime already has timezone {date_time.tzinfo}. No changes made.")
@ -87,6 +85,24 @@ async def dt(
L.ERR(f"Unexpected error in dt: {e}")
raise ValueError(f"Failed to localize datetime: {e}")
async def get_timezone_without_timezone(date_time):
# This is a bit convoluted because we're trying to solve the paradox of needing to know the location in order to determine the timezone, but needing the timezone to be certain we've got the right location if this datetime coincided with inter-timezone travel. Our imperfect solution is to use UTC for an initial location query to determine roughly where we were at the time, get that timezone, then check the location again using that timezone, and if this location is different from the one using UTC, get the timezone again usng it, otherwise use the one we already sourced using UTC.
# Step 1: Use UTC as an interim timezone to query location
interim_dt = date_time.replace(tzinfo=ZoneInfo("UTC"))
interim_loc = await fetch_last_location_before(interim_dt)
# Step 2: Get a preliminary timezone based on the interim location
interim_tz = await GEO.tz_current((interim_loc.latitude, interim_loc.longitude))
# Step 3: Apply this preliminary timezone and query location again
query_dt = date_time.replace(tzinfo=ZoneInfo(interim_tz))
query_loc = await fetch_last_location_before(query_dt)
# Step 4: Get the final timezone, reusing interim_tz if location hasn't changed
return interim_tz if query_loc == interim_loc else await GEO.tz_current(query_loc.latitude, query_loc.longitude)
async def get_last_location() -> Optional[Location]:
query_datetime = datetime.now(TZ)
L.DEBUG(f"Query_datetime: {query_datetime}")
@ -329,7 +345,7 @@ async def post_locate_endpoint(locations: Union[Location, List[Location]]):
# Prepare locations
for location in locations:
if not location.datetime:
tz = DynamicTZ.find(location.latitude, location.longitude)
tz = GEO.tz_current(location.latitude, location.longitude)
location.datetime = datetime.now(tz).isoformat()
if not location.context:

7
sijapi/routers/note.py
View file

@ -30,11 +30,10 @@ from dateutil.parser import parse as dateutil_parse
from fastapi import HTTPException, status
from pathlib import Path
from fastapi import APIRouter, Query, HTTPException
from sijapi import L, OBSIDIAN_VAULT_DIR, OBSIDIAN_RESOURCES_DIR, ARCHIVE_DIR, BASE_URL, OBSIDIAN_BANNER_SCENE, DEFAULT_11L_VOICE, DEFAULT_VOICE, TZ, DynamicTZ, GEO
from sijapi import L, OBSIDIAN_VAULT_DIR, OBSIDIAN_RESOURCES_DIR, ARCHIVE_DIR, BASE_URL, OBSIDIAN_BANNER_SCENE, DEFAULT_11L_VOICE, DEFAULT_VOICE, GEO
from sijapi.routers import cal, loc, tts, llm, time, sd, weather, asr
from sijapi.routers.loc import Location
from sijapi.utilities import assemble_journal_path, assemble_archive_path, convert_to_12_hour_format, sanitize_filename, convert_degrees_to_cardinal, HOURLY_COLUMNS_MAPPING
from sijapi.classes import Location
note = APIRouter()
@ -70,7 +69,7 @@ async def build_daily_note_endpoint(
date_str = dt_datetime.now().strftime("%Y-%m-%d")
if location:
lat, lon = map(float, location.split(','))
tz = ZoneInfo(DynamicTZ.find(lat, lon))
tz = GEO.tz_at(lat, lon)
date_time = dateutil_parse(date_str).replace(tzinfo=tz)
else:
raise ValueError("Location is not provided or invalid.")