lacing

lacing — interval annotation system.

Standoff, interval-keyed annotations with rational time, ELAN tier stereotypes, Allen’s interval algebra, and a MutableMapping facade.

Quick start:

>>> from lacing import RationalTime, TimeInterval, Annotation, MemoryStore
>>> # Load a TextGrid, query overlaps, save as WebVTT — see misc/docs/.

Read CLAUDE.md and misc/docs/Lacing Development Roadmap.md for the full story. .claude/skills/ contains the rules.

class lacing.AllenRelation(value)

The thirteen Allen relations.

Symbols match Allen (1983); inverse pairs end in i.

inverse() → AllenRelation

The inverse relation.

class lacing.Annotation(*, id: UUID, tier: str, reference: MediaRef | NodeRef | AnnotationRef, body: dict, body_schema_uri: Annotated[str, _PydanticGeneralMetadata(pattern='^annot://schema/[a-z0-9-]+/v\\d+$')], provenance: Provenance, confidence: Annotated[float | None, Ge(ge=0.0), Le(le=1.0)] = None)

The single annotation envelope. body is typed by body_schema_uri.

property interval: TimeInterval | None

the reference’s interval, if any.

Type:

Convenience

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.AnnotationRef(*, kind: Literal['annotation'] = 'annotation', target_id: UUID, interval: TimeInterval | None = None)

Reference to another annotation (for discussion threads, review, derivations).

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.Artifact(*, asset_id: Annotated[str, MinLen(min_length=64), MaxLen(max_length=64), _PydanticGeneralMetadata(pattern='^[0-9a-f]{64}$')], kind: Literal['image', 'video', 'audio', 'json', 'text', 'binary'], path: Path | None = None, url: str | None = None, bytes_size: Annotated[int, Ge(ge=0)], duration_s: Annotated[float | None, Ge(ge=0)] = None, mime: str | None = None, provenance: Provenance, cost_usd: Annotated[float | None, Ge(ge=0)] = None, producer_call_id: str | None = None)

A content-addressed generated file with provenance.

asset_id is the SHA-256 hex digest of the artifact’s bytes. Two artifacts with the same asset_id are byte-identical regardless of where they live — so caches keyed on asset_id are safe across machines and re-runs.

provenance reuses lacing.Provenance so the lineage chain (was_derived_from, was_generated_by) is the same for artifacts and annotations. An annotation referencing an artifact does so via MediaRef(asset_id=artifact.asset_id, …).

classmethod from_bytes(data: bytes, *, kind: Literal['image', 'video', 'audio', 'json', 'text', 'binary'], was_generated_by: str, was_attributed_to: str, path: Path | str | None = None, url: str | None = None, was_derived_from: tuple[UUID | Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=64), MaxLen(max_length=64), _PydanticGeneralMetadata(pattern='^[0-9a-f]{64}$')])], ...] = (), activity: str = 'create', generated_at_time: RationalTime | None = None, duration_s: float | None = None, mime: str | None = None, cost_usd: float | None = None, producer_call_id: str | None = None) → Artifact

Create an Artifact from in-memory bytes.

classmethod from_path(path: Path | str, *, kind: Literal['image', 'video', 'audio', 'json', 'text', 'binary'], was_generated_by: str, was_attributed_to: str, was_derived_from: tuple[UUID | Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=64), MaxLen(max_length=64), _PydanticGeneralMetadata(pattern='^[0-9a-f]{64}$')])], ...] = (), activity: str = 'create', generated_at_time: RationalTime | None = None, duration_s: float | None = None, mime: str | None = None, cost_usd: float | None = None, producer_call_id: str | None = None) → Artifact

Create an Artifact from a local file. Hashes the file’s bytes.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_media_ref(interval) → MediaRef

Return a lacing.MediaRef pointing at this artifact.

Use this to attach an annotation to a region of the artifact: MediaRef(asset_id=artifact.asset_id, interval=…).

class lacing.ArtifactStore(catalog: MutableMapping[str, BaseModel], blobs: MutableMapping[str, bytes] | None = None)

Facade over an artifact catalog and an optional blobs store.

The object is a MutableMapping[str, record] over the catalog — store[artifact_id], iteration, len, get, clear and the rest of the mapping surface all act on artifact metadata records. The heavier byte operations (put_blob(), get_blob(), has_blob()) are rich methods that are deliberately not squeezed into the mapping protocol.

Parameters:

• catalog – Injected id -> record store. Records are pydantic models (lacing.Artifact by default, but any BaseModel works — the store does not inspect the record’s shape).

• blobs – Injected content_hash -> bytes store, or None for a catalog-only store (Stage-1 metadata persistence). Blob methods raise / no-op when it is None.

Construct one with in_memory() or from_directory() rather than wiring the backing stores by hand, unless you are injecting a custom backend.

blob_location(content_hash: str) → str | Path | None

