Public API Reference

This page contains the automatically generated public SDK reference for Firecube. Plugin code should import these symbols from firecube.ingestor.api or firecube.core.api — never from deeper internal modules.

The reference is organized in three tiers so plugin authors can find the right surface for the task at hand:

1. Primary Plugin Authoring Surface — everything required to build a working plugin from scratch. Start here.
2. Advanced Plugin Authoring — exceptions, advanced context types, write-strategy customization, and pipeline internals for non-trivial plugins.
3. Utilities & Type System — range/slot helpers, source-file types, protocols, and core infrastructure (firecube.core.api).

Primary Plugin Authoring Surface

Start here. Everything needed to write a working plugin. Import from firecube.ingestor.api unless noted.

Registration & Discovery

firecube.ingestor.api.register_ingestor(name)

Decorator to register a plugin under a given name.

firecube.ingestor.api.discover_ingestors()

Import all plugin modules so that decorators can register them.

Discovery strategy

1) If already loaded, return cached copy. 2) Load entry points from group 'firecube.plugins'. This triggers module imports, ensuring @register_ingestor runs. 3) If FIRECUBE_LOAD_BUILTIN_PLUGINS=1 is set, fallback to walking the firecube.plugins namespace.

Base Class

firecube.ingestor.api.BaseIngestor

Bases: BaseIngestorHookMixin, Ingestor, ABC

Orchestrating base class for all Firecube ingestors.

BaseIngestor is a composition facade. It wires together the runtime services (batching, telemetry, recording, workspace, resume-guard) and exposes a hook surface so that plugin authors only need to implement their domain logic.

run() delegates to: - BatchPlanner — creates PipelineBatch objects from items. - TierConfigurator — splits flat options into EngineConfig / TemplateConfig / PluginConfig options tiers (not storage layers; storage I/O wiring is handled by StorageBinding / StorageSession). - WorkspaceManager — manages per-run temp directories. - ResumeGuard — enforces resume / overwrite safety. - TelemetryService — wraps the telemetry sink with metric limits. - SpanRecorder — writes manifest entries to ChunkManager. - PipelineExecutor — parallel or sequential batch dispatch.

Hooks fall into three categories. The class-variable name is also required.

MUST override (abstract): _process_batch(batch, ctx) -> PipelineResult Core per-batch logic. Called in worker threads for parallel mode. Templates (GenericZarrIngestor, GenericParquetIngestor) implement this; plain plugins may override it directly.

SHOULD override (return empty defaults; rarely correct without change): discover_source_files(ctx) -> Iterable Discovers source files to be batched (paths/URIs/objects) using configured include patterns.

CAN override (optional lifecycle / metadata hooks): filter_item(item, ctx) -> bool Per-item filter applied before batching. Default: keep all. item_size_bytes(item) -> int | None Used for batch size estimation. Default: None. get_batch_groups(items, ctx) -> list[str] Derives logical output groups for a batch. Default: ['default']. slice_meta_keys() -> list[str] Option keys that identify a logical "slice" (used for resume conflict detection). Default: empty list. validation_group(ctx) -> str | None Zarr group path used by the resume-guard's validate_zarr check. on_pipeline_start(ctx, state) Called once before any batch runs. Useful for shared resource setup (e.g. persistent DuckDB schema). on_batch_success(ctx, state, batch, result) Called on the main thread after each successful batch. on_batch_failure(ctx, state, batch, result) Called on the main thread after each failed batch. DO NOT override (framework internals): run(ctx) — top-level orchestration entry point. _create_batches — delegates to BatchPlanner. finalize_pipeline — delegates to PipelineExecutor.

Methods:

__init_subclass__(**kwargs)

Build-time validation for ingestor class contracts.

batch_setup(ctx)

Hook for per-batch setup (e.g. DB connections). Cooperatively calls super.

batch_teardown(ctx)

Hook for per-batch cleanup. Cooperatively calls super.

catalog_group_info(group, store_uri, storage_config=None)

Optionally annotate or hide one discovered catalog group.

cleanup_batch_data(batch, ctx)

Optional hook to clean up batch data (e.g. drop table rows).

default_aggregate_metrics(ctx, state) staticmethod

Default aggregate helper for plugins that do not need custom policy.

describe_options() classmethod

Return structured options documentation for CLI introspection.

discover_source_files(ctx)

Discover source files matching include patterns.

filter_item(item, ctx)

Filter items before batching (Hook). Default: True.

finalize_pipeline(ctx, state)

Delegate finalization to PipelineExecutor.

get_batch_groups(items, ctx)

Return logical write groups for a batch of items (Hook). Default: ['default'].

