Skip to content

Streaming Core/Halo Pipeline

RFC 0008 streaming architecture: full-chip processing where layout growth increases tile count, not per-run memory. See architecture and rfcs/0008 for the design; this page is the API reference.

openlithohub.streaming.geometry

Streaming full-chip geometry primitives (RFC 0008).

Core/halo ownership model:

  • The core is the only region whose result may be committed to the final full-chip output.
  • The halo is read-region context that gives the forward model real neighbourhood content instead of zero-padded artefacts.
  • Adjacent cores must exactly cover the global domain (no gaps, no duplicate ownership); adjacent read regions may overlap freely.

BoundingBox dataclass

Half-open pixel rectangle [x0, x1) x [y0, y1) in global coords.

Source code in src/openlithohub/streaming/geometry.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@dataclass(frozen=True)
class BoundingBox:
    """Half-open pixel rectangle ``[x0, x1) x [y0, y1)`` in global coords."""

    x0: int
    y0: int
    x1: int
    y1: int

    def __post_init__(self) -> None:
        if self.x1 <= self.x0 or self.y1 <= self.y0:
            raise ValueError(f"degenerate bbox: x=[{self.x0},{self.x1}) y=[{self.y0},{self.y1})")

    @property
    def width(self) -> int:
        return self.x1 - self.x0

    @property
    def height(self) -> int:
        return self.y1 - self.y0

    @property
    def area(self) -> int:
        return self.width * self.height

    def clipped_to(self, width: int, height: int) -> BoundingBox:
        """Clip to the global domain ``[0,width) x [0,height)``.

        Returns a bbox that may shrink; callers must treat an empty
        intersection (degenerate result) as "no overlap".
        """
        return BoundingBox(
            x0=max(0, self.x0),
            y0=max(0, self.y0),
            x1=min(width, self.x1),
            y1=min(height, self.y1),
        )

    def contains(self, other: BoundingBox) -> bool:
        return (
            self.x0 <= other.x0
            and self.y0 <= other.y0
            and self.x1 >= other.x1
            and self.y1 >= other.y1
        )

clipped_to(width, height)

Clip to the global domain [0,width) x [0,height).

Returns a bbox that may shrink; callers must treat an empty intersection (degenerate result) as "no overlap".

Source code in src/openlithohub/streaming/geometry.py 
44
45
46
47
48
49
50
51
52
53
54
55
def clipped_to(self, width: int, height: int) -> BoundingBox:
    """Clip to the global domain ``[0,width) x [0,height)``.

    Returns a bbox that may shrink; callers must treat an empty
    intersection (degenerate result) as "no overlap".
    """
    return BoundingBox(
        x0=max(0, self.x0),
        y0=max(0, self.y0),
        x1=min(width, self.x1),
        y1=min(height, self.y1),
    )

HaloSpec dataclass

Per-side halo widths requested for a tile read region.

Source code in src/openlithohub/streaming/geometry.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True)
class HaloSpec:
    """Per-side halo widths requested for a tile read region."""

    left: int
    top: int
    right: int
    bottom: int

    @classmethod
    def uniform(cls, px: int) -> HaloSpec:
        if px < 0:
            raise ValueError(f"halo must be nonnegative, got {px}")
        return cls(left=px, top=px, right=px, bottom=px)

    @property
    def total(self) -> int:
        return self.left + self.top + self.right + self.bottom

core_grid_boxes(shape, core_size, *, core_stride=None)

Partition shape into an exact cover of axis-aligned core boxes.

The last column/row of cores is pulled back to the domain edge so cores cover [0,W) x [0,H) with no gap and no overlap. Returns boxes in row-major order.

Source code in src/openlithohub/streaming/geometry.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def core_grid_boxes(
    shape: tuple[int, int],
    core_size: int,
    *,
    core_stride: int | None = None,
) -> list[BoundingBox]:
    """Partition ``shape`` into an exact cover of axis-aligned core boxes.

    The last column/row of cores is pulled back to the domain edge so
    cores cover ``[0,W) x [0,H)`` with no gap and no overlap.  Returns
    boxes in row-major order.
    """
    h, w = shape
    if core_size <= 0:
        raise ValueError(f"core_size must be positive, got {core_size}")
    stride = core_stride if core_stride is not None else core_size
    if stride <= 0:
        raise ValueError(f"core_stride must be positive, got {stride}")

    xs = list(range(0, w, stride))
    ys = list(range(0, h, stride))
    boxes: list[BoundingBox] = []
    for y in ys:
        for x in xs:
            boxes.append(
                BoundingBox(
                    x0=x,
                    y0=y,
                    x1=min(x + core_size, w) if x + core_size > w else x + core_size,
                    y1=min(y + core_size, h) if y + core_size > h else y + core_size,
                )
            )
    return boxes

read_bbox_for(core, halo, width, height)

Expand a core by its halo, clipped to the global domain.

Source code in src/openlithohub/streaming/geometry.py 
125
126
127
128
129
130
131
132
def read_bbox_for(core: BoundingBox, halo: HaloSpec, width: int, height: int) -> BoundingBox:
    """Expand a core by its halo, clipped to the global domain."""
    return BoundingBox(
        x0=max(0, core.x0 - halo.left),
        y0=max(0, core.y0 - halo.top),
        x1=min(width, core.x1 + halo.right),
        y1=min(height, core.y1 + halo.bottom),
    )

halo_actual(core, read)

Recover the per-side halo actually realised after domain clipping.

Source code in src/openlithohub/streaming/geometry.py 
135
136
137
138
139
140
141
142
def halo_actual(core: BoundingBox, read: BoundingBox) -> HaloSpec:
    """Recover the per-side halo actually realised after domain clipping."""
    return HaloSpec(
        left=core.x0 - read.x0,
        top=core.y0 - read.y0,
        right=read.x1 - core.x1,
        bottom=read.y1 - core.y1,
    )

halo_overhead_stats(core_boxes, read_boxes)

Report halo duplicate-compute overhead (prompt §15).

eta_halo = ((C+2h)^2 - C^2) / C^2 generalised per tile with the realised (possibly clipped) halos.