Resolve the cheapest servable location for a blob — without reading its bytes. The capability the HTTP layer probes to serve a blob the most efficient way, generalizing blob_path() so filesystem / S3 / R2 backends all answer one probe:

• object-store backends (S3/R2) that expose a presigned-URL capability — a url_for(content_hash) callable — return a URL string, so the caller can 302-redirect and let the object store serve the bytes (and HTTP Range) directly, off the app process;

• filesystem backends return a local Path (see blob_path()) so the caller hands the OS the file (Range free);

• everything else (plain dict, or a missing blob) returns None — the cue to fall back to iter_blob() streaming.

Callers treat None as “stream it”, not as an error.

blob_path(content_hash: str) → Path | None

Return the local filesystem path of the blob, or None.

The store’s blob backend is opaque (any MutableMapping), but some backends — notably the filesystem-backed dol.Files produced by from_directory() — store each blob as one file under a known root directory. This method exposes that path when available, so a caller (e.g. a FastAPI route serving video) can hand the OS the file descriptor and let it answer HTTP Range requests directly. It returns None for:

• blob stores without a rootdir attribute (e.g. plain dict, object-store backends — callers should fall back to iter_blob());

• blobs that are not present.

Callers must treat None as the cue to use the streaming read path, not as an error.

count_refs(content_hash: str) → int

How many catalog records point at content_hash (a blob).

The reference count that a garbage collector needs: a blob is safe to delete only when no catalog record still names its content hash. Content-addressed blobs are deduplicated, so two artifacts can share one blob — deleting the blob the moment one of them goes away would orphan the other.

This is the probe half of a capability pair (mirroring blob_location() probing a blob store for url_for): a catalog backend that can answer the count cheaply — the SQL catalog from from_sql(), via an indexed content_hash column — exposes a refcount_by_content_hash callable, and this method uses it so the count is one indexed query, not a full catalog scan. Any other catalog (dict, Files) falls back to scanning every record. The facade is unchanged either way; only the speed differs.

Parameters:

content_hash – The blob’s content hash (Artifact.asset_id).

Returns:

The number of catalog records whose content hash equals content_hash.

classmethod from_aws(bucket_name: str, uri: str, *, record_type: type[BaseModel] = <class 'lacing.artifact.Artifact'>, collection_name: str = 'artifact_catalog', prefix: str | None = None, s3_kwargs: dict | None = None, sql_kwargs: dict | None = None) → ArtifactStore

The production pairing: S3-compatible blobs + SQL catalog.

A thin convenience over from_s3() (blobs) and from_sql() (catalog) so the common cloud deployment is one call. The blob store is an S3Store (AWS S3 / Cloudflare R2 / MinIO / Supabase); the catalog is the durable, queryable SQL table — Postgres in production, SQLite for a smoke test. Vendor specifics live only in the two kwargs dicts.

Parameters:

• bucket_name – Object-store bucket for the content-addressed blobs.