ingest(ctx)

Compatibility entry point matching the public Ingestor protocol.

item_size_bytes(item)

Estimate size of an item in bytes.

prepare_batch_data(batch, ctx)

Optional hook to prepare data (e.g. load files to DB) before group iteration.

resolve_output_uri(ctx, write_mode='staged')

Resolve the canonical output URI (Dataset Directory) for this run.

run(ctx)

Orchestrate the ingestion.

The caller-provided IngestContext is treated as immutable input. Runtime-only fields are carried in an internal RuntimeIngestContext clone for the duration of this run.

ctx.storage.output must already be populated by the caller (CLI/SDK boundary) before invoking run(); this method does not auto-construct sessions.

Context & Configuration

firecube.ingestor.api.PluginContext

Read-only proxy for IngestContext for all plugin-facing hooks.

Attributes

run_id property

Stable run identifier assigned by the engine for this execution.

storage property

Storage sessions bound to this plugin run, keyed by role.

firecube.ingestor.api.EngineConfig dataclass

Configuration for the Firecube Engine (Runtime & Workspace).

These options control HOW the ingestion runs, not WHAT product logic is applied.

Attributes:

Name	Type	Description
pipeline_parallel	bool	Enable parallel batch preprocessing.
pipeline_workers	int	Number of pipeline worker threads.
pipeline_batch_size	int	Number of source items per batch.
cleanup_workspace	bool	Delete temporary workspace files after the run.
workspace	str | None	Optional workspace directory override.
include_patterns	list[str] | None	Optional file patterns for source discovery.
write_mode	str	Write strategy, either "staged" or "direct".
resume_existing	bool	Continue a compatible incomplete or overlapping run.
force_reingest	bool	Re-process existing spans intentionally.
incremental	bool	Reserved incremental-mode switch.
dry_run	bool	Build and validate the run without committing writes.
duckdb_persist_batches	bool	Persist intermediate DuckDB batch data.
upload_workers	int	Number of staged upload workers.
no_progress	bool	Disable progress logging.
validate_zarr	bool	Validate Zarr state as part of resume checks.
validate_zarr_group	str	Zarr group to validate when validation is enabled.
validate_zarr_timeout_s	int | None	Optional Zarr validation timeout in seconds.
validate_zarr_max_chunks	int | None	Optional validation chunk-scan limit.
validate_zarr_on_timeout	str	Timeout behavior, usually "warn".
skip_preflight	bool	Skip storage preflight checks.
slot_start	int | None	First slot index for orchestrated parallel ingestion.
slot_end	int | None	One-past-last slot index for orchestrated parallel ingestion.
slot_size	int | None	Slot width used when deriving slot ranges from the environment.
slot_group	str | None	Zarr group owned by this worker in multi-group slot runs.

Methods:

from_options(options) classmethod

Create config from options, strictly rejecting unknown keys.

firecube.ingestor.api.PluginConfig dataclass

Base configuration for all Firecube ingestors.

Subclasses should define fields as dataclass fields. Use from_options to parse and validate a raw dictionary.

Attributes:

Name	Type	Description
_allow_unknown	bool	Class-level escape hatch for plugins that intentionally accept unknown option keys. Keep this False for strict validation.

Methods:

from_options(options) classmethod

Create a config instance from a dictionary, with strict validation.

Parameters:

Name	Type	Description	Default
options	dict[str, Any]	Dictionary of options (e.g. from CLI or config file).	required

Returns:

Type	Description
T	An instance of the config class.

Raises:

Type	Description
ValueError	If unknown keys are present (and _allow_unknown is False) or if type conversion fails.

Results & Metrics

firecube.ingestor.api.PipelineResult dataclass

Result from a pipeline batch processing.

Attributes

output_path property

Compatibility view of the primary output path.

firecube.ingestor.api.OutputPaths dataclass

Shared outputs shape used by both batch- and run-level results.

Methods:

get(key, default=None)

Return an output path by compatibility key.

items()

Return compatibility key/value pairs.

keys()

Return the populated output keys in insertion order.

firecube.ingestor.api.ResultMetrics dataclass

Shared metrics shape used by both batch- and run-level results.

Methods:

__getitem__(key)

Provide mapping-style access for compatibility code paths.

__iter__()

Iterate over compatibility metric keys.

__len__()

Return the number of populated compatibility keys.

__post_init__()

Seed compatibility mapping state from typed fields when needed.

__setitem__(key, value)

Update a compatibility metric entry and keep typed fields in sync.

get(key, default=None)

Return a compatibility-mapped metric value by key.

items()

Return compatibility key/value pairs.

