Skip to content

Arco surface

arco_surface

Targeted ERA5 top-of-atmosphere fluxes from the ARCO pressure-level store.

ArcoRadiationSource

Hourly ERA5 flux averages, represented at interval midpoints in W/m².

ARCO stores one-hour energies (J/m²) at interval ends. This source divides them by 3600 and moves the interpolation coordinate back 30 minutes, exactly once. It preserves ERA5's signed net-flux convention. These are top-of-atmosphere fields, not surface irradiance. Only requested fields and occupied latitude blocks are read through the existing range reader.

grid_source permits local mirrors/tests; it must use the ERA5 ARCO time and spatial axes. The default opens the public pressure-level store.

Source code in packages/barograph-met/src/baro/met/arco_surface.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class ArcoRadiationSource:
    """Hourly ERA5 flux averages, represented at interval midpoints in W/m².

    ARCO stores one-hour energies (J/m²) at interval ends. This source divides
    them by 3600 and moves the interpolation coordinate back 30 minutes,
    exactly once. It preserves ERA5's signed net-flux convention. These are
    top-of-atmosphere fields, not surface irradiance. Only requested fields
    and occupied latitude blocks are read through the existing range reader.

    ``grid_source`` permits local mirrors/tests; it must use the ERA5 ARCO
    time and spatial axes. The default opens the public pressure-level store.
    """

    time_step = np.timedelta64(1, "h")
    time_origin = np.datetime64("1970-01-01T00:30", "ns")

    def __init__(
        self,
        *,
        fields: Sequence[str] = _FIELDS,
        base_url: str = ARCO_BASE_URL,
        grid_source: GridSource | None = None,
    ) -> None:
        names = tuple(fields)
        if not names or len(set(names)) != len(names) or set(names) - set(_FIELDS):
            raise ValueError("select unique ERA5 top net solar/thermal radiation fields")
        self.grid = grid_source or arco_pressure_levels(
            base_url=base_url, fields={name: name for name in names}
        )
        if any(self.grid.fields.get(name) != name for name in names):
            raise ValueError("grid_source must be configured with the requested radiation arrays")
        self.grid = self.grid.select_fields(names)
        # TOA is a radiative boundary, not a finite height above ground.
        self.variables = {name: SurfaceVariable("W m-2", float("nan")) for name in names}
        self._scale = {name: _flux_scale(self.grid, name) for name in names}

    @property
    def stats(self) -> list[TransferStats]:
        """Underlying transfer counters, including byte-cache hits."""
        return self.grid.stats

    def coverage(self) -> tuple[np.datetime64, np.datetime64]:
        """Effective interpolation coverage after the ERA5 midpoint shift."""
        bounds = self.grid.coverage()
        if bounds is None:
            raise DataUnavailableError("ERA5 radiation coverage is unavailable")
        return bounds[0] - _HALF_HOUR, bounds[1] - _HALF_HOUR

    def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
        """Midpoint samples bracketing the actual query time."""
        end = np.datetime64(t0 + _HALF_HOUR, "h")
        return CubeKey(end - _HALF_HOUR), CubeKey(end + _HALF_HOUR)

    def tiling(self, key: CubeKey) -> Tiling | None:
        """Block-sized latitude bands shared with the range reader."""
        return self.grid.tiling(CubeKey(key.valid + _HALF_HOUR))

    def use_cache(self, cache: RangeCache) -> None:
        """Cache only immutable ERA5 archive ranges."""
        self.grid.use_cache(cache)

    def fetch(
        self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
    ) -> SurfaceCube:
        """Read a midpoint field without downloading unrelated pressure variables."""
        names = tuple(self.variables if fields is None else fields)
        if set(names) - self.variables.keys():
            raise ValueError("requested radiation field is not configured")
        end = key.valid + _HALF_HOUR
        if end != np.datetime64(end, "h") or key.run is not None:
            raise ValueError("ERA5 radiation keys must be hourly interval midpoints, without run")
        try:
            index = self.grid.time_index(end)
        except KeyError as exc:
            raise DataUnavailableError(f"ERA5 radiation: {end} outside archive") from exc
        grid = GridIndex.locate(domain, self.grid.lat, self.grid.lon)
        values = {
            name: (
                self.grid.store.read_rows(name, [(index,)], grid.rows)[0][:, grid.cols][::-1]
                * self._scale[name]
            ).astype(np.float32)
            for name in names
        }
        return SurfaceCube(
            key, grid.lat[::-1], grid.lon, values, {n: self.variables[n] for n in names}
        )
stats property

Underlying transfer counters, including byte-cache hits.

coverage()

Effective interpolation coverage after the ERA5 midpoint shift.

Source code in packages/barograph-met/src/baro/met/arco_surface.py 
76
77
78
79
80
81
def coverage(self) -> tuple[np.datetime64, np.datetime64]:
    """Effective interpolation coverage after the ERA5 midpoint shift."""
    bounds = self.grid.coverage()
    if bounds is None:
        raise DataUnavailableError("ERA5 radiation coverage is unavailable")
    return bounds[0] - _HALF_HOUR, bounds[1] - _HALF_HOUR
cube_keys(t0)

Midpoint samples bracketing the actual query time.

Source code in packages/barograph-met/src/baro/met/arco_surface.py 
83
84
85
86
def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
    """Midpoint samples bracketing the actual query time."""
    end = np.datetime64(t0 + _HALF_HOUR, "h")
    return CubeKey(end - _HALF_HOUR), CubeKey(end + _HALF_HOUR)
fetch(key, domain, *, fields=None)

Read a midpoint field without downloading unrelated pressure variables.

Source code in packages/barograph-met/src/baro/met/arco_surface.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def fetch(
    self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
) -> SurfaceCube:
    """Read a midpoint field without downloading unrelated pressure variables."""
    names = tuple(self.variables if fields is None else fields)
    if set(names) - self.variables.keys():
        raise ValueError("requested radiation field is not configured")
    end = key.valid + _HALF_HOUR
    if end != np.datetime64(end, "h") or key.run is not None:
        raise ValueError("ERA5 radiation keys must be hourly interval midpoints, without run")
    try:
        index = self.grid.time_index(end)
    except KeyError as exc:
        raise DataUnavailableError(f"ERA5 radiation: {end} outside archive") from exc
    grid = GridIndex.locate(domain, self.grid.lat, self.grid.lon)
    values = {
        name: (
            self.grid.store.read_rows(name, [(index,)], grid.rows)[0][:, grid.cols][::-1]
            * self._scale[name]
        ).astype(np.float32)
        for name in names
    }
    return SurfaceCube(
        key, grid.lat[::-1], grid.lon, values, {n: self.variables[n] for n in names}
    )
tiling(key)

Block-sized latitude bands shared with the range reader.

Source code in packages/barograph-met/src/baro/met/arco_surface.py 
88
89
90
def tiling(self, key: CubeKey) -> Tiling | None:
    """Block-sized latitude bands shared with the range reader."""
    return self.grid.tiling(CubeKey(key.valid + _HALF_HOUR))
use_cache(cache)

Cache only immutable ERA5 archive ranges.

Source code in packages/barograph-met/src/baro/met/arco_surface.py 
92
93
94
def use_cache(self, cache: RangeCache) -> None:
    """Cache only immutable ERA5 archive ranges."""
    self.grid.use_cache(cache)