• uri – SQLAlchemy URI for the catalog (postgresql://… in prod).

• record_type – Pydantic model the catalog (de)serializes.

• collection_name – SQL table name for the catalog.

• prefix – Optional key prefix within the bucket.

• s3_kwargs – Extra kwargs forwarded to from_s3() (credentials, endpoint_url for R2/MinIO/Supabase, region_name, …).

• sql_kwargs – Extra kwargs forwarded to from_sql() (content_hash_of, SQLAlchemy engine kwargs, …).

classmethod from_directory(root: Path | str, *, record_type: type[BaseModel] = <class 'lacing.artifact.Artifact'>) → ArtifactStore

An ArtifactStore persisted under root.

Lays out two subdirectories: catalog/ (one <id>.json file per record) and blobs/ (one file per content hash). Both are dol filesystem stores, so the same facade works unchanged over any other dol backend (object storage, etc.) when injected directly.

Parameters:

• root – Directory to hold the store. Created if missing.

• record_type – The pydantic model the catalog deserializes JSON into. Defaults to Artifact; callers with their own record schema pass their model here.

classmethod from_s3(bucket_name: str, *, catalog: MutableMapping[str, BaseModel] | None = None, prefix: str | None = None, **s3_kwargs) → ArtifactStore

An ArtifactStore with blobs in an S3-compatible object store (AWS S3 / Cloudflare R2 / MinIO / Supabase) via s3dol.

Content-addressed blobs are a natural fit for object storage: flat, immutable, dedup-friendly, forever-cacheable keys. blob_location() returns a presigned GET URL (via s3dol’s url_for) so a serving layer can 302-redirect and let the store deliver the bytes (and HTTP Range) directly, off the app process.

The blob and catalog backends are independent by design: the catalog defaults to an in-memory dict here — swap in a durable catalog (e.g. a Postgres-backed MutableMapping) for production.

Parameters:

• bucket_name – The object-store bucket.

• catalog – The id -> record catalog MutableMapping. Defaults to {} (in-memory).

• prefix – Optional key prefix within the bucket (e.g. "blobs").

• **s3_kwargs – Forwarded to s3dol.s3_store — endpoint_url (set for R2 / MinIO / Supabase), region_name, profile, credentials, preset, anon, on_missing_bucket. The pre-v1 spellings (aws_access_key_id / aws_secret_access_key / aws_session_token / profile_name / make_bucket) are still accepted and translated, with a DeprecationWarning.

Requires s3dol>=1 (and boto3) — imported lazily so the dependency is only needed when this constructor is used.

classmethod from_sql(uri: str, *, blobs: ~collections.abc.MutableMapping[str, bytes] | None = None, record_type: type[~pydantic.main.BaseModel] = <class 'lacing.artifact.Artifact'>, collection_name: str = 'artifact_catalog', content_hash_of: ~collections.abc.Callable[[~pydantic.main.BaseModel], str | None] | None = None, **db_kwargs) → ArtifactStore

An ArtifactStore with a SQL-backed catalog (durable, queryable) via sqldol — the SQL counterpart of from_s3()’s object store.

The catalog row schema is deliberately minimal and vendor-neutral: one TEXT column holds the record serialized exactly as from_directory() serializes it (record.model_dump_json out, record_type.model_validate_json in), and one indexed content_hash column carries the record’s blob hash so the GC reference count (count_refs()) is a single indexed query rather than a full scan.

Because the connection is just a SQLAlchemy URI, the same code runs on SQLite for tests and on Postgres in production — that is the whole point of the facade. Pick the backend with the uri alone:

ArtifactStore.from_sql("sqlite:///artifacts.db")          # local / tests
ArtifactStore.from_sql("postgresql://u:p@host:5432/db")   # production

The catalog and blob backends are independent: pass blobs to pair a SQL catalog with any blob store (e.g. from_s3’s S3Store), or leave it None for a catalog-only store (Stage-1 metadata persistence; see from_aws() for the common S3 + SQL pairing).

Parameters:

• uri – SQLAlchemy connection URI (sqlite:///… or postgresql://…). The single knob that selects the vendor.

• blobs – Optional content_hash -> bytes blob store. None for a catalog-only store.

• record_type – The pydantic model the catalog deserializes JSON into. Defaults to Artifact; callers with their own record schema pass their model here.

• collection_name – The SQL table name for the catalog.

• content_hash_of – How to read a record’s blob hash for the indexed content_hash column. Defaults to asset_id (the canonical Artifact field), then content_hash. Pass a callable for a record whose hash lives elsewhere; records without a hash store an empty string.

• **db_kwargs – Forwarded to sqldol.SQLAlchemyStore / SQLAlchemy’s create_engine (e.g. connect_args, pool_size).

Requires sqldol (and SQLAlchemy) — imported lazily so the dependency is only needed when this constructor is used.

get_blob(content_hash: str) → bytes | None

Return the bytes for content_hash, or None if absent.

has_blob(content_hash: str) → bool

Whether the blob store holds content_hash.

classmethod in_memory() → ArtifactStore

An ArtifactStore backed entirely by in-memory dicts.

For tests, scratch work, and as the trivial reference backend. Nothing persists across processes.

index() → dict[str, BaseModel]

Return the whole catalog as a plain dict (e.g. for UI hydration).

iter_blob(content_hash: str, *, chunk_size: int = 65536) → Iterator[bytes]

Yield the blob’s bytes in chunk_size chunks.

The streaming counterpart to get_blob() — what an HTTP response body iterates over when serving a large blob without holding it all in process memory. The default implementation reads the whole blob via get_blob() and re-chunks it; a filesystem-backed store can be swapped for a true streaming reader without changing this API.

Raises:

KeyError – no blob exists for content_hash.

put_blob(data: bytes) → str

Store data content-addressed; return its content hash.

Idempotent: identical bytes always map to the same hash and overwrite an identical blob. Delegates to put_blob_stream() so there is exactly one write path — and therefore one atomicity story (lacing#25).

Raises:

RuntimeError – no blob store is configured.

put_blob_stream(chunks: Iterable[bytes]) → str

Stream chunks content-addressed; return their SHA-256 hash.

The streaming-friendly counterpart to put_blob() — callers hand in an iterable (e.g. requests.Response.iter_content) instead of materializing the whole bytestring upfront. The hash is computed on the fly.

When the blob store is filesystem-backed (exposes rootdir, as the default directory store does), the bytes spool straight to a same-directory tempfile and are renamed into place with os.replace once the hash is known (lacing#25). Consequences a caller may rely on:

• peak memory is one chunk, not 2× the payload — 100 MB videos flow through without a 200 MB spike;

• a partial blob is never observable under its content address — the name a reader could find only exists after the rename, which is atomic on POSIX and on Windows for same-directory renames — so has_blob(h) == True really does mean the full bytes are there;

• a failed or abandoned stream leaves nothing behind (the tempfile is unlinked).

Backends without a rootdir (plain dict, object-store mappings) fall back to buffering the payload and assigning it whole — their __setitem__ is the atomicity story there.

Raises:

RuntimeError – no blob store is configured.

save(artifact_id: str, record: BaseModel, *, data: bytes | None = None) → str | None

Persist one artifact, optionally with its bytes.

Writes the blob (if data is given) before the catalog row, so a crash never leaves the catalog pointing at missing bytes. Idempotent on artifact_id — and, for the blob, on content — so retries are safe.

Parameters:

• artifact_id – The stable string id this artifact is filed under.

• record – The metadata record to store in the catalog.

• data – Optional raw bytes. When given, they are stored content-addressed and the content hash is returned; the caller is responsible for also recording that hash on record.

Returns:

The blob’s content hash if data was written, else None.

Raises:

RuntimeError – data was given but no blob store is configured.

exception lacing.BodySchemaError

Raised when a body fails validation against its registered schema.

class lacing.InMemoryOpLog

Simple list-backed op-log. Thread-safe via an RLock.

class lacing.IntervalAnnotationStore(*args, **kwargs)

Protocol for any interval-keyed annotation store.

Conceptually a MutableMapping[TimeInterval, list[Annotation]]: keys are TimeInterval; values are lists because multiple annotations can share an interval (different tiers, multiple annotators, soft labels).

We use Protocol rather than inheriting from MutableMapping so backends (in-memory, SQLite, Postgres) can structurally conform without forcing a single class hierarchy. The mapping methods below match the MutableMapping ABC; concrete backends like MemoryStore implement the full interface.

add(annotation: Annotation) → None

Append annotation to the list at its reference interval.

all() → Iterator[Annotation]

Iterate every annotation in the store, order unspecified.

at_tier(tier_name: str, query: TimeInterval) → Iterator[Annotation]

Annotations on tier_name that intersect query.

by_tier(tier_name: str) → Iterator[Annotation]

All annotations on tier_name, regardless of interval.

contains(query: TimeInterval) → Iterator[Annotation]

Annotations whose interval strictly contains query (Allen di).

during(query: TimeInterval) → Iterator[Annotation]

Annotations whose interval is strictly inside query (Allen d).

equals(query: TimeInterval) → Iterator[Annotation]

Allen =: identical interval.

extend(annotations: Iterable[Annotation]) → None

Add many; equivalent to repeated .add but adapters can optimize.

finishes(query: TimeInterval) → Iterator[Annotation]

Allen f: later start, same end.

intersects(query: TimeInterval) → Iterator[Annotation]

Annotations whose interval shares any time with query.

meets(query: TimeInterval) → Iterator[Annotation]

Allen m: a.end == q.start.

overlaps(query: TimeInterval) → Iterator[Annotation]

Strict Allen o: a.start < q.start < a.end < q.end.

relate(query: TimeInterval, relations: Iterable[AllenRelation]) → Iterator[Annotation]

Annotations whose interval has any of the named relations to query.

Generic dispatch — useful when relations are computed at runtime.

remove(annotation_id) → Annotation | None

Remove and return the annotation with this id, or None if absent.

starts(query: TimeInterval) → Iterator[Annotation]

Allen s: same start, earlier end.

tiers() → Iterator[Tier]

Registered tiers. Annotations may reference tiers not yet registered; callers decide whether that’s an error.

exception lacing.LossyTimeConversionError

Raised when a rate or seconds conversion would lose precision.

class lacing.MediaRef(*, kind: Literal['media'] = 'media', asset_id: str, interval: TimeInterval)

Reference to a region of a content-addressed media asset.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.MemoryStore

IntervalAnnotationStore implementation over intervaltree.

Conforms to the protocol in lacing.store.base. We don’t formally inherit from IntervalAnnotationStore because it’s a Protocol with method bodies — structural typing is enough.

exception lacing.MigrationError

Raised when a migration step is missing or fails.

class lacing.NodeRef(*, kind: Literal['node'] = 'node', scene_path: str, interval: TimeInterval)

Reference to a node in a structured scene/document graph.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

exception lacing.NonStringBodyKeyError

An annotation body contains a mapping key that is not a str.

JSON object keys are strings, so model_dump(mode="json") coerces non-string keys — and two distinct keys can coerce to the same string, silently annihilating an entry. {1: "a", "1": "b"} dumps to {"1": "b"}; a body differing only in the lost entry would digest identically, which is a wrong cache hit.

Since lacing#24 the envelope refuses such a body at validation — the producer-side fix. This error lives here rather than in lacing.model because this module is deliberately import-light (stdlib only, pinned by test) and the model can import from it, not vice versa.

class lacing.OpLog(*args, **kwargs)

Append-only log of mutations.

Implementations must guarantee that the clock returned by append() is strictly greater than every previously-returned clock value across the lifetime of the log.

append(operation: str, *, target_id: str | None = None, payload: dict[str, Any] | None = None, actor: str = 'anonymous') → int

Append an entry; return its assigned clock.

entries(*, until_clock: int | None = None, from_clock: int | None = None) → Iterator[OpLogEntry]

Iterate entries, optionally bounded by clock range (inclusive).

latest_clock() → int

Highest clock currently in the log; 0 if empty.

class lacing.OpLogEntry(clock: int, operation: str, target_id: str | None, payload: dict[str, ~typing.Any], actor: str = 'anonymous', received_at: float = <factory>)

One row of the op-log.

actor: str

user:<handle> or agent:<model>@<hash> or adapter:<format>.

clock: int

Monotonic Lamport clock starting at 1. Strictly increasing per log.

operation: str

add_annotation, remove_annotation, update_annotation, add_tier, set_meta, import_batch.

Type:

One of

payload: dict[str, Any]

JSON-serializable payload sufficient to replay the operation.

received_at: float

Wall-clock time the operation was received (seconds since epoch).

target_id: str | None

Annotation id, tier name, meta key, or None for batch ops.

exception lacing.ProcessorError

Raised when a registered processor’s invocation fails.

class lacing.Provenance(*, was_generated_by: str, was_attributed_to: str, was_derived_from: list[~uuid.UUID | ~typing.Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=64), MaxLen(max_length=64), _PydanticGeneralMetadata(pattern='^[0-9a-f]{64}$')])]] = <factory>, generated_at_time: RationalTime, activity: str = 'create')

