-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcache_utils.py
More file actions
56 lines (45 loc) · 1.99 KB
/
Copy pathcache_utils.py
File metadata and controls
56 lines (45 loc) · 1.99 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
"""
Shared utilities for per-location and time-bucketed caching.
- Normalizes observer location (lat/lon to 4 decimals, elevation to nearest 10 m)
- Generates 6-hour UTC time buckets (00, 06, 12, 18)
- Location keys for PostgreSQL cache
"""
from __future__ import annotations
import math
from datetime import datetime, timezone
from typing import Optional, Tuple
DEFAULT_BUCKET_HOURS = 6
DEFAULT_TTL_SECONDS = 6 * 3600 # 6 hours
def normalize_location(lat: float, lon: float, elevation: float,
latlon_decimals: int = 4, elevation_step: int = 10) -> Tuple[float, float, int]:
"""Round lat/lon to 4 decimals (~11 m) and elevation to next higher 10 m step.
Returns (lat_norm, lon_norm, elev_norm_int).
"""
try:
lat_f = float(lat)
lon_f = float(lon)
elev_f = float(elevation)
except Exception:
lat_f, lon_f, elev_f = 0.0, 0.0, 0.0
lat_n = round(lat_f, latlon_decimals)
lon_n = round(lon_f, latlon_decimals)
elev_n = int(math.ceil(elev_f / float(elevation_step)) * elevation_step)
return lat_n, lon_n, elev_n
def location_key(lat: float, lon: float, elevation: float) -> str:
"""Build a stable key string from normalized location values.
Example: 'lat+48.2082_lon+16.3738_el+0170'
"""
return f"lat{lat:+.4f}_lon{lon:+.4f}_el{int(elevation):+05d}"
def time_bucket_utc(dt: Optional[datetime] = None, bucket_hours: int = DEFAULT_BUCKET_HOURS) -> str:
"""Return a 6-hour UTC bucket label like 'YYYYMMDDTHH'. Aligns HH to 00/06/12/18.
"""
if dt is None:
dt = datetime.now(timezone.utc)
if isinstance(dt, (list, tuple)):
dt = dt[0] if dt else datetime.now(timezone.utc)
if not isinstance(dt, datetime):
raise TypeError(f"time_bucket_utc expected datetime or sequence of datetimes, got {type(dt).__name__}")
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
bucket_hour = (dt.hour // bucket_hours) * bucket_hours
return f"{dt:%Y%m%d}T{bucket_hour:02d}"