pipekit-ops¶
The pipekit core, imported as pipekit_ops: carrier-agnostic operators
and operator graphs. Pure Python, no third-party dependencies. Until 0.2.0
this package was pipekit. Master plan reference: Report 2.
Foundations¶
pipekit_ops._base.operator
¶
Operator base class — the composition primitive.
Every concrete operator (MaskClouds, NDVI, ModelOp, …) subclasses
Operator and implements _apply. The base class handles three
behaviours every operator inherits:
- Dual-mode
__call__—op(x)runs eagerly on data;op(node)builds aNodein aGraph. Subclasses only implement_apply. - Config round-trip via
ConfigMixin—get_config()is auto-derived from__init__parameter names by reading matching instance attributes. Subclasses opt out with__config_mixin_auto__ = Falseor exclude fields with__config_exclude__ = ("foo",). - JSON-safe
state/from_state— a serialisable record that round-trips through any subclass ofOperatorthat's been imported, nested operators included: a child operator appears in its parent's config as its ownstaterecord, andfrom_staterebuilds the tree from the leaves up.
Operators that themselves hold non-serialisable runtime state — user
closures (Tap, Lambda, Branch, …) or runtime arrays — set
forbid_in_yaml = True to signal that get_config() is a debug repr,
not a faithful round-trip; from_state refuses flagged classes, at any
depth, and parallel.check_pickleable surfaces flagged instances as
pickling hazards. Operators that merely nest other operators (Cache,
Try, the parallel maps, …) do not set the flag: they round-trip, and
nesting is discovered structurally by check_pickleable.
See master plan Report 2, Group A.
ConfigMixin
¶
Auto-derive get_config() from __init__ parameter names.
Reads inspect.signature(type(self).__init__), then for each
declared parameter name looks up the matching instance attribute on
self. The discipline: store __init__ kwargs as attributes of the
same name (the geotoolz / xr_toolz convention).
Opt-out: set __config_mixin_auto__ = False to disable
auto-derivation entirely (subclasses then must override
get_config themselves).
Exclude: set __config_exclude__ = ("field_a",) to skip specific
parameters (useful for runtime-only state).
Subclasses can override get_config directly when auto-derivation
isn't appropriate (e.g. Sequential whose config is a list of
nested operators, not constructor parameters).
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
get_config()
¶
Return a JSON-serialisable dict of constructor args.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
Operator
¶
Bases: ConfigMixin
Base class for pipekit operators.
Subclasses implement _apply(self, *args, **kwargs). The base
class dispatches __call__ to either _apply (eager) or
Node construction (symbolic) based on whether any positional
argument is a graph node.
Attributes:
| Name | Type | Description |
|---|---|---|
forbid_in_yaml |
bool
|
|
_terminal |
bool
|
|
backends |
tuple[str, ...] | None
|
The array backends the operator runs on ( |
Example
Define and compose::
class Scale(Operator):
def __init__(self, factor: float) -> None:
self.factor = factor
def _apply(self, x):
return x * self.factor
pipe = Scale(0.5) | Scale(2.0) # Sequential([Scale(0.5), Scale(2.0)])
assert pipe(10.0) == 10.0
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
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 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 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 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 | |
state
property
¶
Return a JSON-serialisable state record for this operator.
__call__(*args, **kwargs)
¶
Dispatch eager vs symbolic on positional arguments.
If any positional argument is a Node (which Input subclasses),
all positional args must be Nodes and the call returns a new
Node recording this operator and its parents. Otherwise the
call routes to _apply. Subclasses override _apply, not
__call__.
Raises:
| Type | Description |
|---|---|
BackendError
|
|
TypeError
|
if graph-construction mode is triggered (at
least one |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
__or__(other)
¶
op1 | op2 returns Sequential([op1, op2]), flattening on the right.
a | (b | c) produces a single three-element Sequential.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
compute_output_signature(input_signature)
¶
Return the signature of this operator's output given an input signature.
Default is passthrough — most carrier-agnostic operators preserve
shape and dtype. Subclasses that reshape / reduce / retype override.
Returning None signals "this operator does not track shape"
(carriers without named dimensions); Sequential.summary and
Graph.summary show ? for it, and every later step in a
Sequential is unknown too.
Handed a carrier instead of a signature, the default reads the
carrier's Signature through its registered adapter
(pipekit_ops.adapters.signature_of), so Sequential.summary can
start from data.
The annotation is Any rather than Signature | None so domain
libraries can thread their own signature types.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
from_config(config)
classmethod
¶
Build an operator from its decoded get_config() record.
The default calls cls(**config), which is right for every
operator whose config keys are its constructor's arguments.
Operators whose config is not a constructor call (Graph,
whose config is a topology) override this.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
The config with nested operators already rebuilt. |
required |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
from_state(state)
classmethod
¶
Reconstruct an operator, and any operators nested in it, from state.
Resolves classes among cls and all its transitive subclasses,
so Operator.from_state(state) resolves any imported subclass;
nothing is imported. Config values are rebuilt recursively: a
nested state record becomes an operator, an
{"exception": "module.Name"} record becomes that (already
imported) exception class, lists and dicts are rebuilt item by
item, and JSON primitives pass through. The class's
from_config then builds the operator.
Raises:
| Type | Description |
|---|---|
ValueError
|
if a state record is malformed. |
TypeError
|
if a class isn't a loaded |
RuntimeError
|
if any class in the tree is marked
|
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
build_operator(state, resolve)
¶
Rebuild an operator tree from a state record.
The engine behind Operator.from_state and the sandboxed loader:
resolve(module, class) maps each class name in the tree to a
class (and may refuse, by raising); everything else is shared.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state
|
dict[str, Any]
|
A |
required |
resolve
|
Resolver
|
Maps a module and class name to an |
required |
Raises:
| Type | Description |
|---|---|
ValueError / TypeError / RuntimeError
|
as |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
callable_name(fn)
¶
Best-effort display name for a user-supplied callable.
__name__ when present (functions, classes), else repr (lambdas
assigned through functools.partial, callable instances, …).
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
current_module(module_name)
¶
Today's name for a module saved before pipekit 0.2.0.
The package was imported as pipekit until 0.2.0 and as pipekit_ops
since, so a state saved earlier names pipekit.control where today's
says pipekit_ops.control. Every other module name passes through.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
exception_config(exc)
¶
Config payload for an exception class (Try.on, Retry.on).
{"exception": "module.QualName"}, which Operator.from_state
resolves back to the class among already-imported modules.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
nested_config(op)
¶
Config payload for an operator nested inside another's config.
The child's full state record (module, class, config),
which Operator.from_state rebuilds. Every container operator
(Sequential, Cache, Branch, the parallel maps, …) emits its
children through this one helper, so the payload can never drift
between operators.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
require_operator(value, owner, field)
¶
Raise a standardized TypeError unless value is an Operator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
Any
|
The candidate object. |
required |
owner
|
str
|
The validating class's name (for the message). |
required |
field
|
str
|
The parameter/attribute name (may include an index, e.g.
|
required |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
resolve_loaded(module_name, class_name)
¶
The loaded Operator subclass named module.class; nothing is imported.
A class that moved lists its old modules in __former_modules__, so
states saved before the move still load; so do states saved under the
package's pre-0.2.0 name, pipekit (see current_module).
Raises:
| Type | Description |
|---|---|
TypeError
|
if no loaded |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/operator.py
pipekit_ops._base.sequential
¶
Sequential — eager linear composition of operators.
Sequential is the workhorse pipekit pipeline: a list of operators
applied left-to-right, each consuming the previous one's output.
pipe = Sequential([Scale(0.5), Scale(2.0)])
result = pipe(10.0)
Equivalent and idiomatic via the | operator on Operator::
pipe = Scale(0.5) | Scale(2.0)
Sequential is itself an Operator, so it composes recursively. The
right-fold __or__ keeps nested Sequentials flat.
Construction-time invariants:
- No terminal operators in non-terminal position. Subclasses
marked
_terminal = True(write-to-disk, viz, etc.) can only appear as the last step. Catching this at construction gives a clear error rather than a confusing runtime failure. - All steps are
Operatorinstances. Same logic — clearer error than anAttributeErrordeep in the chain.
See master plan Report 2, Group A.2.
Sequential
¶
Bases: Operator
Apply a list of operators in order, threading the output of each into the next.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
operators
|
list[Operator]
|
A list of |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
if any element is not an |
Example
Basic::
pipe = Sequential([Scale(0.5), Scale(2.0)])
assert pipe(10.0) == 10.0
Slice into a sub-pipeline::
head = pipe[:1] # Sequential([Scale(0.5)])
Index a single step::
first = pipe[0] # Scale(0.5)
Source code in packages/pipekit-ops/src/pipekit_ops/_base/sequential.py
42 43 44 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 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 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 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 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 | |
is_stateful
property
¶
True if any contained operator is a StatefulOperator.
Such a pipeline is called as pipe(carrier, state) and returns
(carrier, state), so it can step a pipekit-cycle cycle.
__getitem__(key)
¶
Index or slice into the underlying operator list.
pipe[0] returns the first operator. pipe[1:3] returns a
new Sequential containing the slice.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/sequential.py
__or__(other)
¶
Append on the right; flatten nested Sequentials.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/sequential.py
compute_output_signature(input_signature)
¶
Thread the signature through every operator in order.
A registered carrier is read as its Signature first.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/sequential.py
describe(indent=0)
¶
Return an indented-tree pretty-print of the pipeline.
Recurses into nested Sequential / Graph operators. Useful
for notebook exploration; not parsed.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/sequential.py
summary(input_signature=None)
¶
Render a Keras-style summary table.
Threads input_signature through compute_output_signature;
a registered carrier is read as its Signature first. If any
operator returns None (does not track shape), the summary
still lists the operator but shows ? for its output
signature. With input_signature=None, only operator names
and types are shown.
The return type is str rather than printing directly so the
caller decides where it goes (notebook, log, file).
Source code in packages/pipekit-ops/src/pipekit_ops/_base/sequential.py
pipekit_ops._base.graph
¶
Graph composition — symbolic operator graphs.
Sequential covers the linear case; Graph covers the rest:
branching outputs, multi-input fusion, diamond dependencies.
Construction is symbolic — calling an Operator on Input / Node
instances builds up a graph; running it is _apply(**inputs_by_name)
(or positionally).
Pattern::
img = Input("image")
ref = Input("reference")
ndvi = NDVI(red_idx=2, nir_idx=3)(img)
rmse = RMSE(axis=(-2, -1))(ndvi, ref)
g = Graph(
inputs={"image": img, "reference": ref},
outputs={"ndvi": ndvi, "rmse": rmse},
)
result = g(image=img_gt, reference=ref_gt)
# {"ndvi": GeoTensor, "rmse": scalar}
Input is a subclass of Node (master plan Report 2.A.3): the
topological sort can treat all vertices uniformly, with the special
case being "this node has no operator — supply its value from the
caller's kwargs".
See master plan Report 2, Group A.3.
Graph
¶
Bases: Operator
A symbolic operator graph with multiple inputs and outputs.
Construction is implicit — calling operators on Input / Node
instances builds the graph; Graph(inputs=..., outputs=...)
wraps the result. _apply(**inputs) evaluates in topological
order.
Inherits from Operator so a Graph satisfies the same interface
as any other operator. Operator.__call__ dispatches keyword
args straight through to Graph._apply; positional args are bound
to declared Inputs in declaration order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
dict[str, Input]
|
Map of |
required |
outputs
|
dict[str, Node]
|
Map of |
required |
State
A graph that contains a stateful node (a StatefulOperator, or
a Sequential / Graph that contains one) threads a carry state
the way Sequential does: call it as
graph(a=..., b=..., state=s) and it returns
(outputs, state). The state passes through the stateful nodes
in topological order (the order describe lists); stateless
nodes ignore it. Inside a stateful Sequential the graph receives
(carrier, state) positionally, like any stateful step.
Raises:
| Type | Description |
|---|---|
ValueError
|
if the graph contains a cycle, if an |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
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 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 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 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 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 | |
compute_output_signature(input_signature)
¶
None: a graph returns a dict of outputs, not one carrier.
Use output_signatures for each output's signature, and
summary for the whole table.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
describe()
¶
Return a short topology description.
Lists inputs, internal nodes (in topological order), and outputs. Useful for notebook exploration.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
from_config(config)
classmethod
¶
Rebuild a graph from the topology get_config records.
Re-creates each Input, then calls each node's operator on its
parents in the recorded order, which wires the nodes exactly as
they were; an operator shared by several nodes stays one object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
A |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
if a reference names an undeclared input or a node that comes later in the order. |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
get_config()
¶
The whole topology: inputs, operators, nodes and outputs.
Every node is recorded, upstream ones included, so two graphs
that differ anywhere have different configs (and so different
state records and content hashes). The record::
{
"inputs": ["image", "reference"],
"operators": [<state>, ...],
"nodes": [{"operator": 0, "parents": [{"input": "image"}]}, ...],
"outputs": {"ndvi": {"node": 0}, "raw": {"input": "image"}},
}
operators lists each distinct operator once, by identity, in
first-use order, so an operator shared by several nodes is not
duplicated. nodes is in topological order; a parent or an
output refers either to a declared input by name or to an
earlier node by index.
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
output_signatures(input_signatures)
¶
Each output's signature, propagated node by node without data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_signatures
|
Any
|
|
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
|
dict[str, Any]
|
operator does not track shape, or where a multi-input node's |
dict[str, Any]
|
inputs disagree (see |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
summary(input_signatures=None)
¶
A Keras-style table of the nodes, their input and output signatures.
Nodes are listed in topological order. Signatures propagate
through each node's compute_output_signature; a node with
several inputs gets its operator's signature for them only when
they all agree (an elementwise join), and ? otherwise, since
compute_output_signature takes one input. The outputs follow.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_signatures
|
Any
|
As for |
None
|
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
Input
¶
Bases: Node
A named entry point into a Graph.
Operator.__call__ recognises Node (and any subclass, including
Input) as graph-construction mode. Graph._apply supplies the
value at run-time from the caller's kwargs / positional args.
Subclassing Node (master plan Report 2.A.3, "xr_toolz shape")
lets the topological sort treat all vertices uniformly with one
special case: "this node has no operator — source it from the
caller's inputs".
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
Node
¶
A vertex in a Graph.
Created automatically by Operator.__call__ when any positional
argument is a Node (or Input, which subclasses Node). Carries
the operator and its parent vertices. Compared by identity so the
id(...)-keyed evaluation cache during topological sort works.
Raises:
| Type | Description |
|---|---|
TypeError
|
if any element of |
Source code in packages/pipekit-ops/src/pipekit_ops/_base/graph.py
Composition convenience¶
pipekit_ops.compose
¶
Group B — composition convenience.
Small free functions (not Operator subclasses) that mirror the
toolz composition surface. They cost almost nothing to implement
and signal the lineage clearly.
pipe(value, *operators)— toolz.pipe; sugar forSequential(ops)(value).compose(*operators)— toolz.compose; right-to-left composition.compose_left(*operators)— toolz.compose_left; explicitSequentialalias.complement(predicate)— toolz.complement; negate a callable predicate.juxt(*operators)— toolz.juxt; tuple-returning multi-output.
juxt is the tuple-keyed sibling of Fanout (dict-keyed, Group F).
See master plan Report 2, Group B.
complement(predicate)
¶
Return a predicate that negates predicate.
Useful for inverting a predicate without lambda noise::
is_clean = lambda gt: not has_nans(gt)
# equivalent:
is_clean = complement(has_nans)
branch = Branch(predicate=is_clean, if_true=Identity(), if_false=Clean())
Source code in packages/pipekit-ops/src/pipekit_ops/compose.py
compose(*operators)
¶
Right-to-left composition (mathematical order).
compose(f, g, h) is equivalent to Sequential([h, g, f]).
The rightmost operator runs first, matching the way function
composition is written in math: f ∘ g ∘ h.
Example::
pipe = compose(Scale(2.0), AddConst(1.0)) # adds 1, then scales
assert pipe(3.0) == 8.0
Source code in packages/pipekit-ops/src/pipekit_ops/compose.py
compose_left(*operators)
¶
Left-to-right composition (pipeline order).
Explicit alias of Sequential([...]) for users who prefer the
free-function style. compose_left(f, g, h) is exactly
Sequential([f, g, h]).
Source code in packages/pipekit-ops/src/pipekit_ops/compose.py
juxt(*operators)
¶
Return a callable that applies each operator to its input and tuples the results.
juxt(f, g, h)(x) returns (f(x), g(x), h(x)). The dict-keyed
analogue is Fanout (Group F); use juxt when you want a positional
tuple without naming the outputs.
Example::
triple = juxt(Identity(), Scale(2.0), Scale(3.0))
assert triple(5.0) == (5.0, 10.0, 15.0)
Source code in packages/pipekit-ops/src/pipekit_ops/compose.py
pipe(value, *operators)
¶
Apply operators left-to-right to value.
Equivalent to Sequential(operators)(value) but with less
ceremony when you have a value in hand and want to thread it
through a sequence of operators inline.
Example::
result = pipe(gt, Scale(2.0), AddConst(1.0))
# same as: Sequential([Scale(2.0), AddConst(1.0)])(gt)
Source code in packages/pipekit-ops/src/pipekit_ops/compose.py
Building blocks¶
pipekit_ops.blocks
¶
Group C — building blocks.
Tiny operators that on their own look trivial but in combination unlock common patterns:
| Operator | Purpose |
|---|---|
Identity |
Explicit no-op; the default if_false for Branch. |
Const(value) |
Return a fixed value, ignoring input. |
Lambda(fn, name=...) |
Wrap a callable as an Operator. The escape hatch. |
Sink(write_fn, name=...) |
Side-effect write; returns input unchanged. |
Lambda and Sink set forbid_in_yaml = True — their state is a
user-supplied callable that doesn't round-trip through JSON / YAML.
They're the documented escape hatches; the rest of the operator
catalogue should round-trip cleanly.
See master plan Report 2, Group C.
Const
¶
Bases: Operator
Return a fixed value, ignoring any input.
Useful for test fixtures, Switch default cases that should emit a
sentinel, and golden values used in Try fallbacks.
Example::
zero = Const(0.0)
assert zero() == 0.0
assert zero("anything") == 0.0
The value should be JSON-primitive for round-trip via from_state.
Source code in packages/pipekit-ops/src/pipekit_ops/blocks.py
Identity
¶
Bases: Operator
Explicit no-op. Returns its input unchanged.
Used as the default value for slots that need an Operator but
where "do nothing" is the right behaviour — Branch.if_false,
Switch.default, training-mode-off augmentation, etc.
Example::
Branch(predicate=is_clean, if_true=Identity(), if_false=Despeckle())
Source code in packages/pipekit-ops/src/pipekit_ops/blocks.py
Lambda
¶
Bases: Operator
Wrap an arbitrary callable as an Operator.
The escape hatch for one-off operations that don't warrant a named
subclass. Carries forbid_in_yaml = True because the wrapped
callable typically captures closure state that doesn't survive
JSON / YAML round-trip.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
Callable[..., Any]
|
The callable to wrap. Receives the carrier and any extra
positional / keyword arguments forwarded through |
required |
name
|
str | None
|
Optional name for |
None
|
Example::
twice = Lambda(lambda x: x * 2, name="double")
assert twice(3) == 6
Source code in packages/pipekit-ops/src/pipekit_ops/blocks.py
Sink
¶
Bases: Operator
Apply a side-effecting write and pass the input through unchanged.
The canonical checkpointing primitive: log, persist, broadcast, etc.
The return value of write_fn is discarded; the original input
flows on.
Sink is not marked _terminal because it returns the
carrier — a Sequential can continue past it. For operators that
legitimately return None (write-to-disk, render-and-exit),
define a subclass with _terminal = True and put it at the end.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
write_fn
|
Callable[..., Any]
|
Callable that consumes the carrier for side effects. Its return value is ignored. |
required |
name
|
str | None
|
Optional name for |
None
|
Raises:
| Type | Description |
|---|---|
TypeError
|
If the carrier is a JAX tracer, for example inside
|
Example::
log = Sink(lambda gt: print(gt.shape), name="log_shape")
pipe = Scale(2.0) | log | AddConst(1.0)
# log fires between Scale and AddConst; the carrier is untouched
Source code in packages/pipekit-ops/src/pipekit_ops/blocks.py
Control flow¶
pipekit_ops.control
¶
Group D — control flow.
Conditionals and exception handling as first-class composable
operators. The whole point of this module: these are not Python
if / try statements outside the pipeline — they're operators
inside it. That makes them serializable, profilable, and observable
through the same Tap / Snapshot / ShapeTrace machinery as any
other step.
Branch(predicate, if_true, if_false=Identity())— binary conditional.Switch(key, cases, default=Identity())— multi-way dispatch onkey(x).Try(primary, fallback, on)— direct map fromtoolz.excepts.Coalesce(sources, is_ok)— first source whose output passesis_okwins.Retry(op, attempts, base_delay, on)— exponential-backoff retry loop.
Branch, Switch, and Coalesce carry forbid_in_yaml = True —
they hold user callables (predicates / key functions) that don't JSON /
YAML round-trip and are a pickling hazard. Try and Retry hold only
nested operators and exception types (both picklable), so they are
not flagged; their nested-operator configs are debug payloads either
way and from_state rejects them cleanly.
See master plan Report 2, Group D.
Branch
¶
Bases: Operator
Binary conditional: run if_true or if_false based on a predicate.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
predicate
|
Callable[..., bool]
|
Callable |
required |
if_true
|
Operator
|
|
required |
if_false
|
Operator | None
|
|
None
|
Example::
clean_if_dirty = Branch(
predicate=has_nans,
if_true=GapFill(),
if_false=Identity(),
)
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
compute_output_signature(input_signature)
¶
The signature both branches agree on (else None).
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
Coalesce
¶
Bases: Operator
Try each source in order; return the first output that satisfies is_ok.
The "first-good-result-wins" pattern. Each source runs on the same
input; the first whose output passes is_ok is returned. If all
sources fail the check, Coalesce raises RuntimeError.
By default a source that raises propagates its exception
immediately — only returned values are screened. Pass on (as
with Retry) to also treat listed exception types as "not ok, try
the next source".
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sources
|
list[Operator]
|
Operators to try in order. |
required |
is_ok
|
Callable[[Any], bool]
|
Predicate |
required |
on
|
tuple[type[Exception], ...]
|
Tuple of exception types a source may raise to mean "no
result here, fall through". Exceptions not listed propagate
immediately. Default: |
()
|
Example::
pick_best = Coalesce(
sources=[ReadCloudFree(t), ReadPartialCloud(t), ReadAny(t)],
is_ok=lambda gt: gt.cloud_fraction < 0.1,
on=(IOError,),
)
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
240 241 242 243 244 245 246 247 248 249 250 251 252 253 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 316 317 318 319 320 | |
compute_output_signature(input_signature)
¶
The signature every source agrees on (else None).
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
Retry
¶
Bases: Operator
Retry an operator on transient failures with exponential backoff.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The |
required |
attempts
|
int
|
Total number of attempts (≥ 1). The first attempt is
attempt 1; subsequent retries are 2, 3, ... up to |
required |
base_delay
|
float
|
Base sleep in seconds; attempt |
required |
on
|
tuple[type[BaseException], ...]
|
Tuple of exception types to consider transient. Other exceptions propagate immediately. |
required |
Example::
sturdy_read = Retry(
op=ReadFromS3(uri),
attempts=5,
base_delay=0.5,
on=(ConnectionError, TimeoutError),
)
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 | |
compute_output_signature(input_signature)
¶
The wrapped operator's: retrying does not change the output.
Switch
¶
Bases: Operator
Multi-way dispatch on key(carrier).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
Callable[..., Hashable]
|
Callable |
required |
cases
|
dict[Hashable, Operator]
|
Mapping from key-value to |
required |
default
|
Operator | None
|
|
None
|
Example::
by_sensor = Switch(
key=lambda gt: gt.attrs["sensor"],
cases={
"sentinel-2": Sentinel2Pipeline(),
"landsat-8": Landsat8Pipeline(),
},
default=Identity(),
)
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
compute_output_signature(input_signature)
¶
The signature every case and the default agree on (else None).
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
Try
¶
Bases: Operator
Run primary; on caught exceptions run fallback.
Direct analogue of toolz.excepts. If primary(carrier) raises
one of the exception types listed in on, fallback(carrier)
runs and its result is returned. Uncaught exception types
propagate.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primary
|
Operator
|
|
required |
fallback
|
Operator
|
|
required |
on
|
tuple[type[BaseException], ...]
|
Tuple of exception types to catch. Required — no implicit
|
required |
Example::
safe_read = Try(
primary=ReadCOG(uri),
fallback=Const(empty_tile),
on=(IOError, FileNotFoundError),
)
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
compute_output_signature(input_signature)
¶
The signature primary and fallback agree on (else None).
Source code in packages/pipekit-ops/src/pipekit_ops/control.py
Flow limits¶
pipekit_ops.flow
¶
Flow limits: Timeout, RateLimit and Until.
Timeout bounds how long one call may take, RateLimit bounds how often
calls may start, and Until repeats an operator up to a fixed number of
times. All three are sync wrappers: a child whose _apply is an
async def is refused at construction (decision O15), as Retry does.
Design: packages/pipekit-ops/docs/design/api/operators.md §3.
RateLimit
¶
Bases: Operator
At most calls starts of op in any window of period seconds.
{starts in [t, t + T)} ≤ n for every t, with n = calls, T = period.¶
Algorithm
- Under a lock, read the clock t and the oldest of the last n start times, t₀.
- If fewer than n calls have started, or t - t₀ ≥ T, record t as
a start, release the lock and call
op(x). - Otherwise release the lock, sleep t₀ + T - t, and go to step 1.
The limiter is shared by every thread in the process; the sleep happens outside the lock. It does not cross processes: on a pool of p processes each worker has its own limiter, so the service sees up to p n starts per T. A pickled or reloaded copy starts with no history.
Placement matters: Retry(RateLimit(read)) makes every attempt wait
for a slot, while RateLimit(Retry(read)) lets one slot cover all of
them.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The operator to limit. |
required |
calls
|
int
|
The number of starts n ≥ 1 allowed per window. |
required |
period
|
float
|
The window T > 0, in seconds. |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
|
ValueError
|
|
Examples:
Within the limit, calls pass straight through:
>>> from pipekit_ops import Identity
>>> limited = RateLimit(Identity(), calls=2, period=60.0)
>>> [limited(1), limited(2)]
[1, 2]
A pickled copy forgets the history, so it may start at once:
Inside Retry, every attempt waits for its own slot:
>>> from pipekit_ops import Retry
>>> read = RateLimit(Identity(), calls=5, period=1.0)
>>> Retry(read, attempts=2, base_delay=0.0, on=(OSError,))(4)
4
Source code in packages/pipekit-ops/src/pipekit_ops/flow.py
217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 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 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 | |
compute_output_signature(input_signature)
¶
The wrapped operator's: a rate limit does not change the output.
Timeout
¶
Bases: Operator
Raise TimeoutError if op has not returned after seconds.
y = op(x) if op returns within s seconds, else TimeoutError.
Algorithm
- Start a daemon thread running
op(x)in a copy of the caller's context, soexecutor_scopestays visible. - Wait at most
secondsfor it to finish. - If it finished, return its result or re-raise its error;
otherwise abandon the thread and raise
TimeoutError.
The thread is abandoned, not killed: Python cannot stop a thread
safely, so it runs until op returns and its result is dropped.
That suits the case this exists for, a network read with no timeout of
its own. When more than max_abandoned abandoned threads are still
running in the process, each further timeout logs a warning on
pipekit_ops.flow. A process-based timeout is a later phase.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The operator to bound. |
required |
seconds
|
float
|
The wait s > 0, in seconds. |
required |
max_abandoned
|
int
|
Abandoned threads tolerated, process-wide, before each timeout logs a warning. |
8
|
Raises:
| Type | Description |
|---|---|
TypeError
|
|
ValueError
|
|
Examples:
A call that returns in time passes through:
A call that hangs raises the builtin TimeoutError:
>>> import time
>>> from pipekit_ops import Lambda
>>> hangs = Lambda(lambda x: time.sleep(0.2))
>>> Timeout(hangs, seconds=0.01)(1)
Traceback (most recent call last):
...
TimeoutError: Timeout: Lambda did not return within 0.01 s
Retry a read that sometimes hangs:
>>> from pipekit_ops import Retry
>>> sturdy = Retry(
... Timeout(Identity(), seconds=1.0),
... attempts=3,
... base_delay=0.0,
... on=(TimeoutError,),
... )
>>> sturdy(5)
5
Source code in packages/pipekit-ops/src/pipekit_ops/flow.py
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 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 188 189 190 191 192 193 194 195 196 197 198 199 200 | |
compute_output_signature(input_signature)
¶
The wrapped operator's: a time limit does not change the output.
Until
¶
Bases: Operator
Apply op until pred holds, at most max_iter times.
x₀ = x, xᵢ₊₁ = f(xᵢ); returns x_k for the least k with pred(x_k)
truthy, k ≤ m, where f is op and m is max_iter.
Algorithm
- Set x₀ = x and i = 0; if pred(xᵢ) is truthy, return xᵢ.
- If i = m, raise
RuntimeError(on_max="raise") or return x_m (on_max="return"). - Set xᵢ₊₁ = f(xᵢ), i ← i + 1, and go to step 1.
The stateless sibling of pipekit_cycle.Recurrence, which threads a
carry state. max_iter is required, so a pipeline cannot loop
forever.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The step f. |
required |
pred
|
Operator
|
The stopping test, an |
required |
max_iter
|
int
|
The most calls of f allowed, m ≥ 0. |
required |
on_max
|
str
|
|
'raise'
|
Raises:
| Type | Description |
|---|---|
TypeError
|
|
ValueError
|
|
RuntimeError
|
At call time, under |
Examples:
Double until past 10:
>>> from pipekit_ops import Lambda
>>> double = Lambda(lambda x: 2 * x)
>>> past_ten = Lambda(lambda x: x > 10)
>>> Until(double, past_ten, max_iter=10)(1)
16
The input is checked first, so f may run zero times:
Return x_m instead of raising when the budget runs out:
>>> Until(double, past_ten, max_iter=2, on_max="return")(1)
4
>>> Until(double, past_ten, max_iter=2)(1)
Traceback (most recent call last):
...
RuntimeError: Until: pred still false after max_iter=2 calls of Lambda
Source code in packages/pipekit-ops/src/pipekit_ops/flow.py
342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 | |
compute_output_signature(input_signature)
¶
require_sync(op, owner, field)
¶
Raise TypeError if op has an async def _apply (decision O15).
A sync wrapper calling such a child gets an un-awaited coroutine back, so nothing runs; refusing at construction makes that failure loud.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The child operator. |
required |
owner
|
str
|
The wrapping class's name, for the message. |
required |
field
|
str
|
The constructor argument that holds |
required |
Source code in packages/pipekit-ops/src/pipekit_ops/flow.py
Observers¶
pipekit_ops.observe
¶
Group E — observers.
Identity-with-side-effect operators. The carrier flows through unchanged; something useful happens on the side. Critical for notebook exploration and debug, useful in production logging.
Tap(fn, name=...)— fire a callback, return input unchanged. Distinct fromSink:Tapis named for observation (logging, metrics);Sinkfor persistence. Same machinery, different intent.Snapshot()— controller.Snapshot.at(key)returns an operator that captures the input underkey. Inspect viasnapshot.captures(dict).ShapeTrace(printer=print, mode="every"|"diff_only")— log shape / dtype at every step. Carrier-agnostic viagetattr.Profile()— controller.Profile.wrap(op)returns a wrapped operator that records timings. Inspect viaprofile.report().Histogram(bins=10, to_array=...)— controller.Histogram.at(key)returns an operator that captures a distribution;to_arrayis a user-supplied callable adapting the carrier to a flat sequence of numbers (default:list).Log(message, level, logger)— write oneloggingrecord per call. Holds only strings, so unlikeTapit round-trips.
Snapshot, Profile, and Histogram use the controller-not-operator
pattern: the class itself isn't an Operator, but its .at() /
.wrap() methods return small operators that close over the
controller's state. This keeps the per-step operators tiny while the
collected data lives on a single object the user can inspect.
Under a JAX transformation these run at trace time. Snapshot.at raises,
because a stored tracer is invalid once the trace ends. Tap and Profile.wrap
warn once per instance, because a trace-time callback can be deliberate but a
trace-time timing never measures the compiled function; Log warns once too.
ShapeTrace is meant to run at trace time, so it stays silent.
See master plan Report 2, Group E.
Histogram
¶
Controller that captures simple histograms by key.
Not an Operator itself. Histogram.at(key) returns an
operator that, on each call, runs to_array(carrier) and bins
the result. Bin counts accumulate across calls.
The default to_array (list) handles any iterable of numbers
— including a plain list[float]. Pass a carrier-specific
adapter for numpy / xarray (e.g. to_array=lambda a: a.ravel().tolist()).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bins
|
int
|
Number of equal-width bins. |
10
|
to_array
|
Callable[[Any], Iterable[float]]
|
Adapter |
list
|
Attributes:
| Name | Type | Description |
|---|---|---|
results |
dict[str, dict[str, list[float]]]
|
Dict |
Example::
h = Histogram(bins=10)
pipe = h.at("ndvi_pre") | Clean() | h.at("ndvi_post")
for tile in tiles:
pipe(tile)
h.results["ndvi_post"]
Source code in packages/pipekit-ops/src/pipekit_ops/observe.py
Log
¶
Bases: Operator
Pass the carrier through and write one logging record per call.
The message is formatted with {n} (calls of this Log so far in
this process, from 1), {type} (the input's class name) and
{signature} (the input's Signature through its registered
adapter, else ?). Handlers, filters and formatters are left to
logging. Unlike Tap, Log holds only strings, so it round-trips.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
A |
'call {n}: {type} {signature}'
|
level
|
str
|
A |
'INFO'
|
logger
|
str
|
The name of the logger the record goes to. |
'pipekit_ops'
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Examples:
The input comes back as the same object, and the record goes to
logging:
>>> import logging
>>> records = []
>>> handler = logging.Handler()
>>> handler.emit = records.append
>>> logging.getLogger("demo").addHandler(handler)
>>> note = Log("call {n}: {type} {signature}", level="WARNING", logger="demo")
>>> x = [1.0, 2.0]
>>> note(x) is x
True
>>> records[-1].getMessage()
'call 1: list ?'
It composes like any step, and its config is plain strings:
>>> from pipekit_ops import Identity
>>> (Identity() | note)(3.0)
3.0
>>> records[-1].getMessage()
'call 2: float ?'
>>> note.get_config()["level"], note.get_config()["logger"]
('WARNING', 'demo')
An unknown field is rejected at construction:
>>> Log("{shape}")
Traceback (most recent call last):
...
ValueError: Log.message may name only n, type, signature; got 'shape'.
Source code in packages/pipekit-ops/src/pipekit_ops/observe.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 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 191 192 193 194 195 196 197 | |
Profile
¶
Controller that records per-step wall-clock timings.
Not an Operator itself. Profile.wrap(op) returns a wrapped
operator that times each call and appends the elapsed seconds to
timings[type(op).__name__].
Attributes:
| Name | Type | Description |
|---|---|---|
timings |
dict[str, list[float]]
|
Dict |
Example::
prof = Profile()
pipe = prof.wrap(Scale(2.0)) | prof.wrap(AddConst(1.0))
for v in [1.0, 2.0, 3.0]:
pipe(v)
prof.report()
# {"Scale": {"calls": 3, "total": ..., "mean": ...}, ...}
Source code in packages/pipekit-ops/src/pipekit_ops/observe.py
report()
¶
Return a {name: {calls, total, mean}} summary.
Source code in packages/pipekit-ops/src/pipekit_ops/observe.py
ShapeTrace
¶
Bases: Operator
Log shape / dtype of each carrier passing through.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
printer
|
Callable[[str], None]
|
Callable invoked with the formatted line. Defaults to
the built-in |
print
|
mode
|
str
|
|
'every'
|
label
|
str | None
|
Optional prefix for the printed line (useful when
multiple |
None
|
extra
|
Callable[[Any], str | None] | None
|
Optional callable |
None
|
Example::
trace = ShapeTrace(mode="diff_only", label="post-resample")
pipe = Resample() | trace | NDVI()
# print the coordinate system too
ShapeTrace(extra=lambda gt: f"crs={gt.crs}")
Source code in packages/pipekit-ops/src/pipekit_ops/observe.py
Snapshot
¶
Controller that captures intermediate carriers by name.
Not an Operator itself. Call .at(key) to obtain a per-step
operator that stores the carrier under that key.
Attributes:
| Name | Type | Description |
|---|---|---|
captures |
dict[str, Any]
|
Dict |
Example::
snap = Snapshot()
pipe = snap.at("pre") | Scale(2.0) | snap.at("post") | AddConst(1.0)
pipe(3.0)
snap.captures
# {"after_scale": 3.0, "after_add": 6.0}
Source code in packages/pipekit-ops/src/pipekit_ops/observe.py
Tap
¶
Bases: Operator
Fire a callback on the carrier; return the carrier unchanged.
Distinct from Sink only by naming convention — Tap is for
observation (logging, metrics, breakpoint hooks); Sink is for
persistence. Both pass the carrier through; both carry
forbid_in_yaml = True.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
Callable[[Any], Any]
|
Callable invoked with the carrier. Its return value is ignored. |
required |
name
|
str | None
|
Optional name for |
None
|
Example::
log = Tap(lambda x: print(getattr(x, "shape", x)), name="log_shape")
pipe = Scale(2.0) | log | AddConst(1.0)
Source code in packages/pipekit-ops/src/pipekit_ops/observe.py
Combination¶
pipekit_ops.combine
¶
Group F — combination.
One-input → many-outputs operators. Fanout is dict-keyed; the
tuple-keyed sibling is juxt (in compose.py).
Fanout is intentionally minimal — it's sugar over a single-input
Graph with multiple named outputs. Carrier-specific merging
(Augment, ApplyToEach, anything that uses xr.merge) stays in
sister libraries (xr_toolz) because it isn't carrier-agnostic.
See master plan Report 2, Group F.
Fanout
¶
Bases: Operator
Apply each named operator to the same input; return a dict of outputs.
The dict-keyed companion to juxt. Each operator runs on the same
carrier; the results are collected under their keys.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
branches
|
dict[str, Operator]
|
Mapping from output name to |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
if any branch isn't an |
ValueError
|
if |
Example::
rgb = Fanout({
"red": SelectBand(0),
"green": SelectBand(1),
"blue": SelectBand(2),
})
out = rgb(image)
# {"red": ..., "green": ..., "blue": ...}
Source code in packages/pipekit-ops/src/pipekit_ops/combine.py
compute_output_signature(input_signature)
¶
Dict plumbing¶
pipekit_ops.dicts
¶
Dict plumbing — reshape the dicts that Graph, Fanout and STAC items carry.
Every reshape that used to be a Lambda (and so stopped a pipeline from
round-tripping) is a small operator here, named after its toolz function:
| Operator | Semantics |
|---|---|
Get(key, default) |
x[key]; a tuple of keys returns a tuple |
GetIn(keys, default) |
x[k₀][k₁]… |
Pick(keys) |
keep only keys, in the order given |
Dissoc(keys) |
drop keys; missing keys are ignored |
Rename(mapping) |
rename keys; unlisted keys pass through |
Merge() |
union of several dicts, later inputs win |
Assoc(key, op) |
add key: op(x) |
UpdateIn(keys, op, default) |
replace the value at a key path by op(value) |
ValMap(op) |
op on every value |
Keys are plain values (a str, an int or a tuple of them) and functions
are operators, so every operator here has a faithful config. A bare Python
callable is refused with a TypeError that points to Lambda. No operator
mutates its input: each dict it returns is a new dict, whose values are
shared with the input, not copied.
See the pipekit-ops design, api/operators.md section 1.
Assoc
¶
Bases: _ContainerOperator
Add a key whose value is computed from the whole dict (toolz assoc).
x ↦ {**x, key: op(x)}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
Key
|
The key to set, a |
required |
op
|
Operator
|
The |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import Assoc, Get
>>> Assoc("tile", Get("t"))({"t": "31TCJ"})
{'t': '31TCJ', 'tile': '31TCJ'}
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
Dissoc
¶
Bases: _ContainerOperator
Drop some keys; missing keys are ignored (toolz dissoc).
x ↦ {k: v for k, v in x if k ∉ keys}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
Sequence[Key]
|
The keys to drop, a tuple of |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import Dissoc
>>> Dissoc(("scenes",))({"n_kept": 3, "scenes": [0, 1, 2]})
{'n_kept': 3}
>>> Dissoc(("missing",))({"a": 1})
{'a': 1}
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
Get
¶
Bases: _ContainerOperator
Index into a mapping or sequence (toolz get).
x ↦ x[k], or x ↦ (x[k₀], x[k₁], …) for a tuple of keys.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
Key | tuple[Key, ...]
|
A |
required |
default
|
Any
|
Returned in place of a missing item (a |
_MISSING
|
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import Get
>>> Get("id")({"id": "S2A_31TCJ", "cloud": 12.0})
'S2A_31TCJ'
>>> Get(("a", "z"), default=None)({"a": 1})
(1, None)
>>> Get(-1)([10, 20, 30])
30
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
GetIn
¶
Bases: _ContainerOperator
Walk a key path through nested containers (toolz get_in).
x ↦ x[k₀][k₁]…[kₘ₋₁]; the empty path returns x.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
Sequence[Key]
|
The path, a tuple of |
required |
default
|
Any
|
Returned when the path is missing (a |
_MISSING
|
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import GetIn
>>> item = {"properties": {"eo:cloud_cover": 12.5}}
>>> GetIn(("properties", "eo:cloud_cover"))(item)
12.5
>>> GetIn(("properties", "s2:mgrs_tile"), default="?")(item)
'?'
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
Merge
¶
Bases: _ContainerOperator
Union of several mappings; later inputs win (toolz merge).
(x₁, …, xₘ) ↦ {k: xⱼ[k]}, over every key k of any input, with j the last input that holds k.
Takes the mappings as positional arguments, so it works as a Graph
node, or as one sequence of mappings, so it works after juxt.
Examples:
>>> from pipekit_ops import Merge
>>> Merge()({"a": 1, "b": 1}, {"b": 2})
{'a': 1, 'b': 2}
>>> Merge()([{"a": 1}, {"c": 3}])
{'a': 1, 'c': 3}
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
Pick
¶
Bases: _ContainerOperator
Keep only some keys, in the order given.
x ↦ {k: x[k] for k in keys}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
Sequence[Key]
|
The keys to keep, a tuple of |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
KeyError
|
at call time, if a key is missing from the input. |
Examples:
>>> from pipekit_ops import Pick
>>> Pick(("b", "a"))({"a": 1, "b": 2, "c": 3})
{'b': 2, 'a': 1}
>>> Pick(("z",))({"a": 1})
Traceback (most recent call last):
...
KeyError: 'z'
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
Rename
¶
Bases: _ContainerOperator
Rename keys; unlisted keys pass through and the order is kept.
x ↦ {m(k): v for k, v in x}, with m(k) = mapping[k] if k is listed, else k.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mapping
|
Mapping[str, Key]
|
Old key ↦ new key. Old keys are |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if a key of |
ValueError
|
at construction, if two old keys share a new key; at call time, if a new key collides with a key the input keeps. |
Examples:
>>> from pipekit_ops import Rename
>>> Rename({"n_found": "found"})({"n_found": 300, "n_kept": 212})
{'found': 300, 'n_kept': 212}
>>> Rename({"a": "b"})({"a": 1, "b": 2})
Traceback (most recent call last):
...
ValueError: Rename: 'a' -> 'b' collides with the kept key 'b'.
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
UpdateIn
¶
Bases: _ContainerOperator
Replace the value at a key path by op(value) (toolz update_in).
x ↦ a copy of x with x[k₀]…[kₘ₋₁] replaced by op(x[k₀]…[kₘ₋₁]).
Every dict along the path is copied; nothing else is.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
Sequence[Key]
|
The path, a non-empty tuple of |
required |
op
|
Operator
|
The |
required |
default
|
Any
|
Stands in for a missing value: |
_MISSING
|
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
ValueError
|
at construction, if |
Examples:
>>> from pipekit_ops import Const, UpdateIn
>>> zero_cloud = UpdateIn(("properties", "cloud"), Const(0.0))
>>> zero_cloud({"properties": {"cloud": 9}})
{'properties': {'cloud': 0.0}}
>>> UpdateIn(("a", "b"), Const(1), default=None)({})
{'a': {'b': 1}}
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
ValMap
¶
Bases: _ContainerOperator
Apply an operator to every value, keeping the keys (toolz valmap).
x ↦ {k: op(v) for k, v in x}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import Count, ValMap
>>> ValMap(Count())({"31TCJ": [0, 1], "31TDJ": [2]})
{'31TCJ': 2, '31TDJ': 1}
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
key_path(value, owner, field)
¶
Check a key path (a tuple of str / int); a list becomes a tuple.
Raises:
| Type | Description |
|---|---|
TypeError
|
if |
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
lookup(key, item)
¶
key(item) for an operator, item[key], or a tuple for a tuple of keys.
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
plain_key(value, owner, field)
¶
Check a plain key (a str, an int or a tuple of them); lists become tuples.
A key reloads from JSON as a list, so lists are accepted and turned back into tuples: the rebuilt operator then picks several keys again.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
Any
|
The candidate key. |
required |
owner
|
str
|
The validating class's name (for the message). |
required |
field
|
str
|
The parameter name. |
required |
Returns:
| Type | Description |
|---|---|
Key | tuple[Key, ...]
|
The key, with a list turned into a tuple. |
Raises:
| Type | Description |
|---|---|
TypeError
|
if |
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
require_child(value, owner, field)
¶
Raise TypeError unless value is an Operator; name Lambda if callable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
Any
|
The candidate predicate, key or function. |
required |
owner
|
str
|
The validating class's name (for the message). |
required |
field
|
str
|
The parameter name. |
required |
Source code in packages/pipekit-ops/src/pipekit_ops/dicts.py
Iterables¶
pipekit_ops.seq
¶
Iterables — filter, fold, group and count lists of items inside a pipeline.
| Operator | Output |
|---|---|
Not(pred) |
not pred(x) |
Filter(pred) |
items with pred(item) truthy |
Remove(pred) |
items with pred(item) falsy |
Reduce(op, initial) |
left fold, op((acc, item)) |
PartitionAll(n) |
tuples of n consecutive items |
Concat() |
one level flattened |
Zip(strict=True) |
tuples of aligned items, strict by default |
GroupBy(key) |
dict[group, list] |
CountBy(key) |
dict[group, int] |
Count() |
the number of items |
Unique(key=None) |
the first occurrence of each item or key |
Take(n) |
the first n items |
SortBy(key, reverse) |
a stable sort |
Every operator accepts any iterable and returns a list (a dict for the
group-by family; Not returns a bool), so an output can be inspected,
cached and fed to two Graph nodes. Predicates and folds are operators;
a key is an operator or a plain key (a str, an int or a tuple of
them, looked up as item[key] the way toolz's getter does). A bare
Python callable is refused with a TypeError that points to Lambda.
See the pipekit-ops design, api/operators.md section 2.
Concat
¶
Bases: _ContainerOperator
Flatten exactly one level (toolz concat).
(s₀, s₁, …) ↦ s₀ ⧺ s₁ ⧺ …
Examples:
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Count
¶
Bases: _ContainerOperator
The number of items (toolz count).
(x₀, …, xₙ₋₁) ↦ n; len(x) when defined, else the iterable is consumed.
Examples:
>>> from pipekit_ops import Count
>>> Count()([1, 2, 3])
3
>>> Count()(i for i in range(5) if i % 2)
2
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
CountBy
¶
Bases: _ContainerOperator
Count items per key, in first-seen order (toolz countby).
(x₀, …, xₙ₋₁) ↦ {g: #{i | key(xᵢ) = g}}. CountBy(Identity()) is
toolz frequencies.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
KeyLike
|
An |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import CountBy, Identity
>>> CountBy("tile")([{"tile": "A"}, {"tile": "B"}, {"tile": "A"}])
{'A': 2, 'B': 1}
>>> CountBy(Identity())("abca")
{'a': 2, 'b': 1, 'c': 1}
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Filter
¶
Bases: _ContainerOperator
Keep the items a predicate accepts (builtin filter).
(x₀, …, xₙ₋₁) ↦ [xᵢ | pred(xᵢ) truthy], in input order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pred
|
Operator
|
The predicate |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import Filter, Get
>>> Filter(Get("clear"))([{"clear": True}, {"clear": False}])
[{'clear': True}]
>>> Filter(lambda x: x)
Traceback (most recent call last):
...
TypeError: Filter.pred must be an Operator, got function. Wrap a plain
function in `Lambda(fn)` ...
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
GroupBy
¶
Bases: _ContainerOperator
Group items by a key, in first-seen order (toolz groupby).
(x₀, …, xₙ₋₁) ↦ {g: [xᵢ | key(xᵢ) = g]}, each list in input order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
KeyLike
|
An |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import GroupBy
>>> GroupBy(0)(["ab", "b", "ac"])
{'a': ['ab', 'ac'], 'b': ['b']}
>>> items = [{"tile": "A", "i": 0}, {"tile": "B", "i": 1}]
>>> GroupBy(("tile", "i"))(items)
{('A', 0): [{'tile': 'A', 'i': 0}], ('B', 1): [{'tile': 'B', 'i': 1}]}
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Not
¶
Bases: _ContainerOperator
Negate a predicate; the operator form of complement.
x ↦ ¬ pred(x).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pred
|
Operator
|
The predicate |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
PartitionAll
¶
Bases: _ContainerOperator
Split into tuples of n consecutive items (toolz partition_all).
(x₀, …, xₘ₋₁) ↦ [(x₀ … xₙ₋₁), (xₙ … x₂ₙ₋₁), …]; the last may be shorter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n
|
int
|
The tuple length, |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
ValueError
|
at construction, if |
Examples:
>>> from pipekit_ops import PartitionAll
>>> PartitionAll(2)([1, 2, 3, 4, 5])
[(1, 2), (3, 4), (5,)]
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Reduce
¶
Bases: _ContainerOperator
Left fold (functools.reduce), with the operator called on one pair.
a₀ = initial, aᵢ₊₁ = op((aᵢ, xᵢ)); returns aₙ. Without initial,
a₁ = x₀.
op takes one (acc, item) tuple because an operator takes one
carrier: an operator written for Reduce reads acc, item = pair.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The fold |
required |
initial
|
Any
|
The starting accumulator. Omit it to start from the first item. |
_MISSING
|
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import Get, Reduce
>>> Reduce(Get(1))([3, 4, 5])
5
>>> Reduce(Get(0), initial="start")([3, 4, 5])
'start'
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Remove
¶
Bases: _ContainerOperator
Drop the items a predicate accepts (toolz remove).
(x₀, …, xₙ₋₁) ↦ [xᵢ | pred(xᵢ) falsy], in input order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pred
|
Operator
|
The predicate |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import Get, Remove
>>> Remove(Get("cloudy"))([{"cloudy": True}, {"cloudy": False}])
[{'cloudy': False}]
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
SortBy
¶
Bases: _ContainerOperator
A stable sort by key (builtin sorted).
(x₀, …, xₙ₋₁) ↦ the items ordered by key(xᵢ); equal keys keep input order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
KeyLike
|
An |
required |
reverse
|
bool
|
Sort in descending order (still stable). |
False
|
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import SortBy, Take
>>> SortBy("cloud")([{"cloud": 30}, {"cloud": 5}])
[{'cloud': 5}, {'cloud': 30}]
>>> (SortBy(0, reverse=True) | Take(1))([(1, "a"), (3, "b"), (2, "c")])
[(3, 'b')]
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Take
¶
Bases: _ContainerOperator
The first n items (toolz take), for dry runs.
(x₀, …, xₘ₋₁) ↦ [x₀, …, x_min(n, m)₋₁]; a lazy input is read no further.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n
|
int
|
How many items, |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
ValueError
|
at construction, if |
Examples:
>>> from pipekit_ops import Take
>>> Take(2)([5, 6, 7])
[5, 6]
>>> Take(3)(iter(range(10**9)))
[0, 1, 2]
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Unique
¶
Bases: _ContainerOperator
The first occurrence of each item, or of each key (toolz unique).
(x₀, …, xₙ₋₁) ↦ [xᵢ | key(xᵢ) ∉ {key(xⱼ) | j < i}]; items or keys must be hashable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
KeyLike | None
|
|
None
|
Raises:
| Type | Description |
|---|---|
TypeError
|
at construction, if |
Examples:
>>> from pipekit_ops import Unique
>>> Unique()([3, 1, 3, 2, 1])
[3, 1, 2]
>>> Unique("tile")([{"tile": "A", "i": 0}, {"tile": "A", "i": 1}])
[{'tile': 'A', 'i': 0}]
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Zip
¶
Bases: _ContainerOperator
Align several iterables into tuples (builtin zip), strict by default.
(s₁, …, sₘ) ↦ [(s₁[i], …, sₘ[i]) for i < n].
Takes the iterables as positional arguments, so it works as a Graph
node, or as one sequence of iterables, so it works after juxt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
strict
|
bool
|
Raise |
True
|
Examples:
>>> from pipekit_ops import Zip
>>> Zip()(["a", "b"], [1, 2])
[('a', 1), ('b', 2)]
>>> Zip(strict=False)([1, 2], [1])
[(1, 1)]
>>> Zip()([1, 2], [1])
Traceback (most recent call last):
...
ValueError: zip() argument 2 is shorter than argument 1
Source code in packages/pipekit-ops/src/pipekit_ops/seq.py
Caching¶
pipekit_ops.cache
¶
Group G — caching.
Hash-keyed memoisation as a composable operator, in memory by default
or in a named Store (pipekit_ops.store), such as a DirectoryStore
that survives a restart.
Cache(inner, store=None, key=None)— wraps any operator; caches outputs keyed by a hash of the input (or ofkey(input)) and ofpipeline_hash(inner). Exposeshits/missescounters for cache-effectiveness reporting.Memoize— alias ofCache, named aftertoolz.memoize.
Hash key shape: sha256(input key ‖ 0x00 ‖ pipeline_hash(inner)), with
‖ 0x00 ‖ pipeline_hash(key) appended when a key operator is set.
pipeline_hash records the inner operator's module and class, so two
classes with equal configs never share an entry.
Array-likes (anything exposing tobytes / shape / dtype)
are keyed by their full byte content — plain repr would be
truncated by NumPy's print options, letting two different large
arrays silently collide onto one key. Other inputs fall back to
repr(x); this is deliberately conservative — Cache is opt-in
and the user should know whether their carrier is hash-friendly.
Cache is safe to share across ThreadMap workers: bookkeeping is
lock-protected. The inner operator runs outside the lock, so
concurrent misses on the same key may compute it more than once (the
usual memoisation trade-off; correctness is unaffected).
Cache raises on a JAX tracer: a tracer has no value to key on, and inside
jax.jit JAX's compilation cache already does this job.
See master plan Report 2, Group G.
Cache
¶
Bases: Operator
Memoise an inner operator by hashing its input and the inner pipeline.
With no store the entries live in a private MemoryStore owned by
this instance, as before P4. A store name is looked up in the
enclosing store_scope at call time, so the same pipeline can cache
to a scratch directory on one machine and in memory on another; the
name is in the config, the store is not.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inner
|
Operator
|
The |
required |
store
|
str | None
|
The name of a store bound by |
None
|
key
|
Operator | None
|
An |
None
|
Attributes:
| Name | Type | Description |
|---|---|---|
hits |
Number of cache hits since construction, in this process. |
|
misses |
Number of cache misses since construction, in this process. |
Raises:
| Type | Description |
|---|---|
TypeError
|
If |
LookupError
|
At call time, if no |
Examples:
>>> from pipekit_ops import Get, Identity
>>> cached = Cache(Identity())
>>> cached(1), cached(1), cached(2)
(1, 1, 2)
>>> cached.hits, cached.misses
(1, 2)
>>> by_id = Cache(Identity(), key=Get("id"))
>>> by_id({"id": "a", "cloud": 3})
{'id': 'a', 'cloud': 3}
>>> by_id({"id": "a", "cloud": 9})
{'id': 'a', 'cloud': 3}
Source code in packages/pipekit-ops/src/pipekit_ops/cache.py
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 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 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 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 | |
clear()
¶
Drop the private in-memory entries (counters preserved).
A named store may be shared with other caches and processes, so it is left alone; clear it through the store itself.
Source code in packages/pipekit-ops/src/pipekit_ops/cache.py
compute_output_signature(input_signature)
¶
The inner operator's: caching does not change the output.
Store¶
pipekit_ops.store
¶
Stores: where a Cache keeps its entries.
A Store is three methods over string keys: key in store,
store.load(key) and store.save(key, value). Cache keys are
64-character hex digests, so a store never has to sanitise them.
MemoryStore— a dict and a lock; threads in one process.DirectoryStore— one pickle file per key under a root directory, written atomically, so several processes on one file system can share it.store_scope— binds store names to stores for the duration of a block.Cache(op, store="scenes")names a store; which directory it is, is a deployment choice made around the call, not part of the pipeline's config orpipeline_hash.
Carrier-specific stores (Zarr, COG, an object store) live in the sister packages and only implement the three methods.
Design: packages/pipekit-ops/docs/design/api/operators.md, section 4.
DirectoryStore
¶
One file per key under root, written atomically.
The entry for key abcd… is root/ab/abcd….pkl. A save writes a
temporary file in the same directory and renames it into place with
os.replace, so a reader never sees a partial value, and processes on
one file system can share the directory; when two write one key, the
last rename wins. root/pipekit-store.json records the format
version, and a directory with another version is refused.
Pickle loads run code: point a DirectoryStore only at a directory
you control. A DirectoryStore pickles as its root and codec, so it
can travel to a process pool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
root
|
str | PathLike[str]
|
The directory; created, with its parents, if missing. |
required |
codec
|
str
|
How values are written; |
'pickle'
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Examples:
>>> import tempfile
>>> root = tempfile.mkdtemp()
>>> store = DirectoryStore(root)
>>> key = "ab" + "0" * 62
>>> store.save(key, {"id": "S2A"})
>>> key in store, store.load(key)
(True, {'id': 'S2A'})
>>> key in DirectoryStore(root)
True
Source code in packages/pipekit-ops/src/pipekit_ops/store.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 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 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 | |
MemoryStore
¶
A dict guarded by a lock: shared by the threads of one process.
A pickled copy keeps the entries and gets its own lock, so a copy in another process is a snapshot, not a shared store.
Examples:
>>> store = MemoryStore()
>>> key = "0" * 64
>>> key in store
False
>>> store.save(key, [1, 2])
>>> store.load(key), len(store)
([1, 2], 1)
Source code in packages/pipekit-ops/src/pipekit_ops/store.py
Store
¶
Bases: Protocol
What Cache needs from a store: membership, load and save by key.
Keys are 64-character lowercase hex digests. load of a missing
key raises KeyError. A store shared between threads or processes
must make save atomic: a reader sees the old value or the new
one, never part of one.
Examples:
Source code in packages/pipekit-ops/src/pipekit_ops/store.py
bind_stores(stores)
¶
Bind stores for the rest of the current context (a process worker).
bound_stores()
¶
resolve_store(name, owner='Cache')
¶
The store bound to name by the innermost store_scope.
Raises:
| Type | Description |
|---|---|
LookupError
|
If no scope binds |
Source code in packages/pipekit-ops/src/pipekit_ops/store.py
store_scope(**stores)
¶
Bind store names to stores inside the block.
Bindings nest: an inner scope adds names to the outer ones and may
rebind a name for its block. The stores are not part of any config,
so a pipeline hashes the same on a laptop and on a cluster. Map
worker threads see the scope; process workers get the stores pickled.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**stores
|
Store
|
|
{}
|
Raises:
| Type | Description |
|---|---|
TypeError
|
If a value is not a |
Examples:
>>> from pipekit_ops import Cache, Identity
>>> cached = Cache(Identity(), store="scenes")
>>> scenes = MemoryStore()
>>> with store_scope(scenes=scenes):
... cached(1), cached(1)
(1, 1)
>>> cached.hits, cached.misses, len(scenes)
(1, 1, 1)
Source code in packages/pipekit-ops/src/pipekit_ops/store.py
Quality control¶
pipekit_ops.qc
¶
Group H — quality control.
Assertions as composable operators. Every QC check is "I bet this
won't break" turned into a permanent runtime guard. The pipekit-level
operators stay carrier-agnostic — they look at shape / dtype
via getattr, never inside the array. Numeric checks
(AssertValueRange, AssertNoNaN, etc.) that need to look at array
contents live in sister libraries (Report 3, pipekit-array).
Quarantine(check, sentinel, on_quarantine=None)— non-raising; failure → sentinel.AssertShape(expected_shape)— pass-through; raise on shape mismatch.AssertDType(expected_dtype)— pass-through; raise on dtype mismatch.AssertHasAttribute(name, value=None)— pass-through; check attribute presence (and value, if supplied).AssertCallable(predicate, message=...)— pass-through; user-supplied predicate; raise on failure.AssertThat(pred, message=...)—AssertCallablewith an operator as the predicate, so it round-trips.AssertSignature(pattern, dtype=None, missing="raise")— pass-through; check the named dims (and dtype) read through the carrier's registered adapter. Also checked at build time, bySequential.summary.
See master plan Report 2, Group H.
AssertCallable
¶
Bases: Operator
Pass-through; raise QCError if predicate(x) is falsy.
The escape hatch for QC checks that don't fit the typed asserts.
Carries forbid_in_yaml = True because the predicate is a
user callable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
predicate
|
Callable[[Any], bool]
|
Callable |
required |
message
|
str
|
Error message attached to the raised |
'predicate returned False'
|
Source code in packages/pipekit-ops/src/pipekit_ops/qc.py
AssertDType
¶
Bases: Operator
Pass-through; raise QCError if getattr(x, 'dtype') mismatches.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
expected_dtype
|
Any
|
A dtype label. Compared via |
required |
Source code in packages/pipekit-ops/src/pipekit_ops/qc.py
AssertHasAttribute
¶
Bases: Operator
Pass-through; raise QCError if attribute missing (or value mismatches).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Attribute name to look up via |
required |
value
|
Any
|
Optional expected value. Omit the argument (it
defaults to a private |
_MISSING
|
Source code in packages/pipekit-ops/src/pipekit_ops/qc.py
AssertShape
¶
Bases: Operator
Pass-through; raise QCError if getattr(x, 'shape') mismatches.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
expected_shape
|
tuple[int | None, ...]
|
Tuple of ints. |
required |
Example::
AssertShape((None, 3, 256, 256)) # batch dim is free, rest are pinned
Source code in packages/pipekit-ops/src/pipekit_ops/qc.py
AssertSignature
¶
Bases: Operator
Pass-through; raise QCError unless the carrier's dims fit pattern.
The dims are read through the carrier's registered adapter
(pipekit_ops.adapters.get_adapter), never from the carrier itself.
The same check runs at build time in compute_output_signature, so
Sequential.summary reports a mismatch before any data runs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pattern
|
str
|
Dim names with at most one |
required |
dtype
|
str | None
|
If set, the dtype label the signature must carry. |
None
|
missing
|
str
|
What to do with a carrier that has no adapter:
|
'raise'
|
Raises:
| Type | Description |
|---|---|
QCError
|
If the dims do not fit |
ValueError
|
If |
Examples:
At build time, against a Signature:
>>> from pipekit_ops import Signature
>>> cube = Signature(dims=(("time", 3), ("y", 4), ("x", 4)), dtype="f4")
>>> AssertSignature("... y x").compute_output_signature(cube) == cube
True
>>> check = AssertSignature("band y x")
>>> check.compute_output_signature(cube)
Traceback (most recent call last):
...
QCError: AssertSignature: dims (time, y, x) do not fit 'band y x'.
A value with no adapter passes only with missing="skip":
>>> AssertSignature("y x", missing="skip")([1.0])
[1.0]
>>> AssertSignature("y x")([1.0])
Traceback (most recent call last):
...
QCError: AssertSignature: no named-array adapter for list.
Source code in packages/pipekit-ops/src/pipekit_ops/qc.py
294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 | |
compute_output_signature(input_signature)
¶
The input signature, after checking it against pattern.
An unknown (None) or foreign signature passes unchecked.
Raises:
| Type | Description |
|---|---|
QCError
|
If a known |
Source code in packages/pipekit-ops/src/pipekit_ops/qc.py
AssertThat
¶
Bases: Operator
Pass-through; raise QCError if pred(x) is falsy.
AssertCallable with an operator instead of a closure, so the check
is part of the pipeline's config and hash.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pred
|
Operator
|
The operator run on the carrier; a truthy result means OK. |
required |
message
|
str
|
Error message attached to the raised |
'predicate returned False'
|
Raises:
| Type | Description |
|---|---|
QCError
|
If |
Examples:
A passing check returns the input object itself:
A failing one raises:
>>> AssertThat(Const(False), "too cloudy")(x)
Traceback (most recent call last):
...
pipekit_ops.qc.QCError: AssertThat(Const): too cloudy
The predicate composes like any operator:
>>> from pipekit_ops import Lambda
>>> is_short = Lambda(len) | Lambda(lambda n: n < 3)
>>> AssertThat(is_short)(x)
[1.0, 2.0]
Source code in packages/pipekit-ops/src/pipekit_ops/qc.py
QCError
¶
Quarantine
¶
Bases: Operator
Non-raising QC: on failure log and return sentinel; on success pass through.
The "soft" QC pattern. Useful in batched / catalog-driven pipelines where one bad input shouldn't abort the whole run — quarantine it, keep going, and surface the count in the run report.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
check
|
Callable[[Any], bool]
|
Callable |
required |
sentinel
|
Any
|
Value returned when |
required |
on_quarantine
|
Callable[[Any], None] | None
|
Optional callable |
None
|
Example::
safe = Quarantine(
check=lambda gt: gt.cloud_fraction < 0.5,
sentinel=None,
on_quarantine=lambda gt: log.warning("quarantined %s", gt.id),
)
Source code in packages/pipekit-ops/src/pipekit_ops/qc.py
Shape inference¶
pipekit_ops.signature
¶
Group I — shape inference.
A tiny value class plus a per-operator protocol method (already declared
on Operator) that together let Sequential.summary print a
Keras-style structural table without executing any data.
Signature(dims, dtype)— immutable record of named dimensions and dtype. Helpers:format,drop_dims,replace_dims;matchestests it against a dims pattern ("band y x","... y x") andfrom_patternbuilds a sizeless one from a pattern. Patterns are the one dims vocabulary of the stack: readable types,check_dimsand kernel output signatures are written in them.Operator.compute_output_signature(input_sig) -> Signature | None— per-operator method, default passthrough. Operators that reshape / reduce / retype override; carriers without named dimensions returnNoneto opt out.
Signature is xarray-flavoured (named dims). Carriers without named
dims (raw numpy, GeoTensor) opt out by returning None from
compute_output_signature; Sequential.summary reports ? for
those steps rather than crashing.
See master plan Report 2, Group I.
Signature
dataclass
¶
Immutable shape + dtype record.
Attributes:
| Name | Type | Description |
|---|---|---|
dims |
tuple[tuple[str, int | None], ...]
|
Tuple of |
dtype |
str | None
|
A dtype label (e.g. |
Example::
sig = Signature(dims=(("time", 12), ("y", 256), ("x", 256)), dtype="float32")
sig.format() # "(time=12, y=256, x=256) float32"
sig.drop_dims(("time",)) # drops the time axis
sig.replace_dims({"x": 128}) # change the x size
Source code in packages/pipekit-ops/src/pipekit_ops/signature.py
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 64 65 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 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 | |
drop_dims(names)
¶
Return a new Signature with the given dim names removed.
Raises:
| Type | Description |
|---|---|
KeyError
|
if a name in |
Source code in packages/pipekit-ops/src/pipekit_ops/signature.py
format()
¶
Return a compact "(name=size, …) dtype" string.
Source code in packages/pipekit-ops/src/pipekit_ops/signature.py
from_pattern(pattern, dtype=None)
classmethod
¶
A sizeless Signature from a dims pattern, for annotations.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pattern
|
str
|
Whitespace-separated dim names, e.g. |
required |
dtype
|
str | None
|
An optional dtype label. |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
if the pattern has a |
Example
Signature.from_pattern("band y x").dims (('band', None), ('y', None), ('x', None))
Source code in packages/pipekit-ops/src/pipekit_ops/signature.py
matches(pattern)
¶
Whether these dims fit a dims pattern; sizes are ignored.
A pattern is whitespace-separated dim names with at most one
..., which matches zero or more dims at its position. Names
compare exactly and in order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pattern
|
str
|
E.g. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
if the pattern has more than one |
Examples:
>>> sig = Signature(dims=(("time", 12), ("band", 2), ("y", 4), ("x", 4)))
>>> sig.matches("time band y x"), sig.matches("... y x")
(True, True)
>>> sig.matches("band y x"), sig.matches("time ... x")
(False, True)
Source code in packages/pipekit-ops/src/pipekit_ops/signature.py
replace_dims(sizes)
¶
Return a new Signature with the listed dim sizes overridden.
Raises:
| Type | Description |
|---|---|
KeyError
|
if a name in |
Source code in packages/pipekit-ops/src/pipekit_ops/signature.py
common_signature(*signatures)
¶
The signature every branch agrees on, or None if any differ.
For operators that may run one of several children (Try,
Branch, Switch, Coalesce): the output's signature is known
only when every child would give the same one.
Examples:
>>> s = Signature(dims=(("x", 4),), dtype="float32")
>>> common_signature(s, s) == s
True
>>> common_signature(s, None) is None
True
Source code in packages/pipekit-ops/src/pipekit_ops/signature.py
compute_output_signature(op, input_signature)
¶
Free-function dispatch to op.compute_output_signature.
Convenience for code paths that don't statically know they're
holding an Operator. Returns None when the operator opts out
of shape inference.
Source code in packages/pipekit-ops/src/pipekit_ops/signature.py
Named-array adapters¶
pipekit_ops.adapters
¶
Named-array adapters — how carrier-agnostic code reaches a carrier.
A carrier is anything with named dims and a raw array inside: a
pipekit_array.DimArray, an xr.DataArray, a geoarray GeoArray.
Code that works on any of them (KernelOp, the dim-level functions,
Sequential.summary) reaches it through an adapter registered per
carrier type, not through methods on the carrier, so no carrier library
imports pipekit in its core and xr.DataArray needs no accessor.
NamedArrayAdapter— the protocol every adapter satisfies.LabelledAdapter— one optional method,relabel, for adapters whose carriers hold labels the dim-level functions must set on a new or renamed dim (pipekit-array ADR A22).register_adapter/unregister_adapter/get_adapter— the one registry, looked up along the carrier type's MRO. Carrier libraries register on import: pipekit-array registersDimArray; geoarray registers its carriers.register_lazy_adapter— a loader for a carrier family whose library may not be imported yet: pipekit-array registers one forxr.DataArray, so importing it never imports xarray, and the adapter loads the first time a DataArray arrives.signature_of— a carrier'sSignature, through its adapter.MissingAdapterError— a known carrier family with nothing registered.backend_name,backend_of,BackendError— which array backend a raw array or a carrier sits on ("numpy","jax","dask"), forOperator.backendsand pipekit-array's kernels.
The registry lives here, in pipekit core, because pipekit-array already depends on pipekit: a lookup from core into pipekit-array would be a cycle (PK-P1). pipekit core ships no adapter itself.
BackendError
¶
LabelledAdapter
¶
Bases: Protocol
An adapter that can set the labels of named dims (optional).
rewrap keeps labels by dim name, so it cannot label a dim whose
name or labels are new (rename, merge_dims, split_dims,
broadcast_like). Code that must set them calls relabel after
rewrap; an adapter without it still serves every other function.
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
relabel(x, labels)
¶
x with the labels of the named dims set; None removes a dim's labels.
Each tuple has one entry per index of its dim; dims left out keep theirs.
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
MissingAdapterError
¶
NamedArrayAdapter
¶
Bases: Protocol
What carrier-agnostic code needs from one carrier family.
Every method takes the carrier as its first argument, so one adapter instance serves every carrier of its type.
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
check_compatible(*xs)
¶
Every carrier back, checked to combine element by element.
Raises when they cannot (a different grid, different coordinates).
A check that can only run when a compiled program runs (a traced
transform under jit) is attached to the carriers handed back,
so callers must use what this returns.
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
data(x)
¶
dims(x)
¶
isel(x, indexers)
¶
Select by position on named dims.
Slices the data and every coordinate, label and geometry leaf on those dims together; an integer index drops its dim.
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
labels(x, dim)
¶
rewrap(x, data, dims)
¶
A carrier around new data, built on x.
Dims kept from x keep their metadata, dropped dims lose it,
new dims get none. Never used for selection. A geo carrier whose
spatial dim was consumed may return a different carrier (a
DimArray).
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
backend_name(data)
¶
"jax", "dask" or "numpy" for raw data (tracers count as JAX).
Example
backend_name([1.0, 2.0]) 'numpy'
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
backend_of(x)
¶
The backend of a carrier or a raw array; None for any other value.
A registered carrier answers through its adapter's raw data; a NumPy, JAX (tracers included) or dask array answers for itself. Anything else (a number, a NumPy scalar or dtype, a string, a table) has no backend.
Example
import numpy as np backend_of(np.zeros(3)), backend_of(np.int64(0)), backend_of("a.tif") ('numpy', None, None)
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
get_adapter(x)
¶
The adapter for x's type, or None for an unnamed value.
Looks the type up along its MRO, so a subclass gets its base's adapter unless it has its own (the closest registration wins).
Raises:
| Type | Description |
|---|---|
MissingAdapterError
|
if |
Example
get_adapter([1.0, 2.0]) is None True
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
register_adapter(carrier_type, adapter)
¶
Register adapter for carrier_type and its subclasses.
Re-registering a type replaces its adapter.
Raises:
| Type | Description |
|---|---|
TypeError
|
if |
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
register_lazy_adapter(module_prefix, loader)
¶
Register loader for carrier types defined under module_prefix.
The first time get_adapter meets an unregistered type whose module
starts with module_prefix ("xarray."), it calls loader
once, which registers the real adapter, and looks again. The carrier's
library is already imported by then, so nothing is imported early.
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
signature_of(x)
¶
x's Signature if it is a registered carrier; x itself otherwise.
A Signature and None pass through unchanged, and so does any
other value, so domain libraries that thread their own signature
types keep working.
Source code in packages/pipekit-ops/src/pipekit_ops/adapters.py
unregister_adapter(carrier_type)
¶
Import hooks¶
pipekit_ops.imports
¶
Run code when a module is imported, without importing it.
on_import lets a library hook into an optional dependency (register a
pytree with JAX, a converter with xarray) without paying for that import
itself: the callback runs now when the module is already imported, and
otherwise once something else imports it. Importing pipekit-array or
geoarray therefore never imports JAX, yet their carriers are JAX pytrees
from the moment JAX is loaded.
on_import(name, callback)
¶
Call callback(module) once name is imported, without importing it.
If name is already imported the callback runs now (after waiting
for another thread's import of it to finish); otherwise it runs right
after the module's first import finishes, whoever imports it, packages
and namespace packages alike. Each callback runs once. An exception
from the callback is reported as a RuntimeWarning (on stderr if
warnings are errors) and never breaks the import of name or the
other callbacks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
A top-level module name, e.g. |
required |
callback
|
Callable[[ModuleType], object]
|
Called with the imported module. |
required |
Example
from pipekit_ops import on_import seen = [] on_import("json", lambda module: seen.append(module.name)) import json seen ['json']
Source code in packages/pipekit-ops/src/pipekit_ops/imports.py
Parallelism¶
pipekit_ops.parallel
¶
Group J — parallelism.
Four presets of Map for the four parallelism shapes pipekit deployments
actually hit. Each takes an inner Operator and applies it across a
collection; the constructors, configs and hashes are those of the
classes they replace (decision O6).
- Thread-pool:
ThreadMap(op, n_workers)— I/O-bound (S3, network). - Process-pool:
ProcessMap(op, n_workers, on_error)— CPU-bound. - Async:
AsyncMap(op, semaphore)— async I/O stacks. - Batched:
BatchedMap(op, batch_size)— vectorised / GPU inference.
All four take an iterable as input and return a list of results
(AsyncMap returns the list from inside an async coroutine).
The check_pickleable helper walks an operator tree and surfaces
flagged operators (forbid_in_yaml = True) as warnings — pre-deployment
lint for ProcessMap shapes.
See master plan Report 2, Group J, and
packages/pipekit-ops/docs/design/api/execution.md §2.11.
AsyncMap
¶
Bases: Map
Apply op to each item of an iterable inside an async event loop.
Preset: Map(op, executor="async", max_in_flight=semaphore).
If op._apply is a coroutine function, it's awaited directly.
Otherwise it runs in a thread. At most semaphore items are in
flight.
AsyncMap.__call__ is itself synchronous — it drives an event loop
so callers don't need to be async. For callers already inside an
event loop (a FastAPI or LitServe handler), use
await op.run(items): calling the operator there raises a
RuntimeError that says so.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The inner |
required |
semaphore
|
int
|
Maximum number of concurrent in-flight tasks. |
8
|
Returns from __call__(iterable): a list of results in input order.
Source code in packages/pipekit-ops/src/pipekit_ops/parallel.py
BatchedMap
¶
Bases: Map
Split an iterable into batches and call op on each batch as a list.
Preset: Map(op) (serial) over lists of batch_size items, then
an optional flatten. The carrier-agnostic batching primitive. The
inner operator receives a list of items at a time; what it does
with that list is its problem (vectorise, send to GPU, concatenate
via np.stack, …). Array-aware variants that split along axis 0
of a single array live in pipekit-array.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The inner |
required |
batch_size
|
int
|
Maximum items per batch. |
8
|
flatten
|
bool
|
If |
True
|
Returns from __call__(iterable): a list — either of items
(when flatten=True and the inner returns lists) or of batch
outputs (otherwise).
Source code in packages/pipekit-ops/src/pipekit_ops/parallel.py
ProcessMap
¶
Bases: Map
Apply op to each item of an iterable using a process pool.
Preset: Map(op, executor="process", n_workers=n_workers,
on_error=on_error), started with the platform's default start
method; a FutureWarning says when that is not spawn, which the
preset adopts in the next minor release. Use for CPU-bound,
pickleable workloads. Every operator in the chain — and every
closure inside it — must be pickleable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The inner |
required |
n_workers
|
int
|
Maximum concurrent processes. |
8
|
on_error
|
str
|
|
'raise'
|
Returns from __call__(iterable): a list of results in input order.
Source code in packages/pipekit-ops/src/pipekit_ops/parallel.py
ThreadMap
¶
Bases: Map
Apply op to each item of an iterable using a thread pool.
Preset: Map(op, executor="thread", n_workers=n_workers).
Use for I/O-bound workloads where the GIL releases inside the
underlying call (network, rasterio reads, etc.).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The inner |
required |
n_workers
|
int
|
Maximum concurrent threads. |
8
|
Returns from __call__(iterable): a list of results in input order.
Source code in packages/pipekit-ops/src/pipekit_ops/parallel.py
check_pickleable(op)
¶
Return the list of forbid_in_yaml-flagged operators inside op.
Pre-deployment lint for ProcessMap shapes: any operator carrying
forbid_in_yaml = True (closures, callables, user RNGs) is a
pickleability hazard.
Walks the operator tree generically: every instance attribute of
every operator is scanned for nested operators — held directly,
inside list / tuple / set / dict containers, or inside Graph
nodes — so new operator shapes are covered without registering
their attribute names here. Returns the flagged operators in
encounter order (deduplicated by id).
Source code in packages/pipekit-ops/src/pipekit_ops/parallel.py
Execution¶
pipekit_ops.execution
¶
Execution: one engine, Map, that runs an operator over many items.
Where the work runs is an argument ("serial", "thread",
"process", "async") or a deployment choice bound around the call
("scope" with executor_scope). Every executor shares one window
loop (Map._drive): input is pulled lazily, at most max_in_flight
items are held at a time, and on_error acts on items, not on chunks.
Prefetch and Fold are built on it.
Design: packages/pipekit-ops/docs/design/api/execution.md.
Failed
dataclass
¶
The slot of an item that failed under on_error="collect".
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
index
|
int
|
The item's position in the input. |
required |
exception
|
Exception
|
What the item raised. |
required |
Examples:
Source code in packages/pipekit-ops/src/pipekit_ops/execution.py
Fold
¶
Bases: Operator
Associative fold over chunks, reduced in parallel through Map.
a = c(… c(c(e, f(chunk₀)), f(chunk₁)) …), with f(chunk) the left fold
of op from e over the chunk and c the combine operator.
Equals the left fold of op over all items when op is
associative and e = initial is its identity (toolz
sandbox.fold).
Algorithm
- Split the input into chunks of
chunksizeitems. - Left-fold each chunk from
initialwithop, throughMap(executor=executor). - Left-fold the partial results from
initialwithcombine.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
Binary step called on one tuple |
required |
initial
|
Any
|
The identity e of |
required |
chunksize
|
int
|
Items per chunk. |
128
|
combine
|
Operator | None
|
Binary step for the partial results, called on
|
None
|
executor
|
str
|
Where the chunks are folded, as for |
'scope'
|
Examples:
>>> from pipekit_ops import Operator
>>> class AddPair(Operator):
... def _apply(self, pair):
... acc, item = pair
... return acc + item
>>> Fold(AddPair(), 0, chunksize=3, executor="thread")(range(10))
45
>>> Fold(AddPair(), 0)([])
0
Source code in packages/pipekit-ops/src/pipekit_ops/execution.py
687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 | |
compute_output_signature(input_signature)
¶
Map
¶
Bases: Operator
Apply op to every item, on any executor, with a bounded window.
yᵢ = op(xᵢ) for i = 0 … n - 1, returned in input order.
Algorithm
- Resolve the executor; for a process pool or a borrowed
non-thread executor, check that
oppickles. - Pull items lazily,
chunksizeat a time, while fewer than W =max_in_flightitems are held, and submit each chunk. - Wait for the first completion, buffer its results, release
them (as completed, or in input order for
iter), and applyon_errorto each failed item. - On a failure under
"raise", or any exit, cancel what has not started and shut down an owned pool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The operator run on each item. |
required |
executor
|
str
|
|
'serial'
|
n_workers
|
int | None
|
Workers of an owned pool. Default 1 serial, 8 thread
and async, the CPU count for process; ignored for |
None
|
max_in_flight
|
int | None
|
The window W: items pulled and not yet released. Default 2 n_w for an owned pool and 32 for a borrowed one. |
None
|
on_error
|
str
|
|
'raise'
|
chunksize
|
int
|
Items per task c. A chunk is pulled whole, so up to W + c - 1 items can be held. |
1
|
start_method
|
str | None
|
Process start method; default |
None
|
progress
|
int | None
|
If set, every |
None
|
Raises:
| Type | Description |
|---|---|
TypeError
|
|
RuntimeError
|
|
MapError
|
Under |
Examples:
Threads, results in input order:
>>> from pipekit_ops import Identity
>>> Map(Identity(), executor="thread", n_workers=2)([3, 1, 2])
[3, 1, 2]
Collect failures, keep the rest:
>>> from pipekit_ops import Lambda
>>> inverse = Lambda(lambda x: 1 / x)
>>> try:
... Map(inverse, on_error="collect")([1, 0, 2])
... except MapError as error:
... error.results
[1.0, Failed(index=1, exception=ZeroDivisionError('division by zero')), 0.5]
Stream lazily, in completion order:
Log progress every 2 items:
>>> records = []
>>> handler = logging.Handler()
>>> handler.emit = records.append
>>> log.addHandler(handler)
>>> log.setLevel(logging.INFO)
>>> Map(Identity(), progress=2)(range(5))
[0, 1, 2, 3, 4]
>>> log.removeHandler(handler)
>>> log.setLevel(logging.NOTSET)
>>> [record.done for record in records]
[2, 4]
Source code in packages/pipekit-ops/src/pipekit_ops/execution.py
251 252 253 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 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 | |
compute_output_signature(input_signature)
¶
iter(items, ordered=True)
¶
Yield results lazily, holding at most W items at any point.
With ordered=True, results come in input order, and a finished
result waiting behind a slower item still counts against W. With
ordered=False, (i, result) pairs come as items complete.
Under "collect", a failed slot yields a Failed record and
MapError (with results=None) is raised once the input is
exhausted. Close the iterator, or consume it, to free the pool.
Examples:
>>> from pipekit_ops import Identity
>>> stream = Map(Identity(), executor="thread", max_in_flight=2)
>>> list(stream.iter(range(5)))
[0, 1, 2, 3, 4]
Source code in packages/pipekit-ops/src/pipekit_ops/execution.py
run(items)
async
¶
Return the same list as a call, from inside a running event loop.
With executor="async", coroutines run on the caller's loop and
a sync op on that loop's default thread pool; the window loop
itself waits in a thread, so the caller's loop is never blocked.
Examples:
>>> import asyncio
>>> from pipekit_ops import Identity
>>> asyncio.run(Map(Identity(), executor="async").run([1, 2]))
[1, 2]
Source code in packages/pipekit-ops/src/pipekit_ops/execution.py
MapError
¶
Bases: ExceptionGroup
Every failure of a Map(on_error="collect") run, with the partial results.
An ExceptionGroup with one exception per failed item, in input
order; each carries the note "Map: item i failed".
Attributes:
| Name | Type | Description |
|---|---|---|
results |
list[Any] | None
|
The output list, input order, with a |
Examples:
>>> error = MapError("1 of 2 items failed", [ValueError("x")], [1, None])
>>> len(error.exceptions), error.results
(1, [1, None])
Source code in packages/pipekit-ops/src/pipekit_ops/execution.py
Prefetch
¶
Bases: Operator
Run op up to depth items ahead of the consumer, in input order.
Map(op, executor, max_in_flight=depth).iter(items, ordered=True):
depth bounds items running, queued, or finished and not yet
consumed, so a slow item at the head cannot make it read further ahead.
The result is a lazy iterator; consume or close it to free the pool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The operator run ahead, typically a read. |
required |
depth
|
int
|
Items held ahead of the consumer, d ≥ 1. |
2
|
executor
|
str
|
Where |
'thread'
|
Raises:
| Type | Description |
|---|---|
TypeError
|
If the input is, or holds, a JAX tracer. |
Examples:
Read ahead, then work on each result as it arrives:
>>> from pipekit_ops import Identity
>>> ahead = Prefetch(Identity(), depth=2)
>>> list(ahead(range(4)))
[0, 1, 2, 3]
Compose with a Map that consumes the stream:
Source code in packages/pipekit-ops/src/pipekit_ops/execution.py
compute_output_signature(input_signature)
¶
executor_scope(executor)
¶
Bind the executor that Map(executor="scope") borrows inside the block.
The executor is not part of any config, so a pipeline hashes the same
on a laptop and on a cluster. Map never shuts a borrowed executor
down. Inside a Map task the scope is reset to serial, so a nested
Map(executor="scope") cannot deadlock the pool that runs it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
executor
|
Executor | None
|
Any |
required |
Examples:
>>> from concurrent.futures import ThreadPoolExecutor
>>> from pipekit_ops import Identity
>>> pool = ThreadPoolExecutor(max_workers=2)
>>> with executor_scope(pool):
... Map(Identity(), executor="scope")([1, 2, 3])
[1, 2, 3]
>>> pool.submit(abs, -1).result()
1
>>> pool.shutdown()
Source code in packages/pipekit-ops/src/pipekit_ops/execution.py
Serialisation¶
pipekit_ops.serial
¶
Group L — serialisation glue.
A small surface for JSON serialisation, plus a sandboxed loader for user-uploaded pipelines that only constructs operators registered in an allowlist.
dumps(op) -> str— JSON-encodeop.state.loads(s) -> Operator— round-trip viaOperator.from_state. Any importedOperatorsubclass resolves, at any depth: aSequential,Graph,Cache,Try,Fanoutor parallel map comes back with its children. Operators that hold closures (forbid_in_yaml) refuse.register(cls)— add anOperatorsubclass to the sandboxed loader's allowlist.from_state_sandboxed(state) -> Operator— the same check on a state record already in memory (what other front ends, such as a YAML loader, build on).loads_sandboxed(s) -> Operator— likeloadsbut rejects any class, at any depth, not in the allowlist. Nothing is allowed by default, pipekit's own containers included: registerSequential,Graph, … as well as your steps. Use this for untrusted input (use case 6: user-uploaded pipelines).
YAML / Hydra loaders are intentionally not shipped here — they live
in optional extras (pipekit[yaml], pipekit[hydra]).
See master plan Report 2, Group L.
allowed()
¶
dumps(op)
¶
JSON-encode an operator's state record.
Round-trip: loads(dumps(op)) reconstructs an operator with the
same config, nested operators included. Operators with
forbid_in_yaml = True emit debug-only payloads (their constructors
need closures that aren't in the config), so loads refuses them
with RuntimeError, wherever they sit in the tree.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The operator to encode. |
required |
Returns:
| Type | Description |
|---|---|
str
|
A JSON string suitable for round-trip via |
Raises:
| Type | Description |
|---|---|
TypeError
|
if |
Source code in packages/pipekit-ops/src/pipekit_ops/serial.py
from_state_sandboxed(state)
¶
Build an operator tree from a state record, allowlisted classes only.
The check behind loads_sandboxed, for a record already in memory
(a YAML front end builds on it). Every class in the tree is checked
before anything is built, so a refused pipeline constructs nothing.
Raises:
| Type | Description |
|---|---|
PermissionError
|
if any class in the tree isn't registered. |
ValueError / TypeError / RuntimeError
|
as |
Source code in packages/pipekit-ops/src/pipekit_ops/serial.py
loads(s)
¶
Decode an operator from its JSON state record.
Resolves the target class via Operator.from_state, walking all
imported subclasses. The caller must have imported the module
declaring the target class — pipekit doesn't auto-import.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
s
|
str
|
A JSON string produced by |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
if the JSON is malformed. |
TypeError
|
if a named class isn't a loaded |
RuntimeError
|
if a class in the tree is |
Source code in packages/pipekit-ops/src/pipekit_ops/serial.py
loads_sandboxed(s)
¶
Like loads, but reject any class, at any depth, not in the allowlist.
Use for untrusted input (user-uploaded pipelines, use case 6).
Domain libraries register their operators via register; the
sandboxed loader refuses to construct anything else. Every class in
the tree is checked before any operator is built, so a refused
pipeline constructs nothing. Exception classes (Try.on,
Retry.on) resolve only among already-imported modules.
Raises:
| Type | Description |
|---|---|
PermissionError
|
if any class in the tree isn't registered. |
ValueError / TypeError / RuntimeError
|
as |
Source code in packages/pipekit-ops/src/pipekit_ops/serial.py
register(cls)
¶
Register cls as loadable via loads_sandboxed.
Idempotent — re-registering the same class is a no-op. The class
must subclass Operator.
Returns cls so the function doubles as a class decorator::
@register
class MyOp(Operator):
...
Source code in packages/pipekit-ops/src/pipekit_ops/serial.py
unregister(cls)
¶
upgrade_state(state)
¶
state with module names saved before pipekit 0.2.0 renamed.
The core was imported as pipekit until 0.2.0, so a state saved
earlier names pipekit.control where today's says
pipekit_ops.control. Loading accepts both; this rewrites the old
names in operator records (module beside class) and exception
records ({"exception": "module.Name"}), so a stored state can be
compared with op.state of the operator rebuilt from it. Everything
else is returned unchanged.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state
|
Any
|
An operator state, or any JSON value inside one. |
required |
Returns:
| Type | Description |
|---|---|
Any
|
A new value with today's module names. |
Examples:
>>> upgrade_state({"module": "pipekit.blocks", "class": "Const", "config": {}})
{'module': 'pipekit_ops.blocks', 'class': 'Const', 'config': {}}
Source code in packages/pipekit-ops/src/pipekit_ops/serial.py
State primitives¶
pipekit_ops.state
¶
Group M — state primitives.
Two primitives that enable stateful pipelines — the substrate for the
pipekit-cycle sister package (time-stepping, DA cycles, iterative
inference) and for the pipekit-train stateful trainer loop:
StatefulOperator— subclass marker._applyhas the signature(carrier, state) -> (carrier, state)instead of the statelesscarrier -> carrier.CarryState— base class for state threaded through a stateful pipeline. Subclasses define their own fields; the base provides JSON-roundtrip discipline (to_dict/from_dict).
Sequential detects when any of its operators is a
StatefulOperator and threads state through automatically. Stateless
operators in the chain pass state through unchanged. A Sequential
containing only stateless operators behaves exactly as it did before
this group landed.
Why this lives in pipekit core (not in pipekit-cycle):
Sequentialis defined here; its stateful-detection logic must be here too. PuttingStatefulOperatorin a sibling package would invert the dependency.- Other sibling packages may want state —
pipekit-train'sTrainingLoopis itself stateful (optimizer / step / metrics threaded through epochs). Sharing the base class lets them all reuse the same plumbing.
See master plan Report 2, Group M.
CarryState
¶
Base class for state threaded through a stateful pipeline.
Concrete subclasses define their own fields — the base class
intentionally does not dictate a schema. Different cycle patterns
bring their own CarryState shape (pipekit-cycle.DAState,
IterationState, WindowState, ...).
States compare by value (same type, same fields) but are not hashable: they are mutable (a stateful step may update one in place) and routinely hold arrays and dicts, so a value hash would be wrong after a mutation, or fail outright. Don't use a state as a dict key or a set member; key by something stable instead.
The base supplies JSON-roundtrip discipline:
to_dictreturns the instance's__dict__(so any attributes set in__init__are captured).from_dictreconstructs viacls(**d). Subclasses with non-trivial deserialisation override.
Subclassing pattern::
class IterationState(CarryState):
def __init__(self, residual_norm: float, iter_count: int) -> None:
self.residual_norm = residual_norm
self.iter_count = iter_count
state = IterationState(residual_norm=1.0, iter_count=0)
state.to_dict() # {"residual_norm": 1.0, "iter_count": 0}
IterationState.from_dict(...) # rebuilds
Source code in packages/pipekit-ops/src/pipekit_ops/state.py
from_dict(data)
classmethod
¶
StatefulOperator
¶
Bases: Operator
Marker base for operators whose _apply threads carry-state.
Subclasses implement _apply(carrier, state) -> (carrier, state)
rather than the stateless (carrier,) -> carrier. The
_is_stateful = True ClassVar flags the class so Sequential
detects it and threads state through automatically.
Subclasses may set initial_state_fn to a callable
() -> CarryState that produces a fresh initial state. This is
used by pipekit-cycle's cycle wrappers to bootstrap iteration
without forcing the caller to construct state explicitly.
Example::
class StreamMean(StatefulOperator):
def __init__(self):
self.initial_state_fn = lambda: MeanState(sum=0.0, n=0)
def _apply(self, x, state):
new = MeanState(sum=state.sum + x, n=state.n + 1)
return new.sum / new.n, new
See pipekit-cycle (Report 10) for the cycle / DA wrappers built
on this primitive.
Source code in packages/pipekit-ops/src/pipekit_ops/state.py
Structural protocols¶
pipekit_ops.protocols
¶
Group N — carrier-agnostic structural protocols.
The duck-typed seams that model-inference wrappers across the sister
libraries (geotoolz ModelOp, xrops ModelOp, xrsklearn SklearnOp,
pipekit-jax JaxModelOp) all rely on, promoted to runtime-checkable
Protocols so the contract is written down once instead of re-implied in
each wrapper's docstring. Domain-specific protocols stay in their
domain packages (pipekit_cycle.protocols,
pipekit_experiment.protocols); this module holds only the
carrier-agnostic ones.
runtime_checkable checks method presence, not signatures — an
isinstance(model, Predictor) check is a structural sniff, not a
guarantee that the call will succeed with your carrier.
FittableTransformer
¶
Bases: Protocol
Anything with sklearn-style fit + transform methods.
The split-object pattern's structural contract: an estimator that
learns state from data (fit) and applies it (transform).
scikit-learn transformers satisfy it as-is; the sister libraries'
carrier-aware wrappers (GeoTensor / xarray marshalling) satisfy it
by construction.
Members
fit(x): Learn internal state from x; return the fitted
estimator (conventionally self).
transform(x): Apply the learned state to x.
Source code in packages/pipekit-ops/src/pipekit_ops/protocols.py
Predictor
¶
Bases: Protocol
Anything with a sklearn-style predict method.
The structural contract behind model-inference wrappers: scikit-learn
estimators, Keras models, and the marshalling wrappers in the sister
libraries all satisfy it. Models invoked via bare __call__
(PyTorch modules, JAX functions) are covered by callable(model)
instead — a Protocol member named __call__ would match nearly
every object, so it is deliberately left out.
Members
predict(x): Map a batch of inputs to a batch of outputs.
Source code in packages/pipekit-ops/src/pipekit_ops/protocols.py
Hashing primitives¶
pipekit_ops.hashing
¶
Stable hashing primitives shared across the workspace.
One canonical implementation of the content-hash building blocks that
pipekit_ops.Cache, pipekit_train.TrainingDataset.content_hash, and the
content-addressed registries in pipekit_experiment all rely on.
Before this module each of those sites carried its own copy of the
same json.dumps(sort_keys=True, default=repr) + sha256 dance;
keeping the primitives here means a framing change can never silently
diverge between the cache key, the dataset hash, and the registry hash.
The helpers are deliberately tiny and dependency-free:
stable_json— deterministic JSON encoding (sorted keys,reprfallback for non-JSON values).stable_repr—reprwith a safe fallback for objects whosereprraises.sha256_hex— hex digest over a sequence of chunks; callers control the framing (separators, optional parts) so existing digests stay byte-for-byte stable.pipeline_hash— the content hash of a whole pipeline (and optional weights): the one the pipekit-experiment registries store under, and the key for anything that must recognise "the same pipeline again" (an orchestrator task, a cache, an artifact).
pipeline_hash(op, weights=None)
¶
Content hash of a pipeline's whole configuration, and optional weights.
sha256(stable_json(op.state) ‖ 0x00 ‖ weights). op.state
records every nested operator and, for a Graph, every node and
its wiring, so two pipelines get the same hash exactly when they
are configured the same. It is the hash the pipekit-experiment
registries store under, so registry.store(op) returns
pipeline_hash(op).
Operators that hold closures (forbid_in_yaml) are hashed by
their debug config, which names the callable but not its code: two
different lambdas can share a hash.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The operator (or pipeline) to hash. |
required |
weights
|
bytes | None
|
Optional weight bytes; |
None
|
Returns:
| Type | Description |
|---|---|
str
|
The 64-character lowercase hex digest. |
Examples:
>>> from pipekit_ops import Const
>>> pipeline_hash(Const(1)) == pipeline_hash(Const(1))
True
>>> pipeline_hash(Const(1)) == pipeline_hash(Const(2))
False
Source code in packages/pipekit-ops/src/pipekit_ops/hashing.py
sha256_hex(*chunks)
¶
Hex sha256 digest of the concatenated chunks.
String chunks are UTF-8 encoded; bytes chunks are hashed as-is.
Callers own the framing — pass explicit separator chunks (e.g.
b"\x00") between variable-length parts to avoid ambiguity.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*chunks
|
bytes | str
|
The parts to fold into the digest, in order. |
()
|
Returns:
| Type | Description |
|---|---|
str
|
The 64-character lowercase hex digest. |
Source code in packages/pipekit-ops/src/pipekit_ops/hashing.py
stable_json(obj)
¶
JSON-encode obj deterministically for hashing.
Keys are sorted so dict ordering never changes the digest; values
that are not JSON-serialisable fall back to repr. The result is
stable across processes for JSON-native payloads (the common case:
Operator.get_config() dicts).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
obj
|
Any
|
Any value. Non-JSON values are encoded via |
required |
Returns:
| Type | Description |
|---|---|
str
|
The canonical JSON string. |
Source code in packages/pipekit-ops/src/pipekit_ops/hashing.py
stable_repr(value)
¶
Return a stable string representation of value for hashing.
Tries repr first; falls back to a type-name placeholder for
objects whose repr raises. The result need not be reversible —
only deterministic per-process for equal inputs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
Any
|
Any value. |
required |
Returns:
| Type | Description |
|---|---|
str
|
|
Source code in packages/pipekit-ops/src/pipekit_ops/hashing.py
Testing¶
pipekit_ops.testing
¶
Contract checks for pipekit operators.
Call these from a library's test suite on every operator it ships:
assert_round_trip(op, *inputs)— the config round-trip contract (roadmap gate A1):loads(dumps(op))has the samepipeline_hashand returns the same output on a fixed input.public_operators(module)— everyOperatorsubclass a module exports in its__all__, so a sweep can fail on an operator that has no test case yet.
Example::
from pipekit_ops.testing import assert_round_trip
def test_scale_round_trips():
assert_round_trip(Scale(factor=2.0), 3.0)
assert_round_trip(op, *inputs, **kwargs)
¶
Assert that op survives loads(dumps(op)) unchanged.
Checks, in order:
dumpsthenloadsrebuild an operator of the same class.- The rebuilt operator has the same
pipeline_hash, so its whole configuration (nested operators andGraphwiring included) came back. - Called on the same
inputsandkwargs, both operators return the same output. Comparison is exact: arrays (anything exposing__array__) must match in type, shape, dtype and bytes, soNaNequalsNaNbut-0.0differs from0.0; floats compare exactly (NaNequal toNaN); dicts (whatever their key order, as with==), lists and tuples are compared item by item; an iterator, such asPrefetchreturns, is consumed and compared as a list; any other value with==. Array outputs need NumPy installed.
Both operators receive the same input objects: an operator that mutates its input breaks the carrier contract and may fail here.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
op
|
Operator
|
The operator (or |
required |
*inputs
|
Any
|
Positional inputs for the call. |
()
|
**kwargs
|
Any
|
Keyword inputs for the call (a |
{}
|
Returns:
| Type | Description |
|---|---|
Operator
|
The rebuilt operator, for further checks. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if |
AssertionError
|
on the first violated property, naming the operator. |
Example::
>>> from pipekit_ops import Const
>>> from pipekit_ops.testing import assert_round_trip
>>> back = assert_round_trip(Const(3), "ignored")
>>> back.value
3
Source code in packages/pipekit-ops/src/pipekit_ops/testing.py
public_operators(module)
¶
Every Operator subclass named in module.__all__.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
module
|
ModuleType
|
An imported package or module with an |
required |
Returns:
| Type | Description |
|---|---|
dict[str, type[Operator]]
|
|
dict[str, type[Operator]]
|
( |
dict[str, type[Operator]]
|
caller decides what to skip. |
Raises:
| Type | Description |
|---|---|
AttributeError
|
if |