W3C PROV-O subset, embedded inline on every annotation.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.RationalTime(value: int, rate: int = 24000)

A point in time as value / rate seconds.

Immutable. Two RationalTime values with different rates compare via their rational value, so RationalTime(24, 24) == RationalTime(1, 1).

Examples

>>> RationalTime(24000) == RationalTime(1, 1)
True
>>> RationalTime.from_seconds("1.5", rate=2).value
3

classmethod from_seconds(seconds: float | Fraction | str, rate: int = 24000) → RationalTime

Build from seconds. Quantizes to rate; raises if lossy.

seconds may be a str like "1.001" to avoid float ingestion.

classmethod from_seconds_lossy(seconds: float | Fraction | str | int, *, rate: int = 24000, mode: str = 'round') → RationalTime

Build from seconds, quantizing to the nearest sample at rate.

Unlike from_seconds() — which raises LossyTimeConversionError when the value cannot be represented exactly — this method always succeeds by quantizing. mode selects the rounding rule:

• "round" — nearest sample, ties to even (default)

• "floor" — largest sample <= seconds

• "ceil" — smallest sample >= seconds

Use this when sample-level quantization is knowingly acceptable — the common case for user-supplied durations. Use from_seconds() when exactness matters and a lossy conversion should be an error.

Examples