keys()

Return the current compatibility keys in insertion order.

setdefault(key, default=None)

Set a compatibility metric only when missing.

to_dict()

Render typed fields as a public dict (no private _compat leak).

update(other=None, /, **kwargs)

Update compatibility metrics from another mapping or keyword args.

values()

Return compatibility values.

firecube.ingestor.api.merge_batch_metrics(ctx, state)

Merge successful batch metrics into one run-level mapping.

Default policy: - Numeric values are summed. - List values are concatenated. - Other values keep the first successful value. - Zarr coverage is always merged under metrics["zarr"]["coverage"].

Generic Template Classes

firecube.ingestor.api.GenericZarrIngestor

Bases: BaseIngestor

Thin facade over :class:AppendStrategy for Zarr-based batch ingestion.

Resolves URIs/storage, acquires write claims, then delegates all append logic to AppendStrategy.write_groups().

Subclasses implement build_dataset(group, items, ctx) -> xr.Dataset | None.

Methods:

build_dataset(group, items, ctx) abstractmethod

Convert a sub-batch of items into an Xarray Dataset for the given group.

items is the time-grouped slice for this batch iteration. Returns None to skip writing for this group/batch.

get_zarr_config(ctx)

Return Zarr storage options from validated template config.

on_pipeline_start(ctx, state)

Initialize pipeline resources (including persistent DuckDB).

firecube.ingestor.api.GenericParquetIngestor

Bases: BaseIngestor

Generic Pipelined Ingestor for Parquet outputs.

Methods:

build_dataset(group, batch, ctx) abstractmethod

Convert a batch of items into an in-memory tabular dataset for the given group.

Supported return types for the default Parquet writer

• pyarrow.Table
• pandas.DataFrame (if pandas is installed)

Returns None to skip writing for this group/batch.

get_batch_groups(items, ctx)

Return list of groups to process. Default: ['default'].

output_relpath(group, batch, ctx)

Return a relative output path (within the dataset root) for a group/batch.

write_parquet(dataset, *, output_path, storage_options, ctx)

Write a supported dataset to Parquet and return number of rows written.

firecube.ingestor.api.GenericTensogramIngestor

Bases: BaseIngestor

Template for plugins that write directly to Tensogram .tgm format.

Methods:

build_dataset(group, items, ctx) abstractmethod

Convert a sub-batch of items into an xarray Dataset.

get_batch_groups(items, ctx)

Return list of Tensogram groups/messages to process.

firecube.ingestor.api.DirectZarrIngestor

Bases: BaseIngestor

Abstract template for direct-Zarr region-based ingestors.

Plugins that write directly to Zarr (bypassing xarray) should subclass this template and implement:

• :meth:zarr_schema — declare groups and arrays.
• :meth:build_write_intents — convert a batch into write operations.

The template orchestrates store setup, write execution via a region write strategy, coverage tracking, and metrics aggregation.

Attributes

SUPPORTS_SLOT_RANGE_PARALLELISM = False class-attribute

Explicit opt-in for within-group parallel ingestion.

Set to True and override timestamp_to_ts_index() and global_expected_time_count() to enable slot-range parallelism.

Methods:

__init_subclass__(**kwargs)

Declaration-time validation for parallel ingestion opt-in.

build_write_intents(batch, ctx) abstractmethod

Convert a batch into a list of write operations.

Each WriteIntent describes a single region write, 1-D write, or timestamp write. The template will execute them in order via the configured region write strategy.

Return an empty list to skip the batch.

filter_items_to_slot_range(items, slot_start, slot_end, ctx)

Filter source items to those whose ts_index falls in [slot_start, slot_end).

Strongly recommended for parallel ingestion. The default passthrough returns all items unchanged; if build_write_intents() then emits any WriteIntent whose ts_index falls outside [slot_start, slot_end), the post-intent assertion will raise WriteIntentRangeError and FAIL the batch (it does NOT silently discard out-of-range intents).

get_batch_groups(items, ctx)

Return groups declared in zarr_schema(ctx) — DirectZarr writes via WriteIntent.group.

global_expected_time_count(ctx)

Return max ts_index + 1 per group across the planned run.

slot_index_model(ctx)

Return the slot-index model governing this product's time axes.

Plugins that set SUPPORTS_SLOT_RANGE_PARALLELISM = True MUST override this method. The returned model is persisted to the control plane and mirrored as Zarr root attributes; it becomes the canonical identity for all concurrent writers targeting this product.

Default implementation raises :class:NotImplementedError so that non-parallel plugins calling this accidentally receive a clear error rather than silent incorrect behaviour.

