Slot-Based Parallelism: MTG FCI L1C¶

%pip install -r https://eumetsat.github.io/firecube/latest/showcase/benchmark-requirements.txt

import json
import os
import re
import sys
import tempfile
import time
from pathlib import Path

os.environ["PATH"] = str(Path(sys.executable).parent) + os.pathsep + os.environ["PATH"]
source = Path("inputs/fci").resolve()
grids = Path("fci_grids.npz").resolve()
output = Path("outputs/fci-benchmarks").resolve()
output.mkdir(parents=True, exist_ok=True)

!eumdac download \
    -c EO:EUM:DAT:0662 \
    -s 2025-07-01T00:00:00 \
    -e 2025-07-01T01:59:59 \
    --limit 12 \
    -o "{source}" \
    --integrity
assert _exit_code == 0, "eumdac download failed - see the output above"

files = sorted(source.glob("*.zip"))
expected = [f"20250701{s // 6:02d}{s % 6 * 10:02d}" for s in range(12)]
timestamps = [re.search(r"_OPE_(\d{12})\d{2}_", p.name) for p in files]
assert all("FCI-1C-RRAD-FDHSI" in p.name for p in files)
assert all(timestamps) and sorted(m[1] for m in timestamps) == expected, (
    "Expected all 12 acquisitions"
)
print(f"Ready: {len(files)} acquisitions")

!firecube plugins mtg_fci_l1c geo generate --resolutions 1km,2km --output "{grids}"
assert _exit_code == 0, "firecube plugins mtg_fci_l1c geo generate failed - see the output above"

processes = 12  # Change to 1, 2, 4 or 6 to compare scaling.
assert processes in (1, 2, 4, 6, 12)
assert grids.is_file(), "Generate the shared grids first"
run_dir = Path(tempfile.mkdtemp(prefix=f"p{processes}-", dir=output))
target = run_dir / "cube.zarr"
target_uri = target.as_uri()
scratch_dir = run_dir / "scratch"

# Every firecube command below takes these flags; {common} inserts them.
common = f"""
--target "{target_uri}"
--product-name mtg_fci_l1c
--storage-type local
--storage-driver fsspec
--write-mode direct
--option product_type=FDHSI
--option time_epoch=2025-07-01
--option time_slots=12
--option resolutions=1km,2km
--option pixel_time_dtype=float64
--option include_pixel_time=true
--option include_pixel_quality=true
--option include_calibration=true
--option include_geolocation=true
--option pipeline_workers=1
--option pipeline_batch_size=10
--option extract_workers=4
--option zarr_sharding=false
--option zarr_region_write_concurrency=8
--option zarr_write_empty_chunks=false
--option "fci_grids_file={grids}"
--option "scratch_dir={scratch_dir}"
""".split("\n")
common = " ".join(flag for flag in common if flag)
print(f"{processes} processes → {target}")

!firecube zarr preallocate mtg_fci_l1c {common} \
    --input-data "{source}" \
    --slot-start 0 \
    --slot-end 12 \
    > "{run_dir}/preallocate.log" 2>&1
assert _exit_code == 0, f"firecube zarr preallocate failed - inspect {run_dir}/preallocate.log"
print("Cube ready: 12 time slots at 1 km and 2 km")

slot_size = 12 // processes
!firecube zarr slots mtg_fci_l1c {common} \
    --slot-size {slot_size} \
    --no-resume \
    --format json \
    > "{run_dir}/plan.json"
assert _exit_code == 0, "firecube zarr slots failed - see the output above"
plan = json.loads((run_dir / "plan.json").read_text())
ranges = [(start, start + slot_size) for start in range(0, 12, slot_size)]
assert {g["name"] for g in plan["groups"]} == {"data_1km", "data_2km"}
for group in plan["groups"]:
    actual = sorted(
        (r["slot_start"], r["slot_end"]) for r in plan["ranges"] if r["group"] == group["name"]
    )
    assert actual == ranges and group["static_owner"]["slot_start"] == 0
print("Slot ranges (end excluded):", ranges)

starts = " ".join(str(start) for start, _ in ranges)
started = time.perf_counter()
!for lo in {starts}; do \
    hi=$(expr $$lo + {slot_size}); \
    (firecube ingest mtg_fci_l1c {common} \
        --input-data "{source}" \
        --option force_reingest=true \
        --suppress-static-emission-for-non-owner \
        --static-owner-slot-start 0 \
        --slot-start $$lo \
        --slot-end $$hi \
        > "{run_dir}/slot-$$lo-$$hi.log" 2>&1 \
        || touch "{run_dir}/slot-$$lo-$$hi.failed") & \
done; wait
wall_s = time.perf_counter() - started
failed = sorted(path.name for path in run_dir.glob("slot-*.failed"))
assert not failed, f"Ingest failed: inspect {run_dir}/slot-*.log"
print(f"Ingestion: {wall_s:.2f} seconds")

import pandas as pd
import zarr

!firecube zarr slots mtg_fci_l1c {common} --format json > "{run_dir}/coverage.json"
assert _exit_code == 0, "firecube zarr slots failed - see the output above"
coverage = json.loads((run_dir / "coverage.json").read_text())
assert {g["name"] for g in coverage["groups"]} == {"data_1km", "data_2km"}
assert not coverage["ranges"]
assert all(g["covered_ranges"] == [[0, 12]] for g in coverage["groups"])
cube = zarr.open_group(str(target), mode="r")
for name in ("data_1km", "data_2km"):
    assert cube[name]["counts"].shape[0] == 12
    for field in ("latitude", "longitude", "x", "y"):
        assert cube[name][field].attrs.get("firecube_static_written")
measurement = pd.DataFrame(
    [
        {
            "processes": processes,
            "slots": 12,
            "wall_s": wall_s,
            "output": str(target),
        }
    ]
)
measurement.to_csv(output / f"measurements-{run_dir.name}.csv", index=False)
print("All 12 slots complete in both groups")
measurement