>>> RationalTime.from_seconds_lossy("0.1", rate=3).value
0
>>> RationalTime.from_seconds_lossy("0.1", rate=3, mode="ceil").value
1

classmethod now(rate: int = 24000) → RationalTime

Wall-clock time as a RationalTime, quantized to rate.

Uses time.time_ns() and builds the value directly, sidestepping the float-quantization landmine of from_seconds(float). Every producer of an annotation or artifact needs this for Provenance.generated_at_time.

to_rate(new_rate: int) → RationalTime

Re-express at new_rate. Raises LossyTimeConversionError on loss.

to_seconds() → float

Float seconds — for display only. Never round-trip through this.

exception lacing.SchemaMismatchError

Raised when opening a .annot file with an incompatible schema.

class lacing.SqliteOpLog(path: str, *, check_same_thread: bool = True)

Op-log backed by a SQLite table.

Designed to share a database file with SqliteStore so the store snapshot + the op-log live together and survive a single cp project.annot project.backup step.

class lacing.SqliteStore(path: str | PathLike, *, check_same_thread: bool = True, migrate: bool = False)

SQLite-backed IntervalAnnotationStore.

Parameters:

• path – Path to the .annot file. Use ":memory:" for an ephemeral in-memory database.

• check_same_thread – Forwarded to sqlite3.connect. We hold a single connection guarded by a lock; pass False when sharing across threads.

• migrate – Opt-in to upgrading a file written at an older schema_version on open, via the ladder in lacing.store.migrations. Off by default — silently rewriting someone’s file on open is worse than refusing. Every open-time schema failure — refusal or failed migration — raises SchemaMismatchError; a failed migration chains the ladder’s StoreMigrationError as its cause.

exception lacing.StoreMigrationError

Raised when a store migration step is missing or fails.

class lacing.Tier(name: str, *, stereotype: TierStereotype = TierStereotype.NONE, parent: str | None = None, metadata: dict | None = None)

A named annotation layer with optional parent and stereotype.

Tiers are pure metadata; they don’t own annotations. The store is keyed by interval, not by tier — annotations carry their tier name as a field. This matches ELAN’s TIME_ORDER indirection (see ANN-DOC §C).

