Firecube 101: NetCDF To Zarr¶

lat = np.linspace(30, 40, 40)
lon = np.linspace(-10, 5, 60)
dates = pd.date_range("2024-07-01", periods=6, freq="D")
x, y = np.meshgrid(lon, lat)

for day, date in enumerate(dates):
    temperature = 18 + 0.4 * day + 2 * np.exp(-((x + 7 - 2 * day) ** 2 + (y - 35) ** 2) / 8)
    dataset = xr.Dataset(
        {"sst": (("time", "lat", "lon"), temperature[None].astype("float32"))},
        coords={"time": [date], "lat": lat, "lon": lon},
    )
    dataset.sst.attrs = {"units": "degC", "long_name": "Synthetic sea surface temperature"}
    arrival = first_arrival if day < 3 else second_arrival
    dataset.to_netcdf(arrival / f"sst-{day + 1:02d}.nc", engine="h5netcdf")
print("Ready: 3 files for the first arrival + 3 for the second")

!firecube plugins create incremental-sst \
    --template zarr \
    --target-dir "{work}" \
    --license Apache-2.0 \
    --non-interactive
assert _exit_code == 0, "firecube plugins create failed - see the output above"
plugin = work / "firecube-incremental-sst"
plugin_file = plugin / "src/firecube_incremental_sst/ingestor.py"

%%writefile {plugin_file}
# Copyright 2026 EUMETSAT
# SPDX-License-Identifier: Apache-2.0
from typing import ClassVar

import xarray as xr

from firecube.ingestor.api import GenericZarrIngestor, register_ingestor


@register_ingestor("incremental_sst")
class IncrementalSstIngestor(GenericZarrIngestor):
    PRODUCT_NAME: ClassVar[str] = "incremental_sst"
    time_dim_name: ClassVar[str] = "time"

    def build_dataset(self, group, items, ctx):
        if not items:
            return None
        datasets = []
        for item in items:
            with xr.open_dataset(ctx.materialize(item), engine="h5netcdf") as source:
                datasets.append(source.load())
        return xr.concat(datasets, dim="time", join="exact").sortby("time")

!firecube plugins install --editable "{plugin}"
assert _exit_code == 0, "firecube plugins install failed - see the output above"
!firecube plugins list

!firecube ingest incremental_sst \
    --input-data "{first_arrival}" \
    --target "{target_uri}" \
    --product-name incremental_sst \
    --write-mode direct \
    --option no_progress=true \
    > "{work}/first-arrival.json"
assert _exit_code == 0, "firecube ingest failed - see the log above"
with xr.open_zarr(target, group="default", consolidated=False) as saved:
    before = saved[["sst"]].load()
assert before.sizes["time"] == 3
print("Cube: 3 days · 1-3 July")

!firecube ingest incremental_sst \
    --input-data "{second_arrival}" \
    --target "{target_uri}" \
    --product-name incremental_sst \
    --write-mode direct \
    --option resume_existing=true \
    --option no_progress=true \
    > "{work}/second-arrival.json"
assert _exit_code == 0, "firecube ingest failed - see the log above"
with xr.open_zarr(target, group="default", consolidated=False) as saved:
    after = saved[["sst"]].load()
assert after.sizes["time"] == 6
xr.testing.assert_identical(after.isel(time=slice(0, 3)), before)
print("Cube: 6 days · 1-6 July · first three days unchanged")

mean = after.sst.mean(("lat", "lon"))
figure, axis = plt.subplots(figsize=(7, 3.6), layout="constrained")
axis.plot(dates, mean, color="#b8bec6", linewidth=1.5, zorder=1)
axis.scatter(dates[:3], mean[:3], s=65, color="#246b91", label="First arrival · 3 days", zorder=2)
axis.scatter(dates[3:], mean[3:], s=65, color="#cd592a", label="Appended · 3 days", zorder=2)
axis.axvline(dates[2] + pd.Timedelta(hours=12), color="#8b929a", linestyle="--", linewidth=1)
axis.set(
    title="One Zarr cube: 3 days → 6 days", ylabel="Mean synthetic SST (°C)", xlabel="July 2024"
)
axis.set_xticks(dates, [str(day) for day in range(1, 7)])
axis.spines[["top", "right"]].set_visible(False)
axis.legend(frameon=False, loc="upper left")
axis.grid(axis="y", alpha=0.2)
figure.savefig(work / "incremental-preview.png", dpi=180)
plt.show()

!firecube ingest incremental_sst \
    --input-data "{second_arrival}" \
    --target "{target_uri}" \
    --product-name incremental_sst \
    --write-mode direct \
    --option resume_existing=true \
    --option no_progress=true \
    > "{work}/retry.json" 2> "{work}/retry.log"
print((work / "retry.log").read_text())
if _exit_code:
    assert "overlap" in (work / "retry.log").read_text().lower(), (
        "Unexpected failure; see the log above"
    )
with xr.open_zarr(target, group="default", consolidated=False) as saved:
    retried = saved[["sst"]].load()
xr.testing.assert_identical(retried, after)
assert len(np.unique(retried.time)) == 6
print("Still 6 unique days · values unchanged")