timestamp_to_ts_index(group, timestamp_val)

Globally deterministic mapping from the conceptual time axis to ts_index.

"timestamp" here is the stable contract token, not the on-disk dim name.

zarr_schema(ctx) abstractmethod

Declare the Zarr store layout for this ingestor.

Returns a list of group specifications describing every group and array that the ingestor may write to. Called once per batch to ensure groups and arrays exist before writes begin.

Template Configurations

firecube.ingestor.api.ZarrTemplateConfig dataclass

Bases: TemplateConfig

Configuration for GenericZarrIngestor.

Attributes:

Name	Type	Description
zarr_chunk_shape	dict[str, int] | None	Optional per-dimension inner chunk sizes.
zarr_sharding	bool	Enable Zarr v3 sharding.
zarr_shard_shape	dict[str, int] | None	Optional per-dimension shard sizes.
zarr_compression	bool | str	Compression setting. False disables compression, True selects the default codec, and a string selects a codec name.
zarr_consolidate	bool	Consolidate Zarr metadata after writes.
zarr_time_encoding	str | None	Optional time encoding override.
zarr_async_concurrency	int	Async write concurrency used by Zarr.
dask_scheduler	str | None	Optional Dask scheduler override.
dask_write_threads	int	Optional write-thread count for Dask-backed writes.

firecube.ingestor.api.ParquetTemplateConfig dataclass

Bases: TemplateConfig

Configuration for GenericParquetIngestor.

Attributes:

Name	Type	Description
parquet_partition_by	list[str] | None	Optional Hive-style partition columns.
parquet_row_group_size	int | None	Optional rows per Parquet row group.

firecube.ingestor.api.TensogramTemplateConfig dataclass

Bases: TemplateConfig

Configuration for GenericTensogramIngestor.

Attributes:

Name	Type	Description
tensogram_compression	str	Compression codec for Tensogram output.
tensogram_message_granularity	str	Message grouping strategy.
tensogram_allow_nan	bool	Permit NaN values in archive output.
tensogram_allow_inf	bool	Permit infinite values in archive output.

DirectZarrIngestor Types

firecube.ingestor.api.WriteIntent dataclass

A single write operation to execute against the Zarr store.

Produced by DirectZarrIngestor.build_write_intents() to describe what data should be written, where, and at which timestamp index.

The kind field selects the write method on RegionZarrWriter. Here, "timestamp" means the conceptual time/index axis, not the on-disk dimension name; the latter is configured separately via IndexedRegionStrategy.time_coord_name.

• "region" → write_region(group, array, ts_index, y_slice, data, channel_index=...)
• "1d" → write_1d(group, array, ts_index, data)
• "timestamp" → write_timestamp(group, ts_index, timestamp_val)
• "static" → write_static(group, array_name, data) (non-time-indexed; ts_index is ignored)

Attributes

timestamp_val = None class-attribute instance-attribute

Conceptual time/index axis value for kind="timestamp" writes.

"timestamp" is a stable plugin-contract token, not the on-disk dim name; the actual dim/coord name comes from IndexedRegionStrategy.time_coord_name.

firecube.ingestor.api.ZarrArraySpec dataclass

Specification for a single Zarr array within a group.

firecube.ingestor.api.ZarrGroupSpec dataclass

Specification for a Zarr group and its arrays.

Returned by DirectZarrIngestor.zarr_schema() to describe the full layout.

attrs is optional, convention-agnostic group-level metadata stamped onto the group's zarr.json at schema setup — e.g. dataset-level attributes a plugin chooses to publish. Firecube writes the mapping verbatim and does not interpret it (no convention is assumed); reserved firecube-internal attribute names are rejected at write time.

Advanced Plugin Authoring

Errors to catch, advanced context types, write-strategy customization, and pipeline internals for complex plugins.

Exceptions

firecube.ingestor.api.IngestorError

Bases: FirecubeError

Base exception for all ingestor-layer errors.

firecube.ingestor.api.ConfigurationError

Bases: IngestorError

Configuration validation failed.

firecube.ingestor.api.StorageError

Bases: FirecubeError

Storage backend operation failed (local, S3, etc.).

firecube.ingestor.api.ManifestError

Bases: FirecubeError

Manifest read/write/update operation failed.

firecube.ingestor.api.RangeOverlapError

Bases: ResumeConflictError

Raised when a new slot-range invocation overlaps with an active non-terminal run.

Overlapping ranges risk Zarr chunk-boundary corruption. Abandon the conflicting run first: firecube chunks runs abandon ...

firecube.ingestor.api.ResumeConflictError

Bases: IngestorError

Resume/overwrite conflict detected.

firecube.ingestor.api.SchemaDriftError

Bases: FirecubeError

Existing Zarr array metadata drifted from the declared schema.

Compared fields: dtype, rank, shape[1:], chunks, fill_value. The time axis shape[0] is handled specially: smaller fails, larger warns.

firecube.ingestor.api.SchemaSizeMismatchError

Bases: IngestorError

Raised when an existing Zarr array's shape is smaller than the global expected size.

Existing arrays mismatch the plan. Either delete them or update the plan to match.

firecube.ingestor.api.WriteIntentRangeError

Bases: IngestorError

Raised when a WriteIntent's ts_index falls outside the assigned slot range.

This is a correctness violation — the plugin filter is advisory; this error is the mandatory backstop. NEVER silently drop out-of-range intents.

Advanced Context Types

firecube.ingestor.api.RuntimeIngestContext dataclass

Bases: IngestContext

Engine-owned runtime context copied from IngestContext.

This carries internal execution-only state and must never be reused across runs.

Attributes

dry_run property writable

Whether side-effecting writes should be suppressed when supported.

force_reingest property writable

Whether this run is allowed to overwrite existing slice data.

incremental property writable

Whether plugin logic should prefer incremental update behavior.

temp_root property writable

Per-run workspace root used for materialization and temporary files.

Methods:

from_ingest_context(ctx, *, run_id, temp_root, materializer) classmethod

Build an isolated runtime copy without mutating the caller context.

materialize(source)

Ensure the source file is available locally and return its path.

If the file is remote (e.g. S3), it will be downloaded to the per-run cache. If it's already local, the path is returned directly.

firecube.ingestor.api.StorageContext dataclass

Named storage role bindings for an ingestion run.

firecube.ingestor.api.IngestContext dataclass

Caller-provided input context for one ingestion run.

Methods:

option(key, default=None)

Convenience accessor for optional plugin parameters.

Pipeline Internals

firecube.ingestor.api.IngestManifest dataclass

The public contract for ingestion results.

Methods:

to_dict()

Convert to dictionary for JSON serialization, excluding None values.

firecube.ingestor.api.PipelineRunState dataclass

Immutable run snapshot for pipeline orchestration and hooks.

firecube.ingestor.api.PipelineBatch dataclass

A batch of data ready for processing in the pipeline.

firecube.ingestor.api.IngestResult dataclass

Structured dataset returned by plugins.

Attributes

output_path property

Compatibility view of the primary output path.

Methods:

all_outputs()

Return the typed outputs container for compatibility callers.

Metrics Types

firecube.ingestor.api.PipelineMetrics dataclass

Typed pipeline metrics from plugin results.

Methods:

to_dict()

Render pipeline metrics as a compatibility dictionary.

firecube.ingestor.api.StorageMetrics dataclass

Typed storage summary from plugin results.

Methods:

to_dict()

Render storage metrics as a compatibility dictionary.

Write Strategies

firecube.ingestor.api.AppendWriteStrategy

Bases: Protocol

Protocol implemented by xarray-append Zarr write strategies.

firecube.ingestor.api.RegionWriteStrategy

Bases: Protocol

Protocol implemented by direct region-based Zarr write strategies.

When claim_for_slot is provided, claim_for_group protects schema setup only and claim_for_slot protects per-ts_index intent dispatch. If claim_for_slot is None, implementations may fall back to claim_for_group for dispatch coordination.

firecube.ingestor.api.AppendStrategy

Xarray-append write strategy for Zarr stores.

Wraps :func:~firecube.ingestor.runtime.zarr.append.append_time_groups behind the AppendWriteStrategy Protocol so that GenericZarrIngestor can treat append and region writes uniformly.

All static configuration is captured at construction time; only per-batch dynamic inputs are passed to :meth:write_groups.

Methods:

write_groups(*, group_to_timestamps, dataset_for_batch, batch_size, claim_for_group=None)

Delegate to append_time_groups() and return its metrics dict.

firecube.ingestor.api.IndexedRegionStrategy

Write strategy that dispatches WriteIntent objects to RegionZarrWriter.

Implements the RegionWriteStrategy Protocol.

Constructed with a store URI and schema; write_groups() receives the per-batch intents and returns coverage metrics.

Methods:

write_groups(*, group_to_intents, schema=None, claim_for_group=None, claim_for_slot=None, slot_range=None, slot_group=None)

Execute write intents grouped by Zarr group and return metrics.

Parameters

