PM-pinou
ef60e3e489
3.1 Unified TLS_HEX_MAP: defined once in type_classifier.py, imported in globe.py and clustering.py 3.2 Created analysis/constants.py: centralized RANDOM_SEED, MAX_SAMPLE_ROWS, UMAP constants, GLOBE_MAX_ROWS, etc. 3.3 Removed 14 pointless try/except:raise dead code blocks across tools/ and views/ 3.4 Added logger.error(traceback) to all bare except Exception: blocks in tools/ and views/ 3.5 Moved inline import traceback/asyncio to top-level imports in features.py, clustering.py, pipeline.py 3.6 Removed auto_profile_module side-effect imports from 6 modules
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""Geo extraction helpers and plan utilities."""
|
|
import traceback
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
import json as _json
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _extract_lat(val):
|
|
"""Try to extract a latitude from *val* (number or str)."""
|
|
if val is None:
|
|
return None
|
|
try:
|
|
v = float(val)
|
|
return None if v < -90 or v > 90 else v
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _extract_lon(val):
|
|
"""Try to extract a longitude from *val* (number or str)."""
|
|
if val is None:
|
|
return None
|
|
try:
|
|
v = float(val)
|
|
return None if v < -180 or v > 180 else v
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
# ── Plan index utilities (shared by tools.py and auto.py) ─────────────
|
|
|
|
_PLANS_DIR = Path(__file__).resolve().parent.parent.parent / '.omo' / 'plans'
|
|
_PLANS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
_AUTO_INDEX_PATH = _PLANS_DIR / '_auto_index.json'
|
|
_MAX_AUTO_PLANS = 10
|
|
|
|
|
|
def _read_auto_index():
|
|
"""Read the auto index file, returning list of {file, pinned}."""
|
|
if not _AUTO_INDEX_PATH.exists():
|
|
return []
|
|
try:
|
|
data = _json.loads(_AUTO_INDEX_PATH.read_text(encoding='utf-8'))
|
|
if not isinstance(data, list):
|
|
return []
|
|
return data
|
|
except Exception:
|
|
logger.error("_read_auto_index failed: {}".format(traceback.format_exc()))
|
|
return []
|
|
|
|
|
|
def _write_auto_index(index):
|
|
"""Write the auto index list."""
|
|
_AUTO_INDEX_PATH.write_text(
|
|
_json.dumps(index, ensure_ascii=False, indent=2),
|
|
encoding='utf-8',
|
|
)
|
|
|
|
|
|
def _add_to_auto_index(filename):
|
|
"""Prepend filename to auto index, evict unpinned entries when >10."""
|
|
index = _read_auto_index()
|
|
# Remove if already present
|
|
index = [entry for entry in index if entry.get('file') != filename]
|
|
# Prepend
|
|
index.insert(0, {'file': filename, 'pinned': False})
|
|
# Evict unpinned entries beyond max
|
|
pinned_entries = [e for e in index if e.get('pinned')]
|
|
unpinned_entries = [e for e in index if not e.get('pinned')]
|
|
# Keep all pinned + fill remaining slots with unpinned
|
|
result = pinned_entries + unpinned_entries[:_MAX_AUTO_PLANS - len(pinned_entries)]
|
|
_write_auto_index(result)
|