class lacing.TierStereotype(value)

Constraints on how a child tier relates to its parent tier.

Names match ELAN exactly so EAF round-trips are trivial.

INCLUDED_IN = 'INCLUDED_IN'

Children lie within the parent but gaps between siblings are allowed.

NONE = 'NONE'

No parent constraint. Top-level tier.

SYMBOLIC_ASSOCIATION = 'SYMBOLIC_ASSOCIATION'

One-to-one association with parent; child shares parent’s interval exactly.

SYMBOLIC_SUBDIVISION = 'SYMBOLIC_SUBDIVISION'

Ordered subdivision; children share parent’s interval as a sequence (no times).

TIME_SUBDIVISION = 'TIME_SUBDIVISION'

Children fully partition the parent’s interval (no gaps, no overlap).

class lacing.TimeInterval(start: RationalTime, end: RationalTime)

A half-open interval [start, end).

start == end is a valid point annotation, not a degenerate case. Always start <= end; constructor raises ValueError otherwise.

property duration: RationalTime

end - start at the same rate as start.

exception lacing.UnknownBodySchemaError

Raised when an annotation’s body_schema_uri has no registered model.

lacing.annotation_body_digest(annotation: Annotation) → str

Return the SHA-256 hex digest of {body, body_schema_uri} only.

The narrow sibling of annotation_value_digest(). It drops the entire reference — which asset / which node / which annotation, not merely when — plus tier and confidence. So the same caption over two different assets digests identically here, as does the same body asserted on two different tiers or at two different confidences.

That is a correctness bug in any consumer that reads any of those. Reach for it only when the consumer demonstrably depends on nothing but what the annotation says; prefer annotation_value_digest() otherwise.

>>> from uuid import uuid4
>>> from lacing import Annotation, MediaRef, Provenance
>>> from lacing import RationalTime, TimeInterval
>>> def over(asset):
...     return Annotation(
...         id=uuid4(), tier="words",
...         reference=MediaRef(
...             asset_id=asset,
...             interval=TimeInterval(RationalTime(0), RationalTime(24000)),
...         ),
...         body={"text": "hello"},
...         body_schema_uri="annot://schema/word/v1",
...         provenance=Provenance(
...             was_generated_by="agent:m@1",
...             was_attributed_to="thor",
...             generated_at_time=RationalTime(0),
...         ),
...     )

Different assets, identical body digest — this is the footgun:

>>> annotation_body_digest(over("sha256:interview")) == (
...     annotation_body_digest(over("sha256:broadcast"))
... )
True
>>> annotation_value_digest(over("sha256:interview")) == (
...     annotation_value_digest(over("sha256:broadcast"))
... )
False

>>> from uuid import uuid4
>>> from lacing import Annotation, MediaRef, Provenance
>>> from lacing import RationalTime, TimeInterval
>>> def make(interval):
...     return Annotation(
...         id=uuid4(),
...         tier="words",
...         reference=MediaRef(asset_id="sha256:abc", interval=interval),
...         body={"text": "hello"},
...         body_schema_uri="annot://schema/word/v1",
...         provenance=Provenance(
...             was_generated_by="agent:m@1",
...             was_attributed_to="thor",
...             generated_at_time=RationalTime(0),
...         ),
...     )
>>> early = make(TimeInterval(RationalTime(0), RationalTime(24000)))
>>> late = make(TimeInterval(RationalTime(24000), RationalTime(48000)))
>>> annotation_body_digest(early) == annotation_body_digest(late)
True
>>> annotation_value_digest(early) == annotation_value_digest(late)
False

lacing.annotation_value_digest(annotation: Annotation) → str

Return the SHA-256 hex digest of annotation’s value.

Covers body, body_schema_uri, tier, reference and confidence. Excludes id and provenance, so a regeneration that produces identical content produces an identical digest.

Use this for freshness and early cutoff. For optimistic concurrency use lacing.server.etag.annotation_etag() instead — two digests, two jobs, and neither substitutes for the other.

>>> from uuid import uuid4
>>> from lacing import Annotation, MediaRef, Provenance
>>> from lacing import RationalTime, TimeInterval
>>> def make(**kw):
...     base = dict(
...         id=uuid4(),
...         tier="words",
...         reference=MediaRef(
...             asset_id="sha256:abc",
...             interval=TimeInterval(RationalTime(0), RationalTime(24000)),
...         ),
...         body={"text": "hello"},
...         body_schema_uri="annot://schema/word/v1",
...         provenance=Provenance(
...             was_generated_by="agent:m@1",
...             was_attributed_to="thor",
...             generated_at_time=RationalTime(0),
...         ),
...     )
...     base.update(kw)
...     return Annotation(**base)

A regeneration — new id, new timestamp, same content — digests the same:

