Sentinel-3 SLSTR Level 2 Fire Radiative Power¶

def get_observations():
    if manifest_path.exists():
        manifest = json.loads(manifest_path.read_text())
        if manifest["query"] == query and all(
            (input_directory / item["filename"]).exists() for item in manifest["products"]
        ):
            print(f"Reusing {len(manifest['products'])} downloaded tables.")
            return manifest

    credentials = (getpass("EUMETSAT consumer key: "), getpass("EUMETSAT consumer secret: "))
    token = eumdac.AccessToken(credentials, cache=False)
    del credentials
    store = eumdac.DataStore(token)
    results = store.get_collection(COLLECTION).search(**query)
    print(f"Found {len(results)} products. Downloading their standard MWIR tables...")
    products = []
    for index, product in enumerate(results, start=1):
        product_id = str(product)
        entry = f"{product_id}/FRP_MWIR1km_standard.csv"
        destination = input_directory / f"{product_id}.csv"
        if not destination.exists():
            partial = destination.with_suffix(".csv.part")
            try:
                with product.open(entry=entry) as source, partial.open("wb") as target:
                    shutil.copyfileobj(source, target)
                partial.replace(destination)
            finally:
                partial.unlink(missing_ok=True)
        products.append(
            {
                "product": product_id,
                "entry": entry,
                "filename": destination.name,
                "sha256": hashlib.sha256(destination.read_bytes()).hexdigest(),
                "bytes": destination.stat().st_size,
            }
        )
        if index % 25 == 0:
            print(f"{index} tables ready")
    if not products:
        raise ValueError("No products found. Check the dates and bounding box.")
    manifest = {"collection": COLLECTION, "query": query, "products": products}
    manifest_path.write_text(json.dumps(manifest, indent=2))
    print(f"Ready: {len(products)} tables, {sum(p['bytes'] for p in products) / 1e6:.1f} MB.")
    return manifest


manifest = get_observations()
source_files = [input_directory / item["filename"] for item in manifest["products"]]
for item, path in zip(manifest["products"], source_files, strict=True):
    assert hashlib.sha256(path.read_bytes()).hexdigest() == item["sha256"]

plugin_directory = Path("plugins/firecube-sentinel3-frp")
if not plugin_directory.exists():
    !firecube plugins create sentinel3-frp \
        --template parquet \
        --target-dir plugins \
        --license Apache-2.0 \
        --non-interactive

print("Plugin source:", plugin_directory / "src/firecube_sentinel3_frp/ingestor.py")

%%writefile plugins/firecube-sentinel3-frp/src/firecube_sentinel3_frp/ingestor.py
# Copyright 2026 EUMETSAT
# SPDX-License-Identifier: Apache-2.0

import json
from pathlib import Path
from typing import ClassVar

import pandas as pd

from firecube.ingestor.api import (
    GenericParquetIngestor,
    PipelineBatch,
    PluginContext,
    register_ingestor,
)


@register_ingestor("sentinel3_frp")
class Sentinel3FrpIngestor(GenericParquetIngestor):
    PRODUCT_NAME: ClassVar[str] = "sentinel3_frp"

    def discover_source_files(self, ctx: PluginContext):
        directory = Path(ctx.source)
        manifest = json.loads((directory / "manifest.json").read_text())
        return [directory / item["filename"] for item in manifest["products"]]

    def build_dataset(
        self, group: str, batch: PipelineBatch, ctx: PluginContext
    ) -> pd.DataFrame | None:
        frames = []
        for item in batch.items:
            frame = pd.read_csv(
                ctx.materialize(item),
                comment="#",
                usecols=["day", "time", "lat(deg)", "lon(deg)", "FRP(MW)"],
                dtype={"lat(deg)": float, "lon(deg)": float, "FRP(MW)": float},
            )
            if frame.empty:
                continue
            frame["timestamp"] = pd.to_datetime(
                frame["day"].astype(str) + "T" + frame["time"].astype(str), utc=True
            )
            frame = frame.drop(columns=["day", "time"])
            frame["source_product"] = Path(str(item)).name.removesuffix(".csv")
            frames.append(frame)
        return pd.concat(frames, ignore_index=True) if frames else None

project_file = plugin_directory / "pyproject.toml"
project_text = project_file.read_text()
if '\n    "pandas",' not in project_text:
    project_file.write_text(
        project_text.replace("dependencies = [", 'dependencies = [\n    "pandas",')
    )

!firecube plugins install --editable plugins/firecube-sentinel3-frp
!firecube plugins describe sentinel3_frp

Path("outputs").mkdir(exist_ok=True)
run_directory = Path(mkdtemp(prefix="spain-fires-", dir="outputs")).resolve()
output_path = run_directory / "detections.parquet"
product_uri = output_path.as_uri()
summary_path = run_directory / "run-summary.json"
source_directory = str(input_directory.resolve())

!firecube ingest sentinel3_frp \
    --input-data "{source_directory}" \
    --target "{product_uri}" \
    --product-name sentinel3_frp \
    --output-format parquet \
    --write-mode direct \
    --option pipeline_workers=1 \
    --option pipeline_batch_size=25 \
    --option no_progress=true \
    > "{summary_path}"

print("Parquet output:", output_path.relative_to(Path.cwd()))

observations = pd.read_parquet(output_path).rename(
    columns={"lat(deg)": "latitude", "lon(deg)": "longitude", "FRP(MW)": "frp_mw"}
)
in_area = observations["longitude"].between(WEST, EAST) & observations["latitude"].between(
    SOUTH, NORTH
)
in_month = observations["timestamp"].ge(pd.Timestamp(START, tz="UTC")) & observations[
    "timestamp"
].lt(pd.Timestamp(END, tz="UTC"))
detections = (
    observations.loc[in_area & in_month].sort_values(["timestamp", "latitude", "longitude"]).copy()
)

if detections.empty:
    raise ValueError("No detections in this area and month. Check the selection.")
detections["day"] = detections["timestamp"].dt.strftime("%Y-%m-%d")
days = pd.date_range(START, END, inclusive="left").strftime("%Y-%m-%d")
daily_counts = detections.groupby("day").size().reindex(days, fill_value=0)
print(f"{len(detections):,} detections inside the bounding box during July 2026.")
ax = daily_counts.plot.bar(figsize=(10, 3), color="#dc5a27")
ax.set(xlabel="UTC day in July 2026", ylabel="Satellite detections")
ax.set_xticklabels([day[-2:] for day in days], rotation=0)
plt.tight_layout()
plt.show()

selected_day = daily_counts.idxmax()
selected = detections.loc[detections["day"] == selected_day]
print(f"{selected_day}: {len(selected):,} detections")