class AromeSurfaceSource:
"""Hourly surface fields from one explicit AROME initialization (0..51 h).
``fields`` maps output names to u10, v10, t2m, precipitation or gusts.
Precipitation is a preceding-hour amount, derived from cumulative totals
within the same run. Gusts retain the reported preceding-hour maximum.
Only the requested GRIB messages are downloaded; their immutable ranges
and per-file indexes use the shared range cache. No altitude is required.
``published_runs`` describes files visible now, never historical availability.
"""
def __init__(
self,
*,
run: np.datetime64 | datetime,
fields: dict[str, str] | None = None,
base_url: str = "https://files.data.gouv.fr/meteofrance-pnt/pnt",
index_dir: str | Path | None = None,
) -> None:
stamp = np.datetime64(run, "ns")
hour = stamp.astype("datetime64[h]")
if np.isnat(stamp) or stamp != hour or int(str(hour)[11:13]) % 3:
raise ValueError("AROME run must be an exact UTC three-hour initialization")
self._forecast = ArpegeSource(
run=hour, run_hours=range(0, 24, 3), base_url=base_url, index_dir=index_dir
)
self.run = hour
self.observed_at: np.datetime64 | None = None
self.time_step = np.timedelta64(1, "h")
self.indexes = self._forecast.indexes
self.fields = _checked_fields(fields)
self.variables = {out: _VARIABLES[name] for out, name in self.fields.items()}
self.source_id = "arome-0025-surface"
self._forecast.source_id = self.source_id
self._reader = _NativeFields(self._forecast, self.run, self.file_key)
if index_dir is not None:
self.indexes.directory = Path(index_dir).expanduser() / self.source_id
def __enter__(self) -> AromeSurfaceSource:
return self
def __exit__(self, *exc: object) -> None:
self.close()
@classmethod
def latest(cls, valid_time: np.datetime64, **kwargs: Any) -> AromeSurfaceSource:
"""Newest run whose required files are available now for ``valid_time``.
``observed_at`` records when publication was checked. It is not a
historical publication timestamp and cannot establish a past cutoff.
"""
if np.isnat(valid_time):
raise ValueError("valid time cannot be NaT")
probe = cls(run=np.datetime64("2000-01-01T00"), **kwargs)
try:
run = probe._latest_available_run(valid_time)
result = cls(run=run, **kwargs)
result.observed_at = np.datetime64("now", "s")
return result
finally:
probe.close()
def _latest_available_run(self, valid_time: np.datetime64) -> np.datetime64:
"""Check the required native files of candidate initializations."""
for run in reversed(self.published_runs()):
leads = _required_leads(valid_time, run, self.fields)
if leads and all(self._forecast._published(self.file_key(run, lead)) for lead in leads):
return run
raise DataUnavailableError("no currently published AROME run covers this valid time")
@property
def stats(self) -> list[TransferStats]:
return self._forecast.stats
def close(self) -> None:
close = getattr(self._forecast.store, "close", None)
if close is not None:
close()
def tiling(self, key: CubeKey) -> Tiling | None:
return None
def use_cache(self, cache: RangeCache) -> None:
self._forecast.use_cache(cache)
def published_runs(self) -> list[np.datetime64]:
return self._forecast.published_runs()
def coverage(self, max_age_s: float = 600) -> tuple[np.datetime64, np.datetime64]:
return self.run, self.run + np.timedelta64(51, "h")
def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
hour = np.datetime64(t0, "h")
return CubeKey(hour, self.run), CubeKey(hour + self.time_step, self.run)
def file_key(self, run: np.datetime64, lead_h: int) -> str:
for start, end in _RANGES:
if start <= lead_h <= end:
stamp = str(np.datetime64(run, "h")) + ":00:00Z"
name = f"arome__0025__SP1__{start:02d}H{end:02d}H__{stamp}.grib2"
return f"{stamp}/arome/0025/SP1/{name}"
raise DataUnavailableError(f"AROME lead {lead_h} h outside 0..51")
def fetch(
self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
) -> SurfaceCube:
lead = _forecast_lead(key, self.run)
self.file_key(self.run, lead)
selected = self.fields if fields is None else {out: self.fields[out] for out in fields}
if not selected:
raise ValueError("at least one surface field is required")
arrays = {name: self._reader.field(name, lead) for name in set(selected.values())}
_, lat, lon = next(iter(arrays.values()))
for _, field_lat, field_lon in arrays.values():
_require_same_grid((lat, lon), (field_lat, field_lon))
# The shared GRIB decoder emits descending latitude.
grid = GridIndex.locate(domain, lat, lon)
cube_fields = {
out: arrays[name][0][grid.rows][:, grid.cols][::-1].copy()
for out, name in selected.items()
}
return SurfaceCube(
key,
grid.lat[::-1].copy(),
grid.lon,
cube_fields,
{out: self.variables[out] for out in selected},
)