>>> a = make()
>>> b = make(provenance=Provenance(
...     was_generated_by="agent:m@1",
...     was_attributed_to="thor",
...     generated_at_time=RationalTime(999),
... ))
>>> annotation_value_digest(a) == annotation_value_digest(b)
True

A changed body does not:

>>> annotation_value_digest(make(body={"text": "goodbye"})) == (
...     annotation_value_digest(a)
... )
False

lacing.boundary_iou(a: Iterable[TimeInterval], b: Iterable[TimeInterval]) → float

Mean IoU between two sets of intervals via greedy best-match.

For each interval in a, finds its highest-IoU match in b (without replacement — once a b interval is matched it’s removed from the pool). Unmatched intervals in either set contribute 0.0 to the mean.

Returns:

Mean IoU ∈ [0, 1]. Returns 0.0 if both sets are empty (defensible as a “no agreement to measure” baseline).

lacing.cohen_kappa(a: Sequence[T], b: Sequence[T]) → float

Cohen’s kappa for two annotators on a categorical label.

Parameters:

• a – Annotator A’s labels.

• b – Annotator B’s labels (must be the same length).

Returns:

κ ∈ [-1, 1]. 1 = perfect agreement, 0 = chance, negative = worse than chance.

Raises:

ValueError – If sequences differ in length or are empty.

Edge cases:

If only one category appears across both annotators, both observed and expected agreement are 1.0; we return 1.0 by convention.

lacing.export_json_schemas(target_dir: str | Path, *, overwrite: bool = True, include_meta: bool = True) → list[Path]

Write every registered schema as JSON files under target_dir.

Layout: <target_dir>/<name>/v<N>.json. Returns the list of paths written, in registration order.

Parameters:

• target_dir – Output directory. Created if missing.

• overwrite – If False, refuse to write a file that already exists.

• include_meta – If True, also write a <target_dir>/index.json mapping every URI to its file path and the Pydantic model’s qualified name (helps the codegen pipeline).

lacing.get_tracer(name: str = 'lacing', version: str | None = None) → Any

Return a tracer or a no-op fallback.

Parameters:

• name – Instrumentation name (typically __name__ of the caller’s module or a logical name like "lacing.server").

• version – Optional package version string.

Returns:

opentelemetry.trace.Tracer if OTel is installed, else a no-op object whose start_as_current_span() is a context manager yielding a no-op span.

lacing.hash_bytes(data: bytes) → str

Return the canonical asset_id (SHA-256 hex) for data.

>>> hash_bytes(b"hello world")[:8]
'b94d27b9'

lacing.hash_file(path: Path | str, *, chunk_size: int = 1048576) → str

Return the canonical asset_id (SHA-256 hex) for the file at path.

lacing.instrument_otel(app: Any, *, tracer_name: str = 'lacing.server') → Any

Add OpenTelemetry instrumentation to a FastAPI app.

Wraps every request in a span; tags the span with the response’s X-Lacing-Clock header value (when present) as lacing.clock.

No-op when OTel isn’t installed — the app is returned unchanged.

Parameters:

• app – The FastAPI app (from lacing.server.create_app()).

• tracer_name – Name passed to get_tracer().

Returns:

The same app, with middleware installed if OTel is available.

lacing.interval_iou(a: TimeInterval, b: TimeInterval) → float

Intersection-over-Union for two time intervals.

Returns 1.0 if both are equal point intervals at the same instant; 0.0 if they don’t intersect (including when only one is a point).

lacing.is_otel_active() → bool

Quick check: is OTel installed AND a TracerProvider configured?

lacing.json_schema(uri: str) → dict

Return the JSON Schema for the body model registered at uri.

Pydantic’s model_json_schema() output, unmodified.

lacing.krippendorff_alpha(annotations: ~collections.abc.Sequence[~collections.abc.Sequence[~lacing.quality.T | None]], *, distance: ~typing.Callable[[~lacing.quality.T, ~lacing.quality.T], float] = <function _nominal_distance>) → float

Krippendorff’s α across any number of annotators.

Parameters:

• annotations – A list of annotators, each a sequence of labels (one per unit). Use None for a missing annotation by that annotator on that unit.

• distance – Function (x, y) -> float measuring disagreement between two label values. Default is the nominal (0/1) distance.

Returns:

α. 1.0 = perfect agreement, 0.0 = chance.

Raises:

ValueError – If sequences differ in length, fewer than 2 annotators given, or fewer than 2 paired observations exist.

lacing.maybe_span(tracer: Any, name: str, **attributes: Any)

Open a span on tracer, attaching attributes if supported.

Works with both real OTel tracers and the no-op fallback.

lacing.migrate(body: dict, *, from_uri: str, to_uri: str) → dict

Migrate body from from_uri to to_uri via registered steps.

Composes single-step migrations. Raises MigrationError if any step is missing.

lacing.migrate_annot_file(path: str | PathLike, *, to_version: int | None = None) → tuple[int, int]

Migrate a .annot file in place, returning (from, to) versions.

