110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
from typing import Any, Dict, Optional
|
|
|
|
|
|
FIELD_ALIASES = {
|
|
"country": ("country",),
|
|
"city": ("city",),
|
|
"latitude": ("latitude", "lat"),
|
|
"longitude": ("longitude", "lon", "lng"),
|
|
"value": ("value",),
|
|
"unit": ("unit",),
|
|
"cores": ("cores",),
|
|
"rmax": ("rmax", "r_max"),
|
|
"rpeak": ("rpeak", "r_peak"),
|
|
"power": ("power",),
|
|
}
|
|
|
|
NESTED_FIELD_ALIASES = {
|
|
"latitude": (
|
|
("location", "latitude"),
|
|
("location", "lat"),
|
|
("geo", "latitude"),
|
|
("geo", "lat"),
|
|
("coordinates", "latitude"),
|
|
("coordinates", "lat"),
|
|
),
|
|
"longitude": (
|
|
("location", "longitude"),
|
|
("location", "lon"),
|
|
("location", "lng"),
|
|
("geo", "longitude"),
|
|
("geo", "lon"),
|
|
("geo", "lng"),
|
|
("coordinates", "longitude"),
|
|
("coordinates", "lon"),
|
|
("coordinates", "lng"),
|
|
),
|
|
}
|
|
|
|
|
|
def get_metadata_field(metadata: Optional[Dict[str, Any]], field: str, fallback: Any = None) -> Any:
|
|
if isinstance(metadata, dict):
|
|
for key in FIELD_ALIASES.get(field, (field,)):
|
|
value = metadata.get(key)
|
|
if value not in (None, ""):
|
|
return value
|
|
for path in NESTED_FIELD_ALIASES.get(field, ()):
|
|
current: Any = metadata
|
|
for key in path:
|
|
if not isinstance(current, dict):
|
|
current = None
|
|
break
|
|
current = current.get(key)
|
|
if current not in (None, ""):
|
|
return current
|
|
if field in {"latitude", "longitude"}:
|
|
value = _get_coordinate_sequence_value(metadata, field)
|
|
if value not in (None, ""):
|
|
return value
|
|
return fallback
|
|
|
|
|
|
def _get_coordinate_sequence_value(metadata: Dict[str, Any], field: str) -> Any:
|
|
for key in ("coordinates", "coord", "coords"):
|
|
value = metadata.get(key)
|
|
if not isinstance(value, (list, tuple)) or len(value) < 2:
|
|
continue
|
|
# GeoJSON uses [longitude, latitude]. Most raw collector tuples in this
|
|
# codebase use explicit field names, so only sequence aliases are treated
|
|
# as GeoJSON-shaped to avoid guessing.
|
|
return value[1] if field == "latitude" else value[0]
|
|
return None
|
|
|
|
|
|
def build_dynamic_metadata(
|
|
metadata: Optional[Dict[str, Any]],
|
|
*,
|
|
country: Any = None,
|
|
city: Any = None,
|
|
latitude: Any = None,
|
|
longitude: Any = None,
|
|
value: Any = None,
|
|
unit: Any = None,
|
|
) -> Dict[str, Any]:
|
|
merged = dict(metadata) if isinstance(metadata, dict) else {}
|
|
|
|
fallbacks = {
|
|
"country": country,
|
|
"city": city,
|
|
"latitude": latitude,
|
|
"longitude": longitude,
|
|
"value": value,
|
|
"unit": unit,
|
|
}
|
|
|
|
for field, fallback in fallbacks.items():
|
|
if fallback not in (None, "") and get_metadata_field(merged, field) in (None, ""):
|
|
merged[field] = fallback
|
|
|
|
return merged
|
|
|
|
|
|
def get_record_field(record: Any, field: str) -> Any:
|
|
metadata = getattr(record, "extra_data", None) or {}
|
|
fallback_attr = field
|
|
if field in {"cores", "rmax", "rpeak", "power"}:
|
|
fallback = None
|
|
else:
|
|
fallback = getattr(record, fallback_attr, None)
|
|
return get_metadata_field(metadata, field, fallback=fallback)
|