Source code in src/openlithohub/streaming/geometry.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def halo_overhead_stats(
    core_boxes: list[BoundingBox],
    read_boxes: list[BoundingBox],
) -> dict[str, float]:
    """Report halo duplicate-compute overhead (prompt §15).

    ``eta_halo = ((C+2h)^2 - C^2) / C^2`` generalised per tile with the
    realised (possibly clipped) halos.
    """
    if len(core_boxes) != len(read_boxes):
        raise ValueError("core/read box lists must have the same length")
    core_px = sum(b.area for b in core_boxes)
    read_px = sum(b.area for b in read_boxes)
    if core_px == 0:
        return {
            "core_pixels": 0.0,
            "read_pixels": 0.0,
            "halo_pixels": 0.0,
            "halo_overhead_pct": 0.0,
            "n_tiles": float(len(core_boxes)),
        }
    return {
        "core_pixels": float(core_px),
        "read_pixels": float(read_px),
        "halo_pixels": float(read_px - core_px),
        "halo_overhead_pct": 100.0 * (read_px - core_px) / core_px,
        "n_tiles": float(len(core_boxes)),
    }

openlithohub.streaming.core_halo

Tile requests, planning, and the streaming scheduler (RFC 0008).

TileRequest pairs a trusted core with its context halo. Planning a request list from a layout shape guarantees an exact core cover; the scheduler consumes requests (possibly re-queuing refinements) one tile at a time so full-chip memory stays O(tile area + active batch).

TileRequest dataclass

Read read_bbox (core + halo) and commit only core_bbox.

Source code in src/openlithohub/streaming/core_halo.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@dataclass(frozen=True)
class TileRequest:
    """Read ``read_bbox`` (core + halo) and commit only ``core_bbox``."""

    core_bbox: BoundingBox
    read_bbox: BoundingBox
    halo: HaloSpec
    tile_id: str

    @property
    def core_size(self) -> tuple[int, int]:
        return (self.core_bbox.height, self.core_bbox.width)

    @property
    def read_size(self) -> tuple[int, int]:
        return (self.read_bbox.height, self.read_bbox.width)

RefinementRequest dataclass

Verifier-agnostic advice to re-run a tile with more context.

Produced by verification plugins (prompt §12); the scheduler never needs to know why a tile was inconclusive — only which knob to turn.

Source code in src/openlithohub/streaming/core_halo.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class RefinementRequest:
    """Verifier-agnostic advice to re-run a tile with more context.

    Produced by verification plugins (prompt §12); the scheduler never
    needs to know *why* a tile was inconclusive — only which knob to turn.
    """

    tile_id: str
    action: str = "increase_halo"  # increase_halo | subdivide
    extra_halo_px: int = 0
    subdivision: int = 2

    def __post_init__(self) -> None:
        if self.action not in ("increase_halo", "subdivide"):
            raise ValueError(f"unknown refinement action: {self.action!r}")
        if self.action == "increase_halo" and self.extra_halo_px <= 0:
            raise ValueError("increase_halo refinement needs extra_halo_px > 0")
        if self.action == "subdivide" and self.subdivision < 2:
            raise ValueError("subdivide refinement needs subdivision >= 2")

TileScheduler dataclass

Sequential streaming scheduler with verifier-driven refinement.

The consumer pulls one request at a time; refinement requests re-enqueue work without the scheduler knowing anything about the verifier's maths. domain (global (H, W)) lets grown halos clip to the real layout instead of the tile's previous read region; when omitted, the previous read extent is used as the clip bound.

Source code in src/openlithohub/streaming/core_halo.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
@dataclass
class TileScheduler:
    """Sequential streaming scheduler with verifier-driven refinement.

    The consumer pulls one request at a time; refinement requests re-enqueue
    work without the scheduler knowing anything about the verifier's maths.
    ``domain`` (global ``(H, W)``) lets grown halos clip to the real layout
    instead of the tile's previous read region; when omitted, the previous
    read extent is used as the clip bound.
    """

    requests: list[TileRequest]
    domain: tuple[int, int] | None = None
    _pending: list[TileRequest] = field(default_factory=list, init=False)

    def __post_init__(self) -> None:
        self._pending = list(self.requests)

    def __iter__(self) -> Iterator[TileRequest]:
        return self

    def __next__(self) -> TileRequest:
        if not self._pending:
            raise StopIteration
        return self._pending.pop(0)

    def enqueue(self, requests: list[TileRequest]) -> None:
        self._pending.extend(requests)

    def enqueue_refinement(
        self, request: TileRequest, refinement: RefinementRequest
    ) -> list[TileRequest]:
        """Re-queue refined work for an inconclusive tile."""
        if refinement.action == "subdivide":
            refined = subdivide_request(request, refinement.subdivision)
        else:
            grown = HaloSpec.uniform(
                max(
                    request.halo.left,
                    request.halo.top,
                    request.halo.right,
                    request.halo.bottom,
                )
                + refinement.extra_halo_px
            )
            if self.domain is not None:
                clip_w, clip_h = self.domain
            else:
                clip_w = max(request.read_bbox.x1, request.core_bbox.x1)
                clip_h = max(request.read_bbox.y1, request.core_bbox.y1)
            read = read_bbox_for(request.core_bbox, grown, clip_w, clip_h)
            refined = [
                TileRequest(
                    core_bbox=request.core_bbox,
                    read_bbox=read,
                    halo=halo_actual(request.core_bbox, read),
                    tile_id=request.tile_id,
                )
            ]
        self.enqueue(refined)
        return refined

enqueue_refinement(request, refinement)

Re-queue refined work for an inconclusive tile.

Source code in src/openlithohub/streaming/core_halo.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def enqueue_refinement(
    self, request: TileRequest, refinement: RefinementRequest
) -> list[TileRequest]:
    """Re-queue refined work for an inconclusive tile."""
    if refinement.action == "subdivide":
        refined = subdivide_request(request, refinement.subdivision)
    else:
        grown = HaloSpec.uniform(
            max(
                request.halo.left,
                request.halo.top,
                request.halo.right,
                request.halo.bottom,
            )
            + refinement.extra_halo_px
        )
        if self.domain is not None:
            clip_w, clip_h = self.domain
        else:
            clip_w = max(request.read_bbox.x1, request.core_bbox.x1)
            clip_h = max(request.read_bbox.y1, request.core_bbox.y1)
        read = read_bbox_for(request.core_bbox, grown, clip_w, clip_h)
        refined = [
            TileRequest(
                core_bbox=request.core_bbox,
                read_bbox=read,
                halo=halo_actual(request.core_bbox, read),
                tile_id=request.tile_id,
            )
        ]
    self.enqueue(refined)
    return refined