group_to_intents: Mapping of group name to list of WriteIntent objects. schema: Group specifications for array creation. Falls back to the constructor-provided schema if None. claim_for_group: Optional callback returning a context manager for group/schema coordination. When claim_for_slot is provided, this protects schema setup only. claim_for_slot: Optional callback returning a context manager for per-ts_index intent dispatch. Falls back to claim_for_group and then a null context when omitted. slot_range: Optional half-open [start, end) slot range. When provided, all intents are validated before any schema setup or writes occur. slot_group: Optional group name that this pod owns. When set together with slot_range, the post-intent assertion only validates intents for slot_group; intents for other groups are skipped with a warning (not silently dropped) and no writes happen for them.

Returns

dict with "coverage" list and "duration_s" float.

firecube.ingestor.api.TensogramWriteStrategy

Write strategy that encodes xarray Datasets to a Tensogram .tgm file.

One TensogramWriteStrategy instance = one .tgm file. Each write_groups() call appends a new message to the file. No append semantics — if the target already exists, it is overwritten on first write.

Methods:

close()

Close the .tgm file handle.

write(ds)

Encode a single xarray Dataset as one Tensogram message.

write_groups(*, group_to_timestamps, dataset_for_batch, batch_size, claim_for_group=None)

Write all groups from the batch to the .tgm file.

Advanced Configuration

firecube.ingestor.api.TemplateConfig dataclass

Base for template configurations.

firecube.ingestor.api.CatalogGroupInfo dataclass

Optional plugin annotations for one discovered catalog group.

firecube.ingestor.api.config_keys(cls)

Return the set of known field names for a dataclass.

Utilities & Type System

Range/slot helpers, source-file types, protocols, and core infrastructure. Most symbols here come from firecube.ingestor.api or firecube.core.api.

Range & Slot Utilities

firecube.ingestor.api.chunk_align_ranges(ranges, chunk_size)

Expand ranges to chunk boundaries, then deduplicate and sort them.

firecube.ingestor.api.compute_covered_ranges(coverage)

Convert inclusive coverage ranges into half-open PlannedRange values.

firecube.ingestor.api.validate_chunk_alignment(slot_start, slot_end, chunk_shapes_per_group, global_expected=None)

Raise ConfigurationError when any group's slot range is misaligned.

Terminal partial chunks are allowed when slot_end matches the group's global expected total.

firecube.ingestor.api.validate_slot_range(slot_start, slot_end)

Validate that [slot_start, slot_end) is well-formed and non-negative.

firecube.ingestor.api.warn_if_misaligned(slot_start, slot_end, chunk_shapes_per_group, logger, global_expected=None)

Log a warning for each group whose slot range is not chunk-aligned.

Terminal partial chunks are allowed when slot_end matches the group's global expected total.

Source File Types

firecube.ingestor.api.SourceFile

Bases: Protocol

Abstraction for a file-like object that may be local or remote.

This allows plugins to accept S3 objects as if they were local files, supporting standard python file operations (seek, read).

Attributes

uri property

The canonical URI of the source file.

Methods:

local_path()

Return a local Path if one exists efficiently.

Returns None for remote files (unless they are explicitly materialized). This allows optimizations for libraries that prefer pathlib.Path only when cheap.

open()

Open the file in binary read mode. Must be seekable.

firecube.ingestor.api.LocalSourceFile

Implementation of SourceFile for local filesystem.

Run Tracking

firecube.ingestor.api.SpanCoverage dataclass

Coverage information for a span record.

Strategy-neutral: supports both xarray-append (time_index_ranges) and direct zarr-python region writes (region_spec). At least one of time_index_ranges or region_spec should be set for meaningful coverage, but both default to None so construction never fails.

Attributes

timestamps_written property

Total number of time indices covered by all ranges.

firecube.ingestor.api.PlannedRange dataclass

A half-open slot range [slot_start, slot_end) for one Zarr group.

Uses HALF-OPEN semantics: slot_start is inclusive, slot_end is exclusive. This matches Python slicing convention and engine array indexing.

firecube.ingestor.api.SlotRange = tuple[int, int] module-attribute

firecube.ingestor.api.WriteDomain dataclass

Stable conflict key for mutually exclusive physical writes.

Attributes

claim_name property

Return the SHA-256 hashed filename for the claim file.

identifier property

Return the canonical product:category:name identifier string.

Protocols

firecube.ingestor.api.Ingestor

Bases: Protocol

Protocol defining the core contract for an ingestor.

firecube.ingestor.api.PipelineHost

Bases: Protocol

Protocol for classes that can be executed by the pipeline runner.

The runtime pipeline executor expects these attributes and lifecycle hooks to exist on the host object (typically an ingestor instance).

firecube.ingestor.api.DatasetProducer