to_version defaults to the current build’s lacing.store.sqlite.SCHEMA_VERSION. Already-current files are a no-op (from == to). Each step runs in its own BEGIN IMMEDIATE transaction with the version re-checked under the lock, so concurrent migrators converge and an interrupted chain resumes from the last version that completed (idempotent).

Raises StoreMigrationError when the file does not exist, is not a .annot file, a step is missing, fails, or breaks one of the runner’s in-transaction guarantees.

lacing.register_body_schema(uri: str, model: type[BaseModel]) → type[BaseModel]

Register model as the validator for uri. Returns model.

lacing.register_migration(*, schema_name: str, from_version: int, to_version: int)

Register a forward migration from v<from_version> to v<to_version>.

Decorated function takes a body dict and returns a new body dict. Migrations must be one major-version step at a time (to_version == from_version + 1).

Re-registering the same (schema_name, from_version) pair replaces the previous entry — convenient in tests, intentional for hot-reload.

lacing.register_processor(func: Callable[[...], Any] | None = None, *, name: str | None = None)

Register a processor under name (defaults to the function name).

The function may be sync or async; we wrap sync funcs to a coroutine.

lacing.register_store_migration(*, store_kind: str, from_version: int, to_version: int)

Register a forward store migration from from_version to to_version.

The decorated function takes the backend’s open connection and must perform every change of the step — DDL, row rewrites, and the meta.schema_version write. Steps must be one version at a time (to_version == from_version + 1); the runner chains them.

Step authors: read the module docstring’s rules — no executescript / COMMIT / ROLLBACK inside a step, preserve rowids on table rebuilds, and rebuild the interval index with rebuild_annotations_rtree() if the annotations table was rebuilt.

Re-registering the same (store_kind, from_version) pair replaces the previous entry.

lacing.registered_processors() → list[str]

Names of every registered processor, sorted.

lacing.render_artifact_exhibit(annotations: Iterable, *, out_dir: str | Path, formats: Sequence[str] = ('html', 'pdf', 'md'), title: str = 'Artifact exhibit', image_resolver: Callable[[Mapping], str | None] | None = None) → list[Path]

Render an annotation graph as a human-readable artifact exhibit.

Lays every annotation out as a card — body, panel images, and in-document hyperlinks to the artifacts it derives from / feeds into. HTML is authored; the PDF and Markdown derive from it.

Panel images are written once as content-addressed sibling files under <out_dir>/images/ and referenced relatively, so the HTML and Markdown stay small; the PDF embeds them and stays a single self-contained file. The images/ directory is created only when the graph actually has images.

Parameters:

• annotations – the lacing annotations to exhibit (any iterable). Order is preserved — pass them chain-ordered for a document that reads top-to-bottom.

• out_dir – directory the exhibit.{html,pdf,md} files land in.

• formats – which of html / pdf / md to write. The HTML is always built in memory (the others derive from it).

• title – document title.

• image_resolver – image-reference → local file path callback. Defaults to reading the reference’s own path field; a caller whose images live behind URLs passes a resolver that downloads / caches them (keeping this module media-agnostic).

Returns:

The written file paths.

Raises:

RuntimeError – when pdf / md is requested but its optional converter (weasyprint / dn) is not installed — the message names the install command.

lacing.replay_oplog(log: OpLog, *, until_clock: int | None = None, target_factory=None) → Any

Rebuild a store by replaying log up to (and including) until_clock.

Parameters:

• log – Source op-log.

• until_clock – Stop at this clock (inclusive). None = replay all.

• target_factory – Zero-arg callable returning a fresh empty store. Defaults to MemoryStore. Pass a SqliteStore factory to replay into a persistent file.

Returns:

The rebuilt store. Operations whose payload references unknown body schemas or tier parents are still applied; the caller is responsible for any post-replay validation.

async lacing.run_processor_async(name: str, *, store: Any, oplog: Any, **kwargs: Any) → Any

Run a processor in the current event loop. For async callers.

lacing.run_processor_sync(name: str, *, store: Any, oplog: Any, **kwargs: Any) → Any

Run a processor synchronously and return its result.

If the processor is async, we call it via asyncio.run (when no loop is running) or schedule and wait on it (when a loop is already active). Most callers from sync code want the former.

lacing.traced(tracer: Any, span_name: str | None = None, *, record_args: bool = False)

Decorator: wrap a function in a span on tracer.

Parameters:

• tracer – From get_tracer().

• span_name – Override the span name. Default: func.__qualname__.

• record_args – If True, attach (str-coerced) positional + keyword args as span attributes arg.<n> / kwarg.<name>. Off by default — args may contain large or sensitive data.

lacing.validate_body(body: dict, uri: str) → BaseModel

Validate body against the schema registered for uri.

Returns the parsed Pydantic instance. Raises BodySchemaError on validation failure (wrapping the underlying pydantic.ValidationError) or UnknownBodySchemaError if the URI isn’t registered.