63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
from typing import Any, Dict, Optional
|
|
|
|
|
|
FIELD_ALIASES = {
|
|
"country": ("country",),
|
|
"city": ("city",),
|
|
"latitude": ("latitude",),
|
|
"longitude": ("longitude",),
|
|
"value": ("value",),
|
|
"unit": ("unit",),
|
|
"cores": ("cores",),
|
|
"rmax": ("rmax", "r_max"),
|
|
"rpeak": ("rpeak", "r_peak"),
|
|
"power": ("power",),
|
|
}
|
|
|
|
|
|
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
|
|
return fallback
|
|
|
|
|
|
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)
|