plan_tile_requests(shape, core_size, halo_px, *, core_stride=None, prefix='tile')

Plan core-exact-cover tile requests for a layout shape.

Source code in src/openlithohub/streaming/core_halo.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def plan_tile_requests(
    shape: tuple[int, int],
    core_size: int,
    halo_px: int | HaloSpec,
    *,
    core_stride: int | None = None,
    prefix: str = "tile",
) -> list[TileRequest]:
    """Plan core-exact-cover tile requests for a layout shape."""
    halo = HaloSpec.uniform(halo_px) if isinstance(halo_px, int) else halo_px
    h, w = shape
    requests: list[TileRequest] = []
    for idx, core in enumerate(core_grid_boxes(shape, core_size, core_stride=core_stride)):
        read = read_bbox_for(core, halo, w, h)
        requests.append(
            TileRequest(
                core_bbox=core,
                read_bbox=read,
                halo=halo_actual(core, read),
                tile_id=f"{prefix}_{idx}",
            )
        )
    return requests

subdivide_request(request, subdivision)

Split a tile's core into subdivision^2 smaller cores.

Source code in src/openlithohub/streaming/core_halo.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def subdivide_request(request: TileRequest, subdivision: int) -> list[TileRequest]:
    """Split a tile's core into ``subdivision^2`` smaller cores."""
    ch, cw = request.core_size
    sub_h = max(1, ch // subdivision)
    sub_w = max(1, cw // subdivision)
    out: list[TileRequest] = []
    core = request.core_bbox
    idx = 0
    for y0 in range(core.y0, core.y1, sub_h):
        for x0 in range(core.x0, core.x1, sub_w):
            sub_core = BoundingBox(
                x0=x0,
                y0=y0,
                x1=min(x0 + sub_w, core.x1),
                y1=min(y0 + sub_h, core.y1),
            )
            if sub_core.width <= 0 or sub_core.height <= 0:
                continue
            extra = HaloSpec.uniform(
                max(request.halo.left, request.halo.top, request.halo.right, request.halo.bottom)
            )
            w = request.read_bbox.x1
            h = request.read_bbox.y1
            read = read_bbox_for(
                sub_core,
                extra,
                width=max(w, sub_core.x1),
                height=max(h, sub_core.y1),
            )
            out.append(
                TileRequest(
                    core_bbox=sub_core,
                    read_bbox=read,
                    halo=halo_actual(sub_core, read),
                    tile_id=f"{request.tile_id}_s{idx}",
                )
            )
            idx += 1
    return out

tiling_overhead(requests)

Aggregate core/halo duplicate-compute statistics for a plan.

Source code in src/openlithohub/streaming/core_halo.py 
193
194
195
196
197
198
199
def tiling_overhead(requests: list[TileRequest]) -> dict[str, float]:
    """Aggregate core/halo duplicate-compute statistics for a plan."""
    cores = [r.core_bbox for r in requests]
    reads = [r.read_bbox for r in requests]
    stats = halo_overhead_stats(cores, reads)
    stats["estimated_duplicate_compute_pct"] = stats["halo_overhead_pct"]
    return stats

plan_tiling(layout_shape, *, memory_budget_bytes, bytes_per_pixel=4.0, halo_px=0, min_core_size=32, max_core_size=4096, batch=1)

Pick the largest core size that fits the per-tile memory budget.

Approximates peak per-tile memory as bytes_per_pixel * (C + 2h)^2 * batch and maximises core efficiency C^2 / (C + 2h)^2 subject to the budget (prompt §16). Callers may pass an explicit HaloPolicy-derived halo; the planner itself stays policy-agnostic.

Source code in src/openlithohub/streaming/core_halo.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def plan_tiling(
    layout_shape: tuple[int, int],
    *,
    memory_budget_bytes: int,
    bytes_per_pixel: float = 4.0,
    halo_px: int = 0,
    min_core_size: int = 32,
    max_core_size: int = 4096,
    batch: int = 1,
) -> list[TileRequest]:
    """Pick the largest core size that fits the per-tile memory budget.

    Approximates peak per-tile memory as ``bytes_per_pixel * (C + 2h)^2 *
    batch`` and maximises core efficiency ``C^2 / (C + 2h)^2`` subject to
    the budget (prompt §16).  Callers may pass an explicit
    ``HaloPolicy``-derived halo; the planner itself stays policy-agnostic.
    """
    if memory_budget_bytes <= 0:
        raise ValueError("memory_budget_bytes must be positive")
    if min_core_size <= 0 or max_core_size < min_core_size:
        raise ValueError(f"invalid core size bounds: {min_core_size}..{max_core_size}")

    best_core = min_core_size
    for core in range(min_core_size, max_core_size + 1, 8):
        read = core + 2 * halo_px
        if bytes_per_pixel * read * read * batch <= memory_budget_bytes:
            best_core = core
    h, w = layout_shape
    if best_core >= min(h, w):
        # Small layout: a single core covering the whole domain needs no halo.
        best_core = max(min(h, w), min_core_size)
        return plan_tile_requests(layout_shape, best_core, HaloSpec.uniform(0))
    return plan_tile_requests(layout_shape, best_core, HaloSpec.uniform(halo_px))

openlithohub.streaming.sources

Streaming tile sources (RFC 0008, prompt §2).

A TileSource produces the read region tensor for a tile request on demand. Implementations must never require the whole layout as a dense host tensor:

  • :class:TensorTileSource wraps an already-materialised tensor (compatibility, tests, small benchmarks).
  • :class:MemmapTensorTileSource reads windows out of an on-disk np.memmap raster (out-of-core big raster).
  • :class:VectorLayoutTileSource rasterizes only the objects that intersect the requested window of a vector GDS/OASIS layout through a :class:SpatialLayoutIndex.

TileSource

Bases: Protocol

Read-only windowed view over a full-chip layout.

Source code in src/openlithohub/streaming/sources.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@runtime_checkable
class TileSource(Protocol):
    """Read-only windowed view over a full-chip layout."""

    @property
    def shape(self) -> tuple[int, int]:
        """Global raster shape ``(H, W)`` in pixels."""
        ...

    @property
    def pixel_size_nm(self) -> float: ...

    def read_window(self, bbox: BoundingBox) -> torch.Tensor:
        """Return a ``(H, W)`` float tensor for the half-open bbox."""
        ...

shape property

Global raster shape (H, W) in pixels.

read_window(bbox)

Return a (H, W) float tensor for the half-open bbox.

Source code in src/openlithohub/streaming/sources.py 
39
40
41
def read_window(self, bbox: BoundingBox) -> torch.Tensor:
    """Return a ``(H, W)`` float tensor for the half-open bbox."""
    ...

TensorTileSource

Wrap an existing dense tensor. Zero-copy via narrow/as_strided.

Source code in src/openlithohub/streaming/sources.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class TensorTileSource:
    """Wrap an existing dense tensor. Zero-copy via narrow/as_strided."""

    def __init__(self, tensor: torch.Tensor, pixel_size_nm: float = 1.0) -> None:
        t = tensor.detach()
        if t.ndim != 2:
            raise ValueError(f"expected a 2-D layout tensor, got shape {tuple(t.shape)}")
        self._t = t
        self._pixel_nm = float(pixel_size_nm)

    @property
    def shape(self) -> tuple[int, int]:
        return (self._t.shape[0], self._t.shape[1])

    @property
    def pixel_size_nm(self) -> float:
        return self._pixel_nm

    def read_window(self, bbox: BoundingBox) -> torch.Tensor:
        return self._t[bbox.y0 : bbox.y1, bbox.x0 : bbox.x1].float()

MemmapTensorTileSource

Out-of-core windowed reads over an on-disk raster via np.memmap.

The backing file may be far larger than RAM; only requested windows are paged in. Accepts a .npy/raw path plus dtype/shape, or an existing np.memmap.

Source code in src/openlithohub/streaming/sources.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class MemmapTensorTileSource:
    """Out-of-core windowed reads over an on-disk raster via ``np.memmap``.

    The backing file may be far larger than RAM; only requested windows are
    paged in.  Accepts a ``.npy``/raw path plus dtype/shape, or an existing
    ``np.memmap``.
    """

    def __init__(
        self,
        source: str | Path | np.memmap,
        *,
        dtype: np.dtype[Any] | None = None,
        shape: tuple[int, int] | None = None,
        pixel_size_nm: float = 1.0,
    ) -> None:
        if isinstance(source, (str, Path)):
            if dtype is not None and shape is not None:
                # Raw raster: caller must supply the on-disk layout.
                self._map: np.memmap = np.memmap(source, dtype=dtype, mode="r", shape=shape)
            else:
                # .npy path: np.load with mmap_mode pages windows from disk.
                self._map = np.load(source, mmap_mode="r")
        else:
            self._map = source
        if self._map.ndim != 2:
            raise ValueError(f"expected a 2-D raster, got shape {self._map.shape}")
        self._pixel_nm = float(pixel_size_nm)

    @property
    def shape(self) -> tuple[int, int]:
        return (int(self._map.shape[0]), int(self._map.shape[1]))

    @property
    def pixel_size_nm(self) -> float:
        return self._pixel_nm

    def read_window(self, bbox: BoundingBox) -> torch.Tensor:
        window = np.asarray(self._map[bbox.y0 : bbox.y1, bbox.x0 : bbox.x1])
        return torch.from_numpy(np.ascontiguousarray(window)).float()

SpatialLayoutIndex

Minimal grid-bucket spatial index over vector layout boxes.

Each layout object is an axis-aligned bounding box plus optional fill (1.0 foreground). Objects are bucketed on a coarse uniform grid so a window query only visits intersecting buckets. This is the smallest useful realisation of the SpatialLayoutIndex concept from RFC 0008; a klayout-Region-backed implementation can drop in later behind the same interface.

Source code in src/openlithohub/streaming/sources.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class SpatialLayoutIndex:
    """Minimal grid-bucket spatial index over vector layout boxes.

    Each layout object is an axis-aligned bounding box plus optional fill
    (1.0 foreground).  Objects are bucketed on a coarse uniform grid so a
    window query only visits intersecting buckets.  This is the smallest
    useful realisation of the ``SpatialLayoutIndex`` concept from RFC 0008;
    a klayout-Region-backed implementation can drop in later behind the
    same interface.
    """

    def __init__(
        self,
        boxes: list[tuple[int, int, int, int]],
        raster_shape: tuple[int, int],
        bucket: int = 256,
    ) -> None:
        if not boxes:
            raise ValueError("SpatialLayoutIndex needs at least one layout box")
        h, w = raster_shape
        self._shape = raster_shape
        self._bucket = max(1, int(bucket))
        nbx = max(1, (w + self._bucket - 1) // self._bucket)
        nby = max(1, (h + self._bucket - 1) // self._bucket)
        self._buckets: dict[tuple[int, int], list[tuple[int, int, int, int]]] = {}
        for x0, y0, x1, y1 in boxes:
            if x1 <= x0 or y1 <= y0:
                continue
            for by in range(y0 // self._bucket, min((y1 - 1) // self._bucket, nby - 1) + 1):
                for bx in range(x0 // self._bucket, min((x1 - 1) // self._bucket, nbx - 1) + 1):
                    self._buckets.setdefault((bx, by), []).append((x0, y0, x1, y1))

    def query(self, bbox: BoundingBox) -> list[tuple[int, int, int, int]]:
        found: list[tuple[int, int, int, int]] = []
        seen: set[tuple[int, int, int, int]] = set()
        nbx = max(1, (self._shape[1] + self._bucket - 1) // self._bucket)
        nby = max(1, (self._shape[0] + self._bucket - 1) // self._bucket)
        for by in range(bbox.y0 // self._bucket, min((bbox.y1 - 1) // self._bucket, nby - 1) + 1):
            for bx in range(
                bbox.x0 // self._bucket, min((bbox.x1 - 1) // self._bucket, nbx - 1) + 1
            ):
                for box in self._buckets.get((bx, by), ()):  # may repeat across buckets
                    if box in seen:
                        continue
                    seen.add(box)
                    ix0, iy0, ix1, iy1 = box
                    ox0, oy0 = max(ix0, bbox.x0), max(iy0, bbox.y0)
                    ox1, oy1 = min(ix1, bbox.x1), min(iy1, bbox.y1)
                    if ox1 > ox0 and oy1 > oy0:
                        found.append((ox0, oy0, ox1, oy1))
        return found

VectorLayoutTileSource

Rasterize only the objects intersecting each requested window.

Source code in src/openlithohub/streaming/sources.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class VectorLayoutTileSource:
    """Rasterize only the objects intersecting each requested window."""

    def __init__(
        self,
        index: SpatialLayoutIndex,
        shape: tuple[int, int],
        pixel_size_nm: float = 1.0,
        background: float = 0.0,
        fill: float = 1.0,
    ) -> None:
        self._index = index
        self._shape = shape
        self._pixel_nm = float(pixel_size_nm)
        self._background = float(background)
        self._fill = float(fill)

    @property
    def shape(self) -> tuple[int, int]:
        return self._shape

    @property
    def pixel_size_nm(self) -> float:
        return self._pixel_nm

    def read_window(self, bbox: BoundingBox) -> torch.Tensor:
        window = np.full((bbox.height, bbox.width), self._background, dtype=np.float32)
        for x0, y0, x1, y1 in self._index.query(bbox):
            window[y0 - bbox.y0 : y1 - bbox.y0, x0 - bbox.x0 : x1 - bbox.x0] = self._fill
        return torch.from_numpy(window)

openlithohub.streaming.sinks

Streaming tile sinks (RFC 0008, prompt §3).

A TileSink receives each tile's trusted-core result. Implementations decide the output materialisation:

  • :class:TensorTileSink — assemble a full dense tensor (small layouts, backward-compatible behaviour).
  • :class:MemmapTileSink — page the output to an on-disk memmap (out-of-core big rasters).
  • :class:MetricOnlyTileSink — keep only per-tile metrics/aggregates and never materialise a raster at all (the QDM-critical mode).

Because only trusted cores are ever written, no weight map / blend pass is required: cores are an exact partition of the output domain.

TensorTileSink

Assemble the full dense output tensor in host memory.

Source code in src/openlithohub/streaming/sinks.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class TensorTileSink:
    """Assemble the full dense output tensor in host memory."""

    def __init__(self, shape: tuple[int, int], dtype: torch.dtype = torch.float32) -> None:
        self._shape = shape
        self.output = torch.zeros(shape, dtype=dtype)
        self._written: set[str] = set()

    @property
    def shape(self) -> tuple[int, int]:
        return self._shape

    def write_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        tensor: torch.Tensor,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        if tile_id in self._written:
            raise ValueError(f"duplicate core write for {tile_id!r}")
        self._written.add(tile_id)
        expected = (bbox.height, bbox.width)
        if tuple(tensor.shape) != expected:
            raise ValueError(f"core tensor shape {tuple(tensor.shape)} != bbox {expected}")
        self.output[bbox.y0 : bbox.y1, bbox.x0 : bbox.x1] = tensor

    def finalize(self) -> torch.Tensor:
        return self.output

MemmapTileSink

Page trusted cores straight into an on-disk memmap raster.

Source code in src/openlithohub/streaming/sinks.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class MemmapTileSink:
    """Page trusted cores straight into an on-disk memmap raster."""

    def __init__(
        self,
        shape: tuple[int, int],
        path: str | Path,
        dtype: np.dtype[Any] | None = None,
    ) -> None:
        self._shape = shape
        self.path = Path(path)
        if dtype is None:
            dtype = np.dtype(np.float32)
        self._map: np.memmap = np.memmap(self.path, dtype=dtype, mode="w+", shape=shape)
        self._written: set[str] = set()

    @property
    def shape(self) -> tuple[int, int]:
        return self._shape

    def write_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        tensor: torch.Tensor,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        if tile_id in self._written:
            raise ValueError(f"duplicate core write for {tile_id!r}")
        self._written.add(tile_id)
        window = tensor.detach().cpu().numpy().astype(self._map.dtype, copy=False)
        self._map[bbox.y0 : bbox.y1, bbox.x0 : bbox.x1] = window

    def finalize(self) -> np.memmap:
        self._map.flush()
        return self._map

MetricOnlyTileSink

Aggregate per-tile results without ever materialising a raster.

Reducers receive one tile at a time, so peak memory is O(1) beyond the running aggregates. This is the sink mode QDM-style verification needs (certificates, worst bounds, violation lists, provenance).

Source code in src/openlithohub/streaming/sinks.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class MetricOnlyTileSink:
    """Aggregate per-tile results without ever materialising a raster.

    Reducers receive one tile at a time, so peak memory is O(1) beyond the
    running aggregates.  This is the sink mode QDM-style verification needs
    (certificates, worst bounds, violation lists, provenance).
    """

    def __init__(self, shape: tuple[int, int]) -> None:
        self._shape = shape
        self.tiles: list[str] = []
        self.covered_pixels = 0
        self.metrics: dict[str, float] = {}
        self.metadata: dict[str, Any] = {}

    @property
    def shape(self) -> tuple[int, int]:
        return self._shape

    def write_core(
        self,
        tile_id: str,
        bbox: BoundingBox,
        tensor: torch.Tensor,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        self.tiles.append(tile_id)
        self.covered_pixels += bbox.area
        self.metadata[tile_id] = metadata or {}
        for key, value in (metadata or {}).items():
            if isinstance(value, (int, float)):
                prev = self.metrics.get(f"max_{key}", float("-inf"))
                self.metrics[f"max_{key}"] = max(prev, float(value))

    def finalize(self) -> dict[str, Any]:
        return {
            "n_tiles": len(self.tiles),
            "covered_pixels": self.covered_pixels,
            "metrics": dict(self.metrics),
        }

covered_pixels(sink)

Total trusted-core pixels written (for gap/overlap audits).

Source code in src/openlithohub/streaming/sinks.py 
156
157
158
159
160
161
162
163
164
def covered_pixels(sink: TileSink) -> int:
    """Total trusted-core pixels written (for gap/overlap audits)."""
    written = getattr(sink, "covered_pixels", None)
    if written is not None:
        return int(written)
    written = getattr(sink, "_written", None)
    if isinstance(written, Iterable):
        return len(written)  # type: ignore[arg-type]
    return -1

openlithohub.streaming.halo_policy

Halo policies (RFC 0008, prompt §4-6).

A HaloPolicy answers "how much context halo does this tile need?" and must always explain itself: every answer carries the halo, its provenance, an optional rigorous error bound, and a certification status. Policies composing means taking the max halo; the strictest provenance/certification status wins.

HaloContext dataclass

Everything a policy may need to size a halo.

Source code in src/openlithohub/streaming/halo_policy.py 
85
86
87
88
89
90
91
92
93
94
@dataclass(frozen=True)
class HaloContext:
    """Everything a policy may need to size a halo."""

    pixel_nm: float
    tile_core_px: int
    node: ProcessNodeConfig | None = None
    model: LithographyModel | None = None
    kernel: torch.Tensor | None = None
    kernel_tolerance: float | None = None

LegacyFixedHaloPolicy

Reproduce the pre-RFC-0005 fixed halo (default 128 px).

Source code in src/openlithohub/streaming/halo_policy.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class LegacyFixedHaloPolicy:
    """Reproduce the pre-RFC-0005 fixed halo (default 128 px)."""

    name = "legacy_fixed"

    def __init__(self, halo_px: int = DEFAULT_HALO_PX) -> None:
        if halo_px < 0:
            raise ValueError(f"halo_px must be nonnegative, got {halo_px}")
        self._halo_px = halo_px

    def required_halo(self, context: HaloContext) -> HaloRequirement:
        return HaloRequirement(
            halo_px=min(self._halo_px, max(context.tile_core_px - 1, 0)),
            provenance="legacy_fixed",
            error_bound=None,
            certified=False,
            reason=f"fixed legacy halo of {self._halo_px} px (backward compatibility)",
            status=HaloStatus.FIXED,
        )

PhysicalInteractionHaloPolicy

h = max(OIR_px, receptive-field px), RFC-0005 logic moved here.

Source code in src/openlithohub/streaming/halo_policy.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class PhysicalInteractionHaloPolicy:
    """``h = max(OIR_px, receptive-field px)``, RFC-0005 logic moved here."""

    name = "physical_interaction"

    def __init__(self, round_up_to: int = _HALO_ROUND_PX) -> None:
        self._round = max(1, round_up_to)

    def required_halo(self, context: HaloContext) -> HaloRequirement:
        parts: list[str] = []
        px = 0
        if context.node is not None and context.pixel_nm > 0:
            oir_px = math.ceil(context.node.optical_radius_nm / context.pixel_nm)
            parts.append(f"optical interaction radius {oir_px} px")
            px = max(px, oir_px)
        if context.model is not None:
            rf = getattr(context.model, "RECEPTIVE_FIELD_PX", 0) or 0
            if rf:
                parts.append(f"model receptive field {rf} px")
            px = max(px, rf)
        halo = _round_up(px) if self._round > 1 else px
        return HaloRequirement(
            halo_px=min(halo, max(context.tile_core_px - 1, 0)),
            provenance="physical_interaction",
            error_bound=None,
            certified=False,
            reason="max(" + (", ".join(parts) if parts else "no physics source") + ")",
            status=HaloStatus.PHYSICS_ESTIMATED,
        )

KernelTailHaloPolicy

Pick the smallest halo whose kernel tail mass <= tolerance.

T(h) = sum_{|x|>h} |K(x)| <= eps over the sampled spatial kernel. When the sample truly bounds the full kernel, the emitted halo is CERTIFIED_SUFFICIENT with tail mass as the error bound; if the caller cannot vouch for the sample, pass trusted_sample=False and the answer stays PHYSICS_ESTIMATED (never pretend certification).

Source code in src/openlithohub/streaming/halo_policy.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
class KernelTailHaloPolicy:
    """Pick the smallest halo whose kernel tail mass <= tolerance.

    ``T(h) = sum_{|x|>h} |K(x)| <= eps`` over the sampled spatial kernel.
    When the sample truly bounds the full kernel, the emitted halo is
    CERTIFIED_SUFFICIENT with ``tail mass`` as the error bound; if the
    caller cannot vouch for the sample, pass ``trusted_sample=False`` and
    the answer stays PHYSICS_ESTIMATED (never pretend certification).
    """

    name = "kernel_tail"

    def __init__(self, tolerance: float = 1e-3, trusted_sample: bool = True) -> None:
        if tolerance <= 0.0:
            raise ValueError(f"tolerance must be positive, got {tolerance}")
        self._tolerance = float(tolerance)
        self._trusted = bool(trusted_sample)

    def required_halo(self, context: HaloContext) -> HaloRequirement:
        kernel = context.kernel
        if kernel is None:
            return HaloRequirement(
                halo_px=DEFAULT_HALO_PX,
                provenance="kernel_tail",
                error_bound=None,
                certified=False,
                reason="no kernel sample available; fell back to legacy default",
                status=HaloStatus.INCONCLUSIVE,
            )
        k = kernel.detach().abs()
        total = float(k.sum())
        if total <= 0.0:
            return HaloRequirement(
                halo_px=DEFAULT_HALO_PX,
                provenance="kernel_tail",
                error_bound=None,
                certified=False,
                reason="degenerate (all-zero) kernel sample",
                status=HaloStatus.INCONCLUSIVE,
            )
        # Box-aligned tail sweep: the halo keeps a (2h+1)^2 central box, so
        # the tail at halo h is exactly kernel_tail_mass(k, h). Find the
        # smallest h whose tail meets the tolerance.
        max_r = min(k.shape[0], k.shape[1]) // 2
        best_h: int | None = None
        for h in range(0, max_r + 1):
            if kernel_tail_mass(k, h) <= self._tolerance:
                best_h = h
                break
        if best_h is None:
            # even keeping the whole sampled kernel leaves the tolerance
            # unmet only when the sample is unnormalizable noise; report the
            # tail at the largest box.
            tail = kernel_tail_mass(k, max_r)
            return HaloRequirement(
                halo_px=_round_up(max_r),
                provenance="kernel_tail",
                error_bound=tail,
                certified=False,
                reason=(
                    f"kernel tail {tail:.3e} still exceeds tolerance "
                    f"{self._tolerance:.3e} at the largest sampled radius"
                ),
                status=HaloStatus.INCONCLUSIVE,
            )
        halo = min(_round_up(best_h), max(context.tile_core_px - 1, 0))
        return HaloRequirement(
            halo_px=halo,
            provenance="kernel_tail",
            error_bound=self._tolerance if self._trusted else None,
            certified=self._trusted,
            reason=(f"smallest sampled radius with kernel tail mass <= {self._tolerance:.3e}"),
            status=(
                HaloStatus.CERTIFIED_SUFFICIENT if self._trusted else HaloStatus.PHYSICS_ESTIMATED
            ),
        )

combine_requirements(*requirements)

Max-combine halo requirements; strictest provenance wins.

Any non-certified participant demotes the combination, and any INCONCLUSIVE participant makes the whole answer inconclusive — a halo is only as trustworthy as its weakest contributor.

Source code in src/openlithohub/streaming/halo_policy.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def combine_requirements(*requirements: HaloRequirement | None) -> HaloRequirement:
    """Max-combine halo requirements; strictest provenance wins.

    Any non-certified participant demotes the combination, and any
    INCONCLUSIVE participant makes the whole answer inconclusive — a halo
    is only as trustworthy as its weakest contributor.
    """
    live = [r for r in requirements if r is not None]
    if not live:
        raise ValueError("combine_requirements needs at least one requirement")
    halo = max(r.halo_px for r in live)
    provenance = "+".join(r.provenance for r in live)
    bounds = [r.error_bound for r in live if r.error_bound is not None]
    error_bound = max(bounds) if bounds else None
    if any(r.status is HaloStatus.INCONCLUSIVE for r in live):
        status = HaloStatus.INCONCLUSIVE
        certified = False
    elif all(r.status is HaloStatus.CERTIFIED_SUFFICIENT for r in live):
        status = HaloStatus.CERTIFIED_SUFFICIENT
        certified = all(r.certified for r in live)
    else:
        status = HaloStatus.PHYSICS_ESTIMATED
        certified = False
    reason = "; ".join(r.reason for r in live)
    return HaloRequirement(
        halo_px=halo,
        provenance=provenance,
        error_bound=error_bound,
        certified=certified,
        reason=reason,
        status=status,
    )

kernel_tail_mass(kernel, halo_px)

Fraction of |kernel| mass outside the central (2h+1)^2 box.

Source code in src/openlithohub/streaming/halo_policy.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def kernel_tail_mass(kernel: torch.Tensor, halo_px: int) -> float:
    """Fraction of |kernel| mass outside the central ``(2h+1)^2`` box."""
    k = kernel.detach().abs()
    total = float(k.sum())
    if total <= 0.0:
        return 0.0
    cy, cx = k.shape[0] // 2, k.shape[1] // 2
    # Clamp so a halo larger than the kernel radius cannot wrap via
    # negative slicing — beyond the kernel radius the tail is exactly 0.
    y0, y1 = max(0, cy - halo_px), min(k.shape[0], cy + halo_px + 1)
    x0, x1 = max(0, cx - halo_px), min(k.shape[1], cx + halo_px + 1)
    keep = torch.zeros_like(k, dtype=torch.bool)
    keep[y0:y1, x0:x1] = True
    return float(k[~keep].sum()) / total

estimate_minimum_halo(forward_fn, *, core_bbox_px, full_context, candidate_halos=(0, 8, 16, 32, 64, 128), tolerance=0.001)

Adaptive empirical halo sizing (prompt §6).

Compares the forward result restricted to a fixed core as the halo grows. The first candidate whose discrepancy with the final (largest-halo) reference drops under tolerance — and stays there for the remaining candidates — is returned as EMPIRICALLY_STABLE. This is explicitly not a mathematical certificate; a rigorous bound needs an analytic kernel-tail or QDM argument.

Source code in src/openlithohub/streaming/halo_policy.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def estimate_minimum_halo(
    forward_fn: Callable[[torch.Tensor], torch.Tensor],
    *,
    core_bbox_px: tuple[int, int, int, int],
    full_context: torch.Tensor,
    candidate_halos: Iterable[int] = (0, 8, 16, 32, 64, 128),
    tolerance: float = 1e-3,
) -> HaloRequirement:
    """Adaptive empirical halo sizing (prompt §6).

    Compares the forward result restricted to a fixed core as the halo
    grows.  The first candidate whose discrepancy with the final
    (largest-halo) reference drops under ``tolerance`` — and stays there
    for the remaining candidates — is returned as EMPIRICALLY_STABLE.
    This is explicitly *not* a mathematical certificate; a rigorous bound
    needs an analytic kernel-tail or QDM argument.
    """
    x0, y0, x1, y1 = core_bbox_px
    core_h, core_w = y1 - y0, x1 - x0
    candidates = sorted({int(h) for h in candidate_halos if h >= 0})
    if not candidates:
        raise ValueError("need at least one candidate halo")

    core_errs: list[tuple[int, torch.Tensor]] = []
    for halo in candidates:
        tile = full_context[max(0, y0 - halo) : y1 + halo, max(0, x0 - halo) : x1 + halo]
        out = forward_fn(tile)
        # Extract the core response from the (possibly clipped) read region.
        oy = y0 - max(0, y0 - halo)
        ox = x0 - max(0, x0 - halo)
        core_out = out[oy : oy + core_h, ox : ox + core_w]
        core_errs.append((halo, core_out))

    reference_h, reference = core_errs[-1]
    stable_from: int | None = None
    for i in range(len(core_errs) - 1):
        h_i, out_i = core_errs[i]
        disc = float(torch.max(torch.abs(out_i - reference)).item())
        if disc <= tolerance and stable_from is None:
            stable_from = h_i
        elif disc > tolerance:
            stable_from = None
    if stable_from is None:
        return HaloRequirement(
            halo_px=reference_h,
            provenance="empirical_sweep",
            error_bound=None,
            certified=False,
            reason=f"core discrepancy never stabilised under tolerance {tolerance:g}",
            status=HaloStatus.INCONCLUSIVE,
        )
    return HaloRequirement(
        halo_px=stable_from,
        provenance="empirical_sweep",
        error_bound=tolerance,
        certified=False,
        reason=(
            f"core discrepancy <= {tolerance:g} for all halos >= {stable_from} px "
            f"(reference halo {reference_h} px)"
        ),
        status=HaloStatus.EMPIRICALLY_STABLE,
    )

openlithohub.streaming.pipeline

Streaming full-chip pipeline (RFC 0008, prompt §11).

Runs the per-tile loop::

TileSource.read_window(core+halo)
    → forward model
    → optional VerificationPlugin(s)
    → TileSink.write_core(trusted core only)
    → discard tile tensors

Peak memory is O(tile area + active batch), never O(full-chip raster). Every core is written exactly once; an inconclusive verdict re-runs the same core with a larger halo (bounded by max_requeues) before the best-effort result is committed.

BoundingBox dataclass

Half-open pixel rectangle [x0, x1) x [y0, y1) in global coords.

Source code in src/openlithohub/streaming/geometry.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@dataclass(frozen=True)
class BoundingBox:
    """Half-open pixel rectangle ``[x0, x1) x [y0, y1)`` in global coords."""

    x0: int
    y0: int
    x1: int
    y1: int

    def __post_init__(self) -> None:
        if self.x1 <= self.x0 or self.y1 <= self.y0:
            raise ValueError(f"degenerate bbox: x=[{self.x0},{self.x1}) y=[{self.y0},{self.y1})")

    @property
    def width(self) -> int:
        return self.x1 - self.x0

    @property
    def height(self) -> int:
        return self.y1 - self.y0

    @property
    def area(self) -> int:
        return self.width * self.height

    def clipped_to(self, width: int, height: int) -> BoundingBox:
        """Clip to the global domain ``[0,width) x [0,height)``.

        Returns a bbox that may shrink; callers must treat an empty
        intersection (degenerate result) as "no overlap".
        """
        return BoundingBox(
            x0=max(0, self.x0),
            y0=max(0, self.y0),
            x1=min(width, self.x1),
            y1=min(height, self.y1),
        )

    def contains(self, other: BoundingBox) -> bool:
        return (
            self.x0 <= other.x0
            and self.y0 <= other.y0
            and self.x1 >= other.x1
            and self.y1 >= other.y1
        )

clipped_to(width, height)

Clip to the global domain [0,width) x [0,height).

Returns a bbox that may shrink; callers must treat an empty intersection (degenerate result) as "no overlap".

Source code in src/openlithohub/streaming/geometry.py 
44
45
46
47
48
49
50
51
52
53
54
55
def clipped_to(self, width: int, height: int) -> BoundingBox:
    """Clip to the global domain ``[0,width) x [0,height)``.

    Returns a bbox that may shrink; callers must treat an empty
    intersection (degenerate result) as "no overlap".
    """
    return BoundingBox(
        x0=max(0, self.x0),
        y0=max(0, self.y0),
        x1=min(width, self.x1),
        y1=min(height, self.y1),
    )

HaloSpec dataclass

Per-side halo widths requested for a tile read region.

Source code in src/openlithohub/streaming/geometry.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True)
class HaloSpec:
    """Per-side halo widths requested for a tile read region."""

    left: int
    top: int
    right: int
    bottom: int

    @classmethod
    def uniform(cls, px: int) -> HaloSpec:
        if px < 0:
            raise ValueError(f"halo must be nonnegative, got {px}")
        return cls(left=px, top=px, right=px, bottom=px)

    @property
    def total(self) -> int:
        return self.left + self.top + self.right + self.bottom

run_streaming(source, sink, forward_fn, *, core_size, halo_policy=None, verifiers=(), max_halo_px=1024, pixel_nm=1.0, max_requeues=4)

Process a full chip tile-by-tile under core/halo ownership.

Source code in src/openlithohub/streaming/pipeline.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def run_streaming(
    source: TileSource,
    sink: TileSink,
    forward_fn: Callable[[torch.Tensor], torch.Tensor],
    *,
    core_size: int,
    halo_policy: HaloPolicy | None = None,
    verifiers: Iterable[VerificationPlugin] = (),
    max_halo_px: int = 1024,
    pixel_nm: float = 1.0,
    max_requeues: int = 4,
) -> StreamingRunReport:
    """Process a full chip tile-by-tile under core/halo ownership."""
    policy = halo_policy or LegacyFixedHaloPolicy()
    verifier_list = list(verifiers)

    vctx = VerificationContext(model=forward_fn, pixel_nm=pixel_nm)
    plugin_requirements = []
    for verifier in verifier_list:
        requirement = verifier.required_halo(vctx)
        if requirement is not None:
            plugin_requirements.append(requirement)

    physical = policy.required_halo(HaloContext(pixel_nm=pixel_nm, tile_core_px=core_size))
    if plugin_requirements:
        from .halo_policy import combine_requirements

        requirement = combine_requirements(physical, *plugin_requirements)
    else:
        requirement = physical
    halo_px = min(requirement.halo_px, max_halo_px)

    reducer = StreamingVerificationReducer()
    report = StreamingRunReport(halo_requirement=requirement)
    for verifier in verifier_list:
        verifier.prepare(vctx)

    for base in plan_tile_requests(source.shape, core_size, halo_px):
        current = base
        refinements_left = max_requeues
        while True:
            tile = source.read_window(current.read_bbox)
            source_meta: dict[str, Any] = {}
            metadata_fn = getattr(source, "verification_metadata", None)
            if callable(metadata_fn):
                source_meta = dict(
                    metadata_fn(
                        current.read_bbox,
                        tile_id=current.tile_id,
                        core_bbox=current.core_bbox,
                    )
                )
            result = forward_fn(tile)
            ys, xs = _core_slices(current)
            core_result = result[ys, xs]

            tile_meta: dict[str, Any] = {}
            refinement: RefinementRequest | None = None
            for verifier in verifier_list:
                tctx = TileContext(
                    tile_id=current.tile_id,
                    core_bbox=current.core_bbox,
                    read_bbox=current.read_bbox,
                    halo=current.halo,
                    tensor=tile,
                    metadata=source_meta,
                )
                verdict = verifier.verify_tile(tctx)
                reducer.add(verdict)
                tile_meta[f"{verifier.name}_status"] = verdict.status
                if verdict.status == "INCONCLUSIVE" and refinement is None:
                    if refinements_left > 0:
                        refinement = _default_refinement(current, verifier)
                    else:
                        tile_meta[f"{verifier.name}_note"] = (
                            "inconclusive; refinement budget exhausted"
                        )

            if refinement is None:
                sink.write_core(current.tile_id, current.core_bbox, core_result, tile_meta)
                report.n_tiles += 1
                break

            refinements_left -= 1
            report.n_requeued += 1
            current = _grown_request(current, refinement, source.shape)
            del tile, result, core_result

    report.overhead = tiling_overhead(plan_tile_requests(source.shape, core_size, halo_px))
    if verifier_list:
        report.verification = reducer.finalize()
    return report