Bases: Protocol

Protocol for batch templates that produce in-memory datasets.

Tensogram output is selected by this structural capability rather than by requiring a specific Zarr base class. Implementations share the build_dataset and get_batch_groups hooks provided by BaseIngestor.

Methods:

build_dataset(group, items, ctx)

Convert a sub-batch into a dataset for group.

get_batch_groups(items, ctx)

Return logical dataset groups/messages for items.

firecube.ingestor.api.is_dataset_producer(cls)

Return True iff cls genuinely satisfies the DatasetProducer contract.

@runtime_checkable Protocol's issubclass / isinstance checks validate method-name presence only — Python deliberately does NOT check signatures. That is structurally unsound for DatasetProducer because GenericParquetIngestor declares build_dataset(self, group, batch: PipelineBatch, ctx) which collides with the protocol's build_dataset(self, group, items: list, ctx) by name but is invoked incompatibly by TensogramWriteStrategy (which passes a list of timestamps, not a PipelineBatch).

This helper adds a signature-shape check: the second positional parameter after group must NOT be named batch (the parquet convention). Plugins using *args are accepted.

firecube.ingestor.api.SlotRangeCapable

Bases: Protocol

Protocol for hosts that support slot-range parallel ingestion.

The engine dispatches to the slot-range scheduling path when the host structurally satisfies this protocol and SUPPORTS_SLOT_RANGE_PARALLELISM is True. Plugins opt in by setting the class variable and implementing the three methods below; they do not need to subclass any particular template.

Methods:

filter_items_to_slot_range(items, slot_start, slot_end, ctx)

Filter source items to those whose ts_index falls in [slot_start, slot_end).

global_expected_time_count(ctx)

Return max ts_index + 1 per group across the planned run.

timestamp_to_ts_index(group, timestamp_val)

Globally deterministic mapping from the conceptual time axis to ts_index.

Core Infrastructure (firecube.core.api)

firecube.core.api.parse_uri(target)

Parse a URI/path into a minimal fsspec-like structure.

Returns a dict with

• protocol: "file", "s3", ...
• path: protocol-specific path (e.g. "bucket/prefix/key" for s3, "/abs/path" for file)

firecube.core.api.create_filesystem_for_uri(uri, storage_config, *, format)

Create a driver-aware filesystem for a concrete product/output URI.

firecube.core.api.discover_input_files(source, *, storage_config=None, include_suffixes=('.zip', '.h5', '.nc'), preferred_globs=None, recursive=True, sniff_hdf5=True, exclude=None)

Discover input files from a local path or remote URI.

Selection is intentionally conservative and format-agnostic

• Accept files matching include_suffixes
• Optionally accept extensionless files that look like HDF5
• Optionally add files matched by preferred_globs (glob patterns)

