Skip to content

Streaming CoverageJSON encoder over the polytope-mars block stream - #140

Draft
jameshawkes wants to merge 12 commits into
mainfrom
feat/streaming-encoder
Draft

jameshawkes wants to merge 12 commits into
mainfrom
feat/streaming-encoder

Conversation

@jameshawkes

@jameshawkes jameshawkes commented Oct 7, 2026 •

Copy link
Copy Markdown
Contributor

Part of the feature-extraction streaming work (companion PRs in polytope-mars and polytope). Adds a CoverageJSON encoder that consumes polytope-mars' block stream and hands the document back in bounded pieces, instead of walking the polytope request tree and building the whole document in memory. The legacy API (Covjsonkit().encode(...), from_polytope*, decoders, param_db) is unchanged; all existing tests pass.

covjsonkit.stream.CovjsonStreamEncoder

  • begin(header) -> bytes, encode_iter(block) -> Iterator[bytes], encode(block) -> bytes, end() -> bytes; content_type = "application/prs.coverage+json", file_extension = "covjson".
  • Fragments. encode_iter(block) yields one slice of the block's arrays at a time, each fragment at most max_fragment_bytes (constructor argument or encoders.covjson.max_fragment_bytes, default 8 MiB), so the text of a whole block never exists at once: a whole-world O2560 coverage is ~800 MB of coordinates and ~470 MB of values, and encoding a 5M-point block costs ~40 MB of peak RSS whatever its size. The fragments of a block must be consumed completely and in order before the next block is encoded. encode(block) returns b"".join(encode_iter(block)) for buffered callers, and both give the same bytes. The structural fragments (a coverage's mars:metadata and t, the collection's referencing and parameters) are written whole — they are a few hundred bytes.
  • Blocks are read by attribute only: the module imports neither polytope-feature, polytope-mars nor covjson-pydantic (tested). Parameter metadata arrives resolved in the header, so the encoder does no param_db lookups.
  • MultiPoint coverages are written as the blocks arrive: coverage opening, mars:metadata and t on the first coordinates block, [lat, lon, level] composite tuples, each range opened on its first values block and closed on GroupEnd. Memory per coverage is one fragment, plus the coordinates while a coverage has several levels. PointSeries, VerticalProfile and Trajectory collections are buffered and written by end(): they are small, and the legacy layout is point-major across field groups. The PointSeries layout is isolated in the single function pointseries_coverages, so develop Feat/122 point series spec compliant #137 can be ported there.
  • Numeric arrays are serialised with orjson straight from numpy, per slice. Output bytes are identical to json.dumps(legacy_covjson).encode() — separators, ASCII escapes, key order and float digits — for every case of polytope-mars' golden corpus. orjson and CPython differ only in how numbers with 0 < |x| < 1e-4 are spelled, so arrays holding such values are formatted per value with repr; verified on 5,000 random magnitudes from 1e-12 to 1e20.
  • legacy_referencing reproduces the referencing and composite coordinates each legacy encoder method wrote (latitude/longitude/levelist, x/y/z, or t/x/y/z by feature type and time-axis role).

Output differences against the legacy encoders (deliberate)

  • NaN and inf are written as null; json.dumps wrote the invalid JSON token NaN.
  • A param gribjump has no message for is omitted from that coverage's ranges, and a field group with no data at all produces no coverage. The collection's parameters still lists every parameter of the header.

Everything else, including the known quirks (space-separated datetimes on the _step path, the per-encoder referencing coordinates, int realization, efcl coverages without Forecast date), is reproduced byte for byte; the polytope-mars golden corpus checks every feature type against the output of the encoders this replaces.

Tests

228 passed, 5 skipped (195 existing). tests/test_stream_encoder.py covers the layout per domain type on synthetic blocks, band-size invariance, fragment joins equal to the single-fragment and the legacy bytes at 64 B / 1 KiB / 8 MiB limits, the fragment limit itself, null for NaN and omitted ranges, float and composite formatting against json.dumps, the referencing quirks, and that no polytope module is imported. tests/test_stream_memory.py runs a subprocess in which a 5M-point block pair produces 318 MB of CoverageJSON in 54 fragments for ~41 MB of peak RSS above the blocks, asserted below 100 MB and against any growth with the block size.

CovjsonStreamEncoder writes the collection opening in begin(), MultiPoint
coverages as coordinate and value blocks arrive, and buffers the small
PointSeries, VerticalProfile and Trajectory collections until end(). The
bytes equal json.dumps of the legacy encoders' output, with NaN written as
null. Blocks are read by attribute only: no polytope-feature, polytope-mars
or covjson-pydantic imports.
Checks legacy-identical bytes for MultiPoint, identical output for any band
split, null for NaN, omitted ranges for absent params, float formatting
against json.dumps, the referencing quirks, the point-feature layouts and
that the module imports no polytope packages.
Composite tuples and value lists are formatted by orjson over the whole
array and adjusted to json.dumps separators with bytes.replace; arrays
holding values below 1e-4 (printed differently by orjson) keep the per-value
path. Halves peak memory of a 634k-point coverage.
A whole-world coverage is one block of up to 26.3M points: its coordinates are
~800 MB of JSON text and its values ~470 MB, which the encoder used to return
as a single bytes object (and the fe-worker keeps up to four of those in
flight).  encode_iter(block) now yields the block a slice at a time, each
fragment at most max_fragment_bytes (constructor argument or config key,
default 8 MiB): the (n, 3) composite array and the per-value float formatting
are done per slice, so the text of a whole block never exists at once.

encode(block) stays as b"".join(encode_iter(block)) for buffered callers, and
the bytes are unchanged - separators, float spelling and the 1e-5-style
per-value fallback are decided slice by slice but produce the same output as
one pass over the array (tested at 64 B, 1 KiB and 8 MiB fragment limits
against the single-fragment and legacy json.dumps bytes).

Also reads numpy scalars with .item() instead of float() in the buffered
point-feature layouts.
The measurement lives in tests/stream_memory_probe.py (run it directly for the
numbers) and runs in a subprocess so the reported peak is only this encoding;
it resets the process' peak RSS once the arrays are allocated, so a parent's
footprint cannot mask the growth.

5M points -> 318 MB of CoverageJSON in 54 fragments of at most 8 MiB, for
~41 MB of peak RSS above the blocks themselves; 1M points cost ~30 MB, 20M
points ~49 MB.  The test asserts <100 MB and that growth does not follow the
block size.
…tion

CHANGES.md names the block stream and the missing-param rule the encoder
implements instead of citing an external design document, and opens with the
compatibility statement a reader of the branch needs.
… hold

A parameter gribjump has no message for in any field (mx2t6 at hourly steps,
a parameter a stream does not produce) has no range in any coverage. The
collection's parameters object is written after the coverages, so it can
list what the result holds instead of what was requested, and a reader may
take it as the list of present parameters. An empty collection has {}.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants