API reference¶

Generated from the docstrings in chaski's source.

Service ¶

Service(name: str, mount: str = '', *, node: Any = None, display_name: str | None = None, description: str | None = None, version: str | None = None, logs: bool = True, state_dir: Path | None = None, mqtt_port: int | None = None, api_port: int | None = None, metadata: dict[str, Any] | None = None, architecture_metadata: dict[str, Any] | None = None, health_metrics: Iterable[HealthMetricDeclaration] | None = None, max_queued_messages: int = 0, clock: Clock | None = None, step_dependencies: list[str] | None = None)

A publisher on either the local or the external door — one implementation, one constructor. See the module docstring for the lifecycle.

node says who this Service is to Colca: None (default, inside a deployment), a node= URL string (outside one — the identity is loaded or minted under state_dir), or a :class:LocalDoor (an embedded node's own local door — internal, chaski.Node.service() only).

mqtt_port/api_port are for a node whose published doors sit behind a non-standard port (a test harness, a port-mapped deployment).

metadata and architecture_metadata ride the retained _ServiceDetails record as given: what an editor shows about this service beyond its health — protocol, icon, the driver behind it. architecture_metadata is merged under the live status/detail :meth:status writes.

health_metrics names the Prometheus metrics that describe this service's health (_ServiceDetails.health_metrics); a container passes colca_data_contracts.container_resource_health_metrics().

Construction does not dial the node; it only loads or mints an external identity on disk. See :meth:start.

node_id `property` ¶

node_id: str | None

The node this service is registered at — level 4 of every topic it writes. None before :meth:start.

service_id `property` ¶

service_id: str | None

This service's identity at the node: the registry ULID minted at self-registration (local) or the pinned one (external) — what the catalogue's connector field carries. None before :meth:start.

mount `property` ¶

mount: str

Where the node currently has this service (re-resolved on every reconnect for a local service).

cursor_prefix `property` ¶

cursor_prefix: str

The cursor namespace this identity owns, which :meth:stream prepends: c/{name}/ for a local service, {ulid}/ for an external one. The door refuses cursors outside it.

start ¶

start(*, connect_timeout: float = 10.0) -> Service

Connect, self-register (local) or authenticate (external), publish the initial retained _ServiceDetails, and subscribe to this service's own _Signal records. Idempotent — a second call on an already-started Service is a no-op.

Outside a deployment, a refused CONNECT (the identity is not yet enrolled) raises :class:NotEnrolled — see :meth:wait_enrolled to poll instead of raising once.

is_broker_connected ¶

is_broker_connected() -> bool

Whether the MQTT door is currently reachable — the fact a health endpoint reports. False before :meth:start; afterwards the real socket state, including an outage the caller is buffering against.

enroll_hint ¶

enroll_hint() -> str

The exact one-liner an operator runs to enroll this identity (also the text of a raised :class:NotEnrolled). Outside a deployment only.

wait_enrolled ¶

wait_enrolled(timeout: float = 60.0, *, poll_interval: float = 2.0) -> Service

Poll start() — reconnecting — until the operator enrolls this identity, or timeout elapses. Outside a deployment only.

publish ¶

publish(path: str, value: Any, *, unit: str | None = None, timestamp: Any | None = None) -> None

Publish one sample at path.

The first call for a never-seen path mints a DataTag and grows the catalogue (republished immediately); every call resolves the Signal the node minted for that tag from this service's own _Signal subscription. Before that Signal exists, the sample is buffered (bounded) and flushed the moment it binds.

send ¶

send(topic: str, payload: str, *, retain: bool = False) -> None

Publish one record at topic on this service's MQTT session, at QoS 1, and wait for the node's PUBACK.

payload is the record as a JSON string. retain for a state record. A record the node refuses raises franzmq.errors.PublishRejected; no PUBACK in time raises PublishTimeout. Must not be called from an MQTT callback, which cannot wait for its own PUBACK.

retract ¶

retract(topic: str) -> None

Retire the state record at topic: an empty retained payload, which the node keeps as a tombstone and drops from its KV.

command ¶

command(contract: str, path: str, fields: dict[str, Any] | None = None, *, timeout: float = 30.0) -> dict[str, Any]

Send one command to path on this service's node and return its _Ack as the executor wrote it (result_code, message, and whatever else it carries, such as state_writes). See :meth:chaski.command.CommandSender.command.

kv ¶

kv(prefix: str = '', *, contract: str | Iterable[str] | None = None) -> list[KvEntry]

A snapshot of the node's retained state under prefix — every page of GET /kv followed to the end — optionally narrowed to one or more uns contracts (contract="_Signal", contract=["_SystemElement", "_Group"]), which the node filters before decoding any payload. Only entries this identity may read are returned. Requires :meth:start.

stream ¶

stream(name: str, *, cursor: str | None = None, max: int = 1000, signal_ids: Iterable[str] | None = None) -> Stream

A named, durable cursor over the node's stream name (metrics, annotations, alarms, ...) — see :class:chaski.door.Stream for the fetch → process → ack contract.

cursor names the cursor within :attr:cursor_prefix and defaults to the stream's name. Pass another name to follow a stream twice or to start fresh (svc.stream("metrics", cursor="ingest-02")), and retire the old one with Stream.retire(). max bounds one page. signal_ids filters the metrics stream at the door. Requires :meth:start.

pending ¶

pending() -> list[tuple[str, str]]

(path, reason) for every published path with no bound Signal yet — see :func:_pending_reason for what each reason means.

report_progress ¶

report_progress(processed_at: float, *, force: bool = False) -> bool

Report application progress at most once per real second.

This is control/health state, not a business or historian fact. A simulation must report what it processed, not just its target clock. Reporting is best effort: a broker outage must not abort completed work. Returns whether the status was published. Business progress must already be checkpointed by the caller before calling this method.

status ¶

status(ok: bool, detail: str = '') -> None

Republish _ServiceDetails with architecture_metadata.status healthy/unhealthy (+detail) — what a health view reads.

close ¶

close() -> None

Seal the catalogue (any known path not published this run goes stale), republish _ServiceDetails with is_active=False, and disconnect. Registration and ids stay. Idempotent.

retire ¶

retire(token: str | None = None) -> None

Remove this service for good: publish empty retained _ServiceDetails and _DataTags so the tree forgets it. Outside a deployment it also revokes the enrollment (DELETE /enroll), which needs token. The context manager never calls this.

ConnectorService ¶

ConnectorService(name: str, mount: str = '', *, driver: Driver, interval: float = 1.0, heartbeat_interval: float = 5.0, max_pending: int = 10000, reconnect_retries: int = 5, outage_reminder: float = 300.0, summary_interval: float = 60.0, telemetry: Telemetry | None = None, timestamp_source: Literal['acquisition', 'source'] = 'acquisition', **service_kwargs: Any)

Bases: Service

A Service that discovers tags through a :class:Driver and polls them. Construct it like a :class:~chaski.Service (the same node= rule decides the door), then :meth:run it.

interval is the poll cadence in seconds; heartbeat_interval how often the synthetic heartbeat flips; max_pending the bound on metrics kept through a broker outage; reconnect_retries the source reconnect attempts per outage before the loop stays alive and tries again next poll; outage_reminder how many seconds a broker outage may last before it is logged again (the first failure and the recovery are always logged); summary_interval the cadence of the [DATA] throughput line.

run ¶

run(*, health_port: int | None = 8888) -> None

Serve until stopped: :func:run with this service.

serve `async` ¶

serve() -> None

Register, discover, poll. Returns when :meth:stop is called; raises if the node cannot be reached at all (a connector without a node has nothing to do — the container restarts it).

Driver ¶

Driver(*, logger: Logger | None = None)

The protocol half of a connector. Subclass it, implement the four async methods, hand an instance to :class:ConnectorService.

connect `async` ¶

connect() -> None

Open the source. Raise on failure, with a message naming what failed (host, port, endpoint) — the loop logs it and retries.

discover `async` ¶

discover() -> Discovery

Discover the source's tags. Called after connect (or without it, when catalogue_requires_connection is False).

read `async` ¶

read(targets: list[Target]) -> Iterable[tuple[Topic, Any, SignalRecord] | Reading]

One poll of targets. A target a driver could not read is simply absent from the result; a lost source raises :class:SourceDisconnectedError.

close `async` ¶

close() -> None

Release the source. Must tolerate being called on a half-open or already-closed connection.

DataOpsService ¶

DataOpsService(name: str, mount: str = '', *, node: Any = None, data_dir: Path | None = None, retention: float | None = None, historian: Historian | None = None, poll_interval: float = 1.0, trim_interval: float = 3600.0, health_port: int = health.PORT_DEFAULT, **service_kw: Any)

Bases: Service

A :class:chaski.Service that runs :class:~chaski.dataops.Producer classes. Everything the base class does works unchanged (node=, registration, logs, kv(), stream()); this adds the producer runtime described in the module docstring.

::

import chaski
from chaski.dataops import Producer, SignalRangeInput, SignalOutput, every

class Oee(Producer):
    name = "oee"
    system_element_name = "press3"
    speed = SignalRangeInput("speed", window="1h")
    oee = SignalOutput("oee", data_type="float", description="Rolling OEE")

    @every("1m")
    async def tick(self):
        frame = self.speed.fetch(self.watermark, now())
        self.oee.publish(compute_oee(frame))
        self.advance_watermark(now())

svc = chaski.DataOpsService("analytics", mount="site1/line1")
svc.add(Oee)
svc.run()

data_dir holds the buffer (default: the service's own state directory, ~/.colca/services/<name>); retention is the broker's metrics retention in seconds, what a declared window is validated against; historian an optional read-only :class:~chaski.dataops.inputs.Historian; poll_interval/ trim_interval the ingest poll and buffer-trim cadences; health_port the health door (0 for an ephemeral port). Every other keyword is the base class's.

producers `property` ¶

producers: list[type[Producer]]

The producer classes this service runs, sorted by name.

door `property` ¶

door: Door

The door every input resolves through — the base class's own, open after :meth:start. Outputs write with :meth:send.

buffer `property` ¶

buffer: Buffer

The one local state, open after :meth:start.

add ¶

add(producer_cls: type[Producer]) -> DataOpsService

Run producer_cls. Explicit registration — the counterpart of :meth:discover. A second class with the same name replaces the first (one producer per name, the buffer keys watermarks by it). Returns self for chaining.

discover ¶

discover(package: str) -> int

Import package (and every submodule under it) and run every concrete producer it defines. Returns how many were added.

discover_directory ¶

discover_directory(path: Path) -> int

Import every top-level *.py under path (see :func:import_directory) and run every concrete producer those files define. Returns how many were added.

start ¶

start(*, connect_timeout: float = 10.0) -> DataOpsService

The base class's :meth:~chaski.Service.start, then open the buffer. Idempotent.

instantiate ¶

instantiate() -> list[Producer]

Instantiate every added producer and attach it to this runtime — without running setup(); :meth:serve does that. A producer whose constructor raises is skipped with a logged reason.

bind_outputs ¶

bind_outputs(instances: list[Producer]) -> dict[str, str]

Catalogue every declared SignalOutput and bind every AnnotationOutput on instances. Returns the catalogue's {source: tag_id}. Requires :meth:start.

serve `async` ¶

serve(stop: Event | None = None) -> None

Run the service on the current event loop until stop is set — see the module docstring for the startup order. :meth:run is the blocking wrapper with signal handling.

run ¶

run() -> None

Block: :meth:serve on a fresh event loop until SIGINT/SIGTERM.

Node ¶

Node(name: str, *, parent: str | tuple[str, str] | None = None, data_dir: str | Path | None = None, retention: str | None = None, log_level: str = 'info', binary: str | None = None, contracts_bundle: str | None = None)

An embedded colca node: colcad as a supervised subprocess.

parent is None (root / not yet enrolled), a bare URL (trust-on-first-use), or (url, pubkey) (an explicit pin — see the module docstring). retention is a Go duration string (e.g. "336h") applied to the node's metrics stream retention; None leaves colcad's own default in force.

start ¶

start(*, timeout: float = 30.0) -> Node

Write config, launch colcad, and wait for its local /healthz.

status ¶

status() -> NodeStatus

wait_enrolled ¶

wait_enrolled(timeout: float = 60.0) -> None

Block until status() is "enrolled", or raise. A TimeoutError carries the current status's detail (the enroll hint, most often); a colcad crash while waiting raises NodeCrashed immediately rather than waiting out the deadline for a process that is never coming back.

enroll_hint ¶

enroll_hint(*, mount: str | None = None, token_env: str = 'COLCA_ADMIN_TOKEN') -> str

The command the parent's operator runs once to enroll this node. The node's API door is loopback-only, so the hint carries the key.

retire_hint ¶

retire_hint(*, token_env: str = 'COLCA_ADMIN_TOKEN') -> str

The one-liner that removes this node from its parent's registry. Deleting the local data directory (this node's identity and its durable streams) is deliberately the integrator's own act, not something this method does.

service ¶

service(name: str, mount: str = '', *, clock: Clock | None = None, step_dependencies: list[str] | None = None) -> Service

A local Service on this node's own local door.

Stream ¶

Stream(door: Door, name: str, cursor: str, *, max: int = 1000, signal_ids: Iterable[str] | None = None)

A named cursor over one colca stream, as Service.stream() returns it.

The cursor lives at the door: /fetch reads from its stored position and only /ack moves it. Reopening the same name resumes where it was acked; a new name starts at the stream's first retained record.

Iterating drains the stream, acking page by page. for record in stream: yields records in stream order and acks a page only when the consumer asks for the record after its last one. A consumer that raises or stops mid-page gets that page again next time, so handlers must be idempotent. Iteration stops at the first empty page.

:meth:ack commits before the page boundary if needed. :meth:follow repeats the drain, sleeping poll_interval after an empty page. A pruned range (Page.gap) is logged as a warning.

fetch ¶

fetch() -> Page

One page from the cursor's stored position. Never moves the cursor.

ack ¶

ack(upto: Record | int) -> bool

Ack upto (a record, or its offset) as the last PROCESSED position. Returns whether the cursor moved.

retire ¶

retire() -> None

Delete this cursor at the door — idempotent, also when it never existed. A later fetch under the same name starts over.

drain ¶

drain() -> Iterator[Record]

Yield every record from the cursor's position to the head, page by page, acking each page after its records were consumed (see the class docstring). Stops at the first empty page.

follow ¶

follow(*, poll_interval: float = 1.0, stop: Event | None = None) -> Iterator[Record]

:meth:drain forever — after an empty page, sleep poll_interval (waking early when stop is set) and drain again. Ends when stop is set.

NotEnrolled ¶

NotEnrolled(name: str, node_url: str, command: str)

Bases: RuntimeError

Raised by :meth:Service.start when the node refuses this identity's CONNECT — outside a deployment only. The message says how an operator enrolls it; also available bare via :meth:Service.enroll_hint.

API reference¶

Service ¶

node_id property ¶

service_id property ¶

mount property ¶

cursor_prefix property ¶

start ¶

is_broker_connected ¶

enroll_hint ¶

wait_enrolled ¶

publish ¶

send ¶

retract ¶

command ¶

kv ¶

stream ¶

pending ¶

report_progress ¶

status ¶

close ¶

retire ¶

ConnectorService ¶

run ¶

serve async ¶

Driver ¶

connect async ¶

discover async ¶

read async ¶

close async ¶

DataOpsService ¶

producers property ¶

door property ¶

buffer property ¶

add ¶

discover ¶

discover_directory ¶

start ¶

instantiate ¶

bind_outputs ¶

serve async ¶

run ¶

Node ¶

start ¶

status ¶

wait_enrolled ¶

enroll_hint ¶

retire_hint ¶

service ¶

Stream ¶

fetch ¶

ack ¶

retire ¶

drain ¶

follow ¶

NotEnrolled ¶

node_id `property` ¶

service_id `property` ¶

mount `property` ¶

cursor_prefix `property` ¶

serve `async` ¶

connect `async` ¶

discover `async` ¶

read `async` ¶

close `async` ¶

producers `property` ¶

door `property` ¶

buffer `property` ¶

serve `async` ¶