Returns URI/path strings (for example /tmp/data/file.nc or s3://bucket/prefix/file.nc). Path inputs are accepted for backward compatibility and are converted to strings internally.

firecube.core.api.StorageConfig dataclass

Global storage configuration for the service.

Location-specific URI fields are intentionally excluded here; runtime storage code now consumes the trimmed config plus a separate target URI boundary where needed.

Attributes:

Name	Type	Description
storage_type	str	Storage locality/class. Use "local" or "s3".
endpoint_url	str | None	Optional S3-compatible endpoint URL.
access_key	str | None	Optional access key for explicit S3 credentials.
secret_key	str | None	Optional secret key for explicit S3 credentials.
region	str | None	Optional S3 region name.
path_style	bool	Whether to use S3 path-style addressing.
storage_driver	str	Storage I/O driver. Use "fsspec" or "obstore".

Methods:

validate()

Validate that required fields are present for the storage type.

firecube.core.api.RunInfo dataclass

Projected information for one ingestion run.

Attributes

is_terminal property

True if the run status is complete, failed, or abandoned.

slot_range = None class-attribute instance-attribute

Half-open slot range [start, end) for parallel ingestion pods.

None for single-pod (Phase 2 and pre-Phase-3) runs.

stale property

True if a non-terminal run has not been updated within the stale threshold.

firecube.core.api.describe_control_plane(*, product=None, base_uri=None, product_uri=None)

Return canonical .firecube/ control-plane locations for one product.

firecube.core.api.RegionZarrWriterProtocol

Bases: Protocol

Structural interface for direct-Zarr region writers.

Plugins and templates may depend on this protocol without coupling to the concrete RegionZarrWriter implementation.

firecube.core.api.delete_path(base_uri, subpath, *, storage_config=None, allow_manifest_paths=False, dry_run=False, filesystem=None)

Delete a path (file or directory) from storage.

Parameters:

Name	Type	Description	Default
base_uri	str	Base URI (e.g., 's3://bucket/store.zarr' or '/data/store.zarr')	required
subpath	str	Relative path within base_uri to delete (e.g., 'F072')	required
storage_config	Any | None	Optional StorageConfig for S3 credentials	None
allow_manifest_paths	bool	If True, allow deleting manifest paths	False
dry_run	bool	If True, only check existence without deleting	False

Returns:

Type	Description
dict[str, Any]	Dict with keys: path, exists, deleted

firecube.core.api.ensure_directory(path)

Ensure a directory exists, creating parents if needed.

firecube.core.api.ensure_product_uri(base_uri, product)

Append product to base_uri if the last path segment doesn't exactly match.

Uses URI path parsing to avoid substring false positives (e.g. 'daily' matching 'daily_archive').

firecube.core.api.resolve_dataset_target(target, *, write_mode='staged', temp_root=None, direct_base_uri=None)

Resolve the URI for a dataset target (Dataset Directory).

This determines the "Package Root" for Zarr/Parquet output, considering: 1. Absolute/Full URIs (s3://, /abs/path) -> Used as-is. 2. Relative paths + Staged mode -> Anchored to temporary workspace. 3. Relative paths + Direct mode -> Anchored to the typed product URI's parent.

firecube.core.api.infer_target_protocol(target)

Infer the fsspec protocol for a target string.

Returns "file" for local paths and file:// URIs.

firecube.core.api.is_remote_target(target)

Return True when target is a non-local URI (e.g. s3://...).

firecube.core.api.local_path_from_target(target)

Resolve a target string into an absolute local path.

Supports relative paths and file:// URIs.

firecube.core.api.path_stats(uri, *, storage_config=None, exclude_substrings=None)

Get bytes and file count for a path (local or remote).

Works with both local paths and S3 URIs.

Parameters:

Name	Type	Description	Default
uri	str	Local path or remote URI	required
storage_config	Any | None	Optional StorageConfig for S3 credentials	None
exclude_substrings	Iterable[str] | None	Path substrings to exclude (default: manifest paths)	None

Returns:

Type	Description
dict[str, int]	Dict with 'bytes' and 'files' keys

firecube.core.api.clean_netcdf_encoding(ds)

Strip HDF5 chunk hints from all variable encodings.

Removes 'chunks', 'chunksizes', and 'preferred_chunks' from each variable's encoding dict so they do not conflict with the Zarr chunk layout chosen by the core writer. Modifies encoding dicts in place; returns the same dataset object.

firecube.core.api.prepare_netcdf_for_zarr(ds, time_dim='time', target_time_dim='timestamp')

Convenience wrapper: rename time dimension then clean encodings.

Applies all standard NetCDF→Zarr V3 preparation steps in order: 1. Rename time dimension from time_dim to target_time_dim. 2. Strip HDF5 chunk hints from all variable encodings.

firecube.core.api.read_hdf5_array(hdf5_path, *, variable, logger=None)

Read a named array from a local HDF5(-like) file, with xarray fallback.

Parameters:

Name	Type	Description	Default
hdf5_path	Path	Path to the HDF5 file.	required
variable	str	Name of the variable/dataset to read.	required
logger	Logger | None	Optional logger for debug messages.	None

Returns:

Type	Description
Any	Numpy array of the data (float32).

Raises:

Type	Description
KeyError	If the variable is not found.
RuntimeError	If dependencies are missing or read fails.

firecube.core.api.materialize_hdf5_path(file_path, *, extract_root=None, logger=None)

Return a local HDF5 path for a source path (ZIP or direct HDF5-like file).

If the input is a ZIP file, it extracts the content to a temporary directory. If it is already an HDF5-like file, it is returned as-is.

Parameters:

Name	Type	Description	Default
file_path	Path	Source file path.	required
extract_root	Path | None	Optional root directory for temporary extraction.	None
logger	Logger | None	Optional logger.	None

Returns:

Type	Description
tuple[Path, TemporaryDirectory | None]	Tuple of (path_to_hdf5_file, temporary_directory_object_or_None).

firecube.core.api.rename_time_dim(ds, source='time', target='timestamp')

Rename the time dimension to Firecube's append convention.

If source exists as a dimension or coordinate, rename it to target. If source is not present, return the dataset unchanged (no error).

firecube.core.api.require_tensogram(operation)

Raise ImportError with install instructions if tensogram is not available.

Next Steps

• Plugin Contract — check required plugin rules
• Plugin Storage Access — use PluginContext storage safely
• Plugins — choose a base class and plugin path