s6.utils package

Submodules

s6.utils.bounding_box_selector module

class s6.utils.bounding_box_selector.BoundingBoxSelector(color=(0, 255, 0), border_width=2, show_text=True)

Bases: object

Interactive bounding box selector using OpenCV window.

Example

selector = BoundingBoxSelector(color=(0,255,0), border_width=2, show_text=True) bbox2d = selector.select(image)

Parameters:

  • color – BGR tuple for rectangle color.

  • border_width – Thickness of the rectangle border.

  • show_text – Whether to display x,y,w,h text on the rectangle.

select(image: ndarray, window_name: str = 'Select BoundingBox') → BoundingBox2D

Show image in a window and let user draw a bounding box by dragging the mouse. Press Enter to confirm selection and close the window.

Parameters:

  • image – The image array to select bounding box from.

  • window_name – The name of the OpenCV window.

Returns:

The selected bounding box.

Return type:

BoundingBox2D

s6.utils.camera_identifier module

Rule‑based camera identifier using simple IR marker rectangular ROIs.

Provides CameraIDIdentifier to determine camera IDs by checking predefined regions for bright markers in grayscale frames. A small OpenCV UI in main() helps author a JSON config of ROIs and a detection threshold.

The expected input frames are 8‑bit single‑channel images with intensity values in [0, 255]. Regions are rectangles and may be marked as negative (region must not contain a marker).

class s6.utils.camera_identifier.CameraIDIdentifier(rule_config, threshold=250)

Bases: object

Identify camera IDs by testing configured regions for bright markers.

The identifier loads a rule_config mapping camera IDs to a list of region specs. Each spec describes a rectangle with bbox=(x1, y1, x2, y2), plus an optional negate flag: - negate=False → region must contain a marker (mean > threshold) - negate=True → region must not contain a marker (mean <= threshold)

Parameters:

  • rule_config (dict[int, list[dict]]) – Mapping from camera ID to a list of region specs. See above for keys.

  • threshold (float, default 250) – Mean intensity threshold in [0, 255] used to decide if a marker is present inside a region.

static from_json(config_path: str = 'configs/camera_ident.config.json') → CameraIDIdentifier

Create an identifier from a JSON configuration file.

Parameters:

config_path (str, default "configs/camera_ident.config.json") – Path to the JSON config. Must contain threshold and rule_config keys.

Returns:

Instance configured with the given rules and threshold.

Return type:

CameraIDIdentifier

Raises:

  • FileNotFoundError – If the file does not exist.

  • json.JSONDecodeError – If the file is not valid JSON.

static load_default() → CameraIDIdentifier

Load the default config from configs/camera_ident.config.json.

order(images: list[ndarray], erase_marker_region: bool = False) → dict

Identify camera IDs and return a mapping from ID to image.

The input order is left unchanged. For each image, the first camera ID whose region specs are satisfied is assigned. If erase_marker_region is True, positive regions for matched specs are zeroed in‑place.

Parameters:

  • images (list[ndarray]) – Grayscale frames of shape (H, W).

  • erase_marker_region (bool, default False) – If True, zero out detected positive regions for matched IDs.

Returns:

Mapping of camera ID to its corresponding image.

Return type:

dict[int, ndarray]

s6.utils.camera_identifier.main()

Interactive ROI authoring tool for camera identification.

Connects to a running s6.utils.streamer.VideoStreamClient, lets users draw rectangle ROIs per camera role via mouse, and writes configs/camera_ident.config.json with threshold and rule_config. Use keys: ‘m’ toggle negate, ‘r’ cycle role, ‘d’ change displayed camera, ‘c’ clear ROIs, ENTER/SPACE to finish.

s6.utils.camera_identifier_aruco module

Camera ID identifier using ArUco marker presence within rectangular regions.

This module provides CameraIDIdentifierAruco, a drop-in alternative to the intensity-ROI based s6.utils.camera_identifier.CameraIDIdentifier.

Instead of thresholding brightness, it detects ArUco markers from the cv2.aruco.DICT_4X4_50 dictionary and decides camera IDs by checking whether detected marker positions fall inside user-configured rectangular regions. Each configured rectangular region can be positive (must contain at least one detected ArUco center) or negative (must contain none) via negate. Cameras without any defined regions are treated as fallbacks and are assigned after region-constrained cameras are matched (useful when only some views contain ArUco markers).

The expected input frames are 8-bit single-channel images (H, W). The interface mirrors CameraIDIdentifier: from_json, load_default and order.

class s6.utils.camera_identifier_aruco.CameraIDIdentifierAruco(rule_config: Dict[Any, List[Dict[str, Any]]], dictionary_id: int | None = None)

Bases: object

Identify camera IDs using ArUco detections within configured regions.

Parameters:

  • rule_config (dict[str | int, list[dict]]) – Mapping from camera ID (e.g., “R”, “B”, “L” or integer) to a list of rectangular region specs of the form {“type”: “rect”, “bbox”: (x1, y1, x2, y2), “negate”: bool}.

  • dictionary_id (int, default cv2.aruco.DICT_4X4_50) – ArUco dictionary used for detection.

static from_json(config_path: str = 'configs/camera_ident.config.json', dictionary_id: int | None = None) → CameraIDIdentifierAruco

Create an identifier from a JSON configuration file.

The JSON is expected to contain a top-level rule_config mapping camera IDs to ROI specs (see class docstring). Any threshold value present in the file is ignored.

static load_default() → CameraIDIdentifierAruco

Load default config from configs/camera_ident.config.json.

Reuses the same ROI config file as the intensity-based identifier. The threshold entry is ignored.

order(images: List[ndarray]) → Dict[Any, ndarray]

Identify camera IDs for images and return mapping id -> image.

Cameras with defined rectangular regions are matched first: all regions for a camera must contain at least one marker. Remaining images are then assigned to cameras without defined regions (fallbacks). If erase_marker_region is True, detected markers in matched regions are erased in-place by filling the marker polygon.

s6.utils.cleartext module

class s6.utils.cleartext.ClearText(font='/Users/tim/Workspace/sense-core/src/s6/utils/fonts/arialbd.ttf', fontsize=24, fontsize_alt=16, **kwargs)

Bases: object

DEFAULT_FONT = '/Users/tim/Workspace/sense-core/src/s6/utils/fonts/arialbd.ttf'
text_block(image, position, text, bgra=(255, 255, 255, 255), alt=False, normalized_position=False, background=(255, 255, 255), blending_opacity=1.0)
class s6.utils.cleartext.TextAlign

Bases: object

BOTTOM = 5
CENTER = 3
LEFT = 1
RIGHT = 2
TOP = 4

s6.utils.datapipe module

Torch-compatible Dataset around StructuredDataset.

Provides datakey-based selection of nested fields and optional balancing across one or more StructuredDataset directories.

Example

Suppose you have a StructuredDataset stored in “./temp” with samples like:

{“image”: <np.ndarray>, “label”: 0} {“image”: <np.ndarray>, “label”: 1}

You can wrap it as a PyTorch Dataset:

from s6.utils.datapipe import StructuredDatasetTorch
from torch.utils.data import DataLoader

dataset = StructuredDatasetTorch(
    "./temp",                  # path to one or more base dirs
    datakeys=["image", "label"],
    balance=False,              # no balancing across dirs
    shuffle=True,               # shuffle entries
    seed=42,                    # reproducible
)
loader = DataLoader(dataset, batch_size=8, shuffle=False)
for images, labels in loader:
    # images: torch.Tensor of shape [8, ...]
    # labels: torch.Tensor of shape [8]
    ...
class s6.utils.datapipe.StructuredDatasetTorch(dataset_dirs: str | List[str], datakeys: List[str], balance: bool = True, shuffle: bool = True, seed: int | None = None)

Bases: Dataset

A torch.utils.data.Dataset wrapper over one or more StructuredDataset directories.

It extracts specific nested fields (via datakey strings) and returns them as torch.Tensor when possible, falling back to Python values otherwise.

Supports balancing across multiple StructuredDataset sources and initial shuffling.

Nested fields are specified with dot-list syntax, e.g.: ‘a.b[0].c’ → sample[‘a’][‘b’][0][‘c’] or getattr if it is a model.

Parameters:

  • dataset_dirs (str or List[str]) – Path or list of paths to one or more StructuredDataset directories.

  • datakeys (List[str]) – List of datakey strings indicating which fields to extract per sample.

  • balance (bool, default=True) – If True and multiple dirs are given, undersample larger datasets to match the smallest one. If False, include all samples sequentially.

  • shuffle (bool, default=True) – If True, shuffle the index mapping once upon initialization.

  • seed (Optional[int], default=None) – Optional random seed for reproducible shuffling.

Return type:

Single torch.Tensor if only one datakey is provided, else a tuple of Tensors/values.

Example

from s6.utils.datapipe import StructuredDatasetTorch

# Wrap a single StructuredDataset directory and extract ‘image’ & ‘label’ ds = StructuredDatasetTorch(

‘./temp’, datakeys=[‘image’, ‘label’], balance=False, shuffle=True, seed=0,

) # Now ds[i] → (image_tensor, label_tensor) img, lbl = ds[0] print(img.shape, lbl.item())

property line_data: List[dict]

List of raw JSON records (unrestored) for all underlying StructuredDatasets.

s6.utils.datastore module

Directory‑backed storage for heterogeneous samples.

This module provides StructuredDataset, a minimal container that stores one sample per line in data.jsonl under a given directory. Larger 2D/3D NumPy arrays can be saved as JPEG images on disk and referenced from the JSON record, and Pydantic models are serialized with enough metadata to be reconstructed on read.

Intended for quick experiments and logs — not a database. Uses Pillow for image I/O and dynamic imports to rebuild Pydantic models when possible.

class s6.utils.datastore.StructuredDataset(base_dir: str, min_size: int = 32)

Bases: object

Small dataset that reads/writes heterogeneous samples in a folder.

Each call to write() appends a JSON object to data.jsonl. NumPy arrays of shape (H, W[, C]) with min(H, W) >= min_size are stored as JPEGs; smaller arrays or higher‑dimensional arrays are embedded as JSON lists with shape hints. Pydantic models are stored with their module and class for later reconstruction.

Parameters:

  • base_dir (str) – Directory that holds data.jsonl and any image files.

  • min_size (int, default 32) – Minimum height/width required to store arrays as images.

jsonl_path

Full path to the backing JSON Lines file.

Type:

str

line_data

In‑memory cache of parsed JSON lines.

Type:

list[dict]

delete_one(idx: int) → None

Delete a sample by index and remove any referenced images.

Parameters:

idx (int) – Sample index; negative values are supported (-1 is the last).

Raises:

IndexError – If the dataset is empty or the index is out of range.

property line_data: List[dict]

Read‑only list of in‑memory JSON records.

write(data: Any)

Append a single sample to the dataset.

Parameters:

data (Any) – Heterogeneous structure composed of dicts/lists/scalars, NumPy arrays, or Pydantic models. Arrays may be saved as images based on min_size.

s6.utils.devices module

s6.utils.entrypoint_loader module

Dynamic entrypoint discovery and dispatch for Sense Core apps.

This utility discovers Python modules and shell scripts under a package root and dispatches subcommands to either module.main() or an executable script. It also provides --list to enumerate available commands or functions within modules. The default root is s6.app.

class s6.utils.entrypoint_loader.EntryPointLoader(module_prefix='s6.app', exclude_prefixes=None)

Bases: object

Discover and invoke subcommands under a package tree.

Parameters:

  • module_prefix (str, default "s6.app") – Dotted package path that serves as the root for discovery.

  • exclude_prefixes (list[str] | None) – Optional list of filename prefixes to exclude when listing commands. The comparison is applied to the bare module name (without .py).

get_module_functions(module)

Return public callables defined on module.

Parameters:

module (ModuleType) – Imported Python module.

Returns:

Names of callables that do not start with __.

Return type:

list[str]

list_commands(directory)

List available commands and subdirectories under directory.

Parameters:

directory (str) – Filesystem path to scan.

Returns:

Two lists: module/script basenames and subdirectory names.

Return type:

tuple[list[str], list[str]]

load_module(module_name)

Dynamically import a module by name.

Parameters:

module_name (str) – Fully-qualified Python module name.

Returns:

The imported module object.

Return type:

ModuleType

run()

Parse CLI args and dispatch to a module or script.

s6.utils.filters module

Generic temporal smoothing filters.

This module provides a lightweight sliding-window smoother that accepts lists/tuples/arrays of arbitrary length and returns a moving-average value. An optional acceptance function allows rejecting outlier measurements before they are added to the window, which is useful for stabilizing detectors.

class s6.utils.filters.TupleSmoother(window: int = 3, accept_fn: ~typing.Callable[[~numpy.ndarray, ~numpy.ndarray], bool] | None = None, dtype: ~numpy.dtype = <class 'numpy.float64'>)

Bases: object

Sliding-window smoother for numeric tuples/lists/arrays.

  • Accepts 1D numeric sequences (list/tuple/ndarray) of any length.

  • Maintains a fixed-size history of accepted measurements.

  • Returns the moving average over the history as the smoothed value.

  • Supports an optional accept_fn(new, prev) -> bool predicate to reject outliers before they are incorporated into the history.

property has_value: bool
property history: List[ndarray]

Return a list copy of internal history arrays (oldest..newest).

reset() → None
update(value: Sequence[float]) → Tuple[ndarray, bool]

Push a new measurement and return (smoothed_value, accepted).

The measurement is only appended to history if accept_fn allows it (or if there is no prior history). The smoothed value is the moving average over the current history window.

property value: ndarray | None

Return the current smoothed value (moving average) or None if empty.

property window: int

s6.utils.infra module

Infrastructure gateway server for database and file operations.

Provides a Gateway class that encapsulates both the FastAPI application (with DB CRUD, aggregation, and file-proxy endpoints) and a matching Python client for easy use.

class s6.utils.infra.Gateway(config: Any)

Bases: object

property app: FastAPI

Build and return the FastAPI application with DB CRUD and file proxy routes.

property client: GatewayClient

Return a Python client for this gateway using the same routes for DB and files.

static connect(config: Any)

Context manager that launches a local Gateway server on an ephemeral port and yields a GatewayClient configured to localhost on that port.

Usage:
with Gateway.connect(config) as client:

# use client to call the local server

class s6.utils.infra.GatewayClient(gateway: Gateway, base_url: str | None = None)

Bases: object

aggregate(pipeline: list, collection: str | None = None)

Generator that streams documents from an aggregation pipeline via WebSocket on the (default) collection.

create(payload: Dict[str, Any], collection: str | None = None) → str

Insert a document into the (default) collection and return its ID.

delete(doc_id: str, collection: str | None = None) → None

Delete a single document by ID from the (default) collection.

delete_file(filename: str) → None
delete_many(flt: Dict[str, Any], collection: str | None = None) → int

Delete multiple documents in the (default) collection via a JSON filter. Returns the number of documents deleted.

download(doc_id: str, collection: str | None = None) → Any

Fetch and reconstruct a nested structure previously uploaded.

download_file(filename: str) → bytes
get(doc_id: str, collection: str | None = None) → Dict[str, Any]

Fetch a single document by ID from the (default) collection.

post_file(filename: str, fobj: Any, content_type: str) → Any
query(flt: Dict[str, Any], collection: str | None = None) → Any

Query documents in the (default) collection via a JSON filter.

update(doc_id: str, payload: Dict[str, Any], collection: str | None = None) → int

Update a document by ID in the (default) collection and return count.

upload(obj: Dict[str, Any], collection: str | None = None, **kwargs) → str

Serialize and upload a nested structure to the (default) collection. Returns the inserted document ID. Additional top-level kwargs are merged into the stored record.

upload_file(filename: str, fobj: Any, content_type: str) → Any

s6.utils.profiler module

Profiling utilities for Sense Core.

Collects lightweight timing information and exports it in two formats:

  • Chrome/Perfetto Trace Events (via s6.schema.trace.TraceEvent) for interactive timeline inspection.

  • A human‑readable “frames” export that groups nested traces per execution frame and is convenient for offline statistics.

The primary surface is Profiler, which offers context managers for scoped timings (trace), a frame context (frame), a function decorator, and exporters. See process_frames() for computing summary stats from the frames export.

class s6.utils.profiler.Profiler

Bases: object

Simple profiler with nested scopes and frame grouping.

  • trace(name, remark=None) creates a timed scope; users can attach metadata while inside the scope.

  • frame() groups all nested traces to represent a single execution cycle (e.g., a render or processing frame).

  • trace_function decorates a callable to measure its runtime.

  • export writes Chrome Trace Events.

  • export_metrics writes a combined JSON with session metadata, optional per-scope remarks, and a frames list suitable for offline analysis.

The profiler keeps a global event list and a per‑frame buffer when a frame context is active. Timestamps are relative to process start and exported in microseconds for Chrome traces and milliseconds in the human‑readable view.

classmethod disable_logging()

Disable verbose debug logging for scope durations.

classmethod enable_logging()

Enable verbose debug logging for scope durations.

classmethod export(filename='trace.json')

Write Chrome Trace Events to filename.

Parameters:

filename (str, default "trace.json") – Output path for the Chrome/Perfetto trace JSON.

classmethod export_metrics(filename: str)

Export session‑level data and per‑frame traces as JSON.

The output contains: - metadata: Session metadata dict attached via metadata(). - remarks: Optional mapping from full scope path to remark string. - frames: A list where each frame has keys frame_id,

start_time_ms, end_time_ms, duration_ms, and traces.

Parameters:

filename (str) – Path to the output file.

classmethod frame()

Group nested traces into a single execution frame.

All trace events generated while inside the context are captured and exported in the human‑readable frames list by Profiler.export_metrics().

Returns:

A context manager representing the active frame.

Return type:

Profiler._Frame

classmethod metadata(**kwargs)

Attach key–value metadata to the current session.

The data is included under the metadata key in Profiler.export_metrics() and intended for run configuration, environment info, or other contextual details.

classmethod trace(name: str = '', remark: str = None)

Create a timed scope.

Parameters:

  • name (str, default "") – Identifier for the trace scope.

  • remark (str | None, optional) – Optional descriptive label associated with this specific scope instance. Stored in the metrics export under a full scope path.

Returns:

A context manager that records begin/end events and lets callers attach metadata via Profiler._Timer.attach().

Return type:

Profiler._Timer

classmethod trace_function(func)

Decorator that measures the runtime of func.

The function body executes inside Profiler.trace() with the scope name set to func.__name__.

Parameters:

func (Callable) – Function to wrap.

Returns:

A wrapped function with identical signature.

Return type:

Callable

s6.utils.profiler.compute_statistics(durations)

Aggregate descriptive statistics for each event.

For every event key, computes min/max/mean/median, standard deviation, variance, common percentiles, interquartile range, count/total, and the mean of the fastest/slowest 5% samples.

Parameters:

durations (dict[str, list[float]]) – Mapping from event name to a list of durations in milliseconds.

Returns:

A mapping from event name to a statistics dict containing keys such as min_ms, max_ms, average_ms, median_ms, p5_ms, p25_ms, p75_ms, p95_ms, iqr_ms, best_5pct_avg_ms, worst_5pct_avg_ms, count, and total_ms.

Return type:

dict[str, dict[str, float | int | None]]

s6.utils.profiler.get_percentile(sorted_values, p)

Compute a percentile with linear interpolation.

Parameters:

  • sorted_values (list[float]) – Sequence sorted in non-decreasing order.

  • p (float) – Percentile in [0, 100].

Returns:

The interpolated percentile value, or None if the input is empty.

Return type:

float | None

s6.utils.profiler.main()

Example entry-point that prints per-scope timing stats.

Reads frames.log.jsonl from the current working directory, computes statistics via process_frames() and compute_statistics(), and prints a human-readable report. Intended for ad‑hoc inspection.

s6.utils.profiler.process_frames(filename)

Parse a frames metrics file and compute sub-trace durations.

Accepts either the combined JSON produced by Profiler.export_metrics() (with a top-level frames list) or a legacy JSONL file with one frame per line. Durations are computed by matching begin/end events of the same name and depth within each frame.

Parameters:

filename (str) – Path to the metrics file (combined JSON or legacy JSONL).

Returns:

Mapping from event full path to a list of durations in milliseconds.

Return type:

dict[str, list[float]]

s6.utils.r2 module

Thin Cloudflare R2 client (S3‑compatible).

Provides a minimal wrapper around boto3 for listing, uploading, downloading, and deleting objects in Cloudflare R2 using the S3 API. It is deliberately small and explicit, with sensible defaults and safeguards against accidental overwrites.

Key concepts

  • Uses standard S3 credentials and custom endpoint_url for R2.

  • Environment variables can supply credentials/region when arguments are omitted: R2_ACCESS_KEY_ID/AWS_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY/AWS_SECRET_ACCESS_KEY, and optionally R2_REGION_NAME/AWS_REGION/AWS_DEFAULT_REGION.

  • Upload helpers default to no‑overwrite behaviour; pass overwrite=True where supported to replace existing objects.

Motivation

Experiments often produce large datasets and intermediate artifacts that need to be shared between local development machines, remote teammates, and lab hardware. Cloudflare R2 provides inexpensive S3‑compatible object storage without egress fees. Wrapping it behind a tiny, documented interface lets the pipeline move data in and out of a relatively abstracted storage backend without depending on any vendor‑specific SDK beyond the ubiquitous S3 API.

This module aims to be: - Minimal: only the operations we actually use (list, get/put, delete). - Predictable: no silent overwrites by default; explicit overwrite=True

when you intend to replace data.

  • Portable: credentials via env or args; works with R2 and any S3‑compatible endpoint (e.g., MinIO) by changing endpoint_url.

Design

The R2Client is a thin façade over a boto3 S3 client:

  • Construction resolves credentials/region from arguments or environment variables and creates a namespaced S3 client using a custom endpoint_url.

  • Listings use ListObjectsV2 with pagination (ContinuationToken) and optional delimiter semantics when recursive=False. Results are returned as a list of R2Object plus any child prefixes.

  • Upload helpers intentionally refuse to overwrite existing keys unless you pass overwrite=True (for directory uploads). Byte and file uploads raise FileExistsError if the key is present.

  • download_directory recreates the relative structure under a prefix.

  • _exists issues a HEAD request and interprets common not‑found error codes; other errors are re‑raised to the caller.

  • To keep documentation builds importable without cloud dependencies, boto3 imports are guarded. Instantiating R2Client fails fast with a clear ImportError if boto3 is unavailable.

class s6.utils.r2.R2Client(access_key_id: str | None = None, secret_access_key: str | None = None, bucket_name: str = '', endpoint_url: str = '', region_name: str = 'auto', session: Session | None = None)

Bases: object

Minimal S3‑compatible client for Cloudflare R2.

Parameters:

  • access_key_id (str, optional) – Access key ID. When None, resolved from environment variables R2_ACCESS_KEY_ID or AWS_ACCESS_KEY_ID.

  • secret_access_key (str, optional) – Secret access key. When None, resolved from R2_SECRET_ACCESS_KEY or AWS_SECRET_ACCESS_KEY.

  • bucket_name (str, optional) – Target bucket name.

  • endpoint_url (str, optional) – R2 endpoint URL, e.g. "https://<account>.r2.cloudflarestorage.com".

  • region_name (str, optional) – Region name. If set to "auto" (default), resolved from R2_REGION_NAME/AWS_REGION/AWS_DEFAULT_REGION when present.

  • session (boto3.session.Session, optional) – Existing boto3 session to reuse. A new session is created by default.

Examples

Create a client and perform common operations:

client = R2Client(
    access_key_id="...",
    secret_access_key="...",
    bucket_name="my-bucket",
    endpoint_url="https://<account>.r2.cloudflarestorage.com",
)

client.upload_file("local.txt", "folder/remote.txt")
client.download_file("folder/remote.txt", "downloaded.txt")
client.delete_object("folder/remote.txt")
objects, prefixes = client.list(prefix="folder/", recursive=False)
delete_object(key: str, missing_ok: bool = True) → None

Delete an object.

Parameters:

  • key (str) – Object key to delete.

  • missing_ok (bool, optional) – If True (default), succeed even if the object is missing. If False, raise an error when the object does not exist.

Raises:

FileNotFoundError – If missing_ok is False and the object does not exist.

download_directory(prefix: str, local_dir: str, *, max_workers: int = 8, progress: bool = False, overwrite: bool = False) → None

Download all objects under a prefix into a local directory, optionally in parallel.

Preserves the relative structure underneath prefix.

Parameters:

  • prefix (str) – Key prefix to download (e.g., "folder/sub/").

  • local_dir (str) – Destination local directory. Created if missing.

download_file(key: str, local_path: str, *, progress: bool = False) → None

Download an object to a local file.

Parameters:

  • key (str) – Object key in the bucket.

  • local_path (str) – Local filesystem path to write to.

get_object_bytes(key: str) → bytes

Download an object into memory (as bytes).

Parameters:

key (str) – Object key.

Returns:

Object data.

Return type:

bytes

list(prefix: str = '', recursive: bool = True) → tuple[List[R2Object], List[str]]

List objects and pseudo‑directories under a prefix.

Parameters:

  • prefix (str, optional) – Key prefix to list under, e.g., "folder/sub/".

  • recursive (bool, optional) – If True (default), returns all objects under prefix. If False, treats "/" as a delimiter and returns only one level, also returning sub‑prefixes.

Returns:

A pair (objects, prefixes). objects contains metadata for each object; prefixes lists child “directories” (only when recursive=False).

Return type:

(list[R2Object], list[str])

upload_bytes(data: bytes, key: str) → None

Upload raw bytes as an object without overwriting.

Parameters:

  • data (bytes) – Bytes to upload.

  • key (str) – Destination object key.

Raises:

FileExistsError – If an object already exists at key.

upload_directory(local_dir: str, prefix: str, *, overwrite: bool = False, max_workers: int = 8, progress: bool = False) → None

Upload a directory tree under a prefix, optionally in parallel.

Preserves the relative path structure under local_dir. Keys are normalised to use / separators.

Parameters:

  • local_dir (str) – Path to the local directory whose contents to upload.

  • prefix (str) – Destination key prefix (e.g., "folder/sub/").

  • overwrite (bool, optional) – If False (default), raise FileExistsError when a destination key already exists. If True, existing objects are replaced.

  • Parameters (Additional)

  • ---------------------

  • max_workers (int, optional) – Number of worker threads for parallel uploads (default 8).

  • progress (bool, optional) – If True, prints a simple aggregate progress indicator to stderr.

Raises:

  • NotADirectoryError – If local_dir is not a directory.

  • FileExistsError – If overwrite is False and any destination key exists.

upload_file(local_path: str, key: str, *, progress: bool = False) → None

Upload a local file without overwriting existing objects.

Parameters:

  • local_path (str) – Path to the local file.

  • key (str) – Destination object key in the bucket.

Raises:

FileExistsError – If an object already exists at key.

class s6.utils.r2.R2Object(key: str, size: int, last_modified: Any, etag: str)

Bases: object

Lightweight object metadata returned by listings.

key

Object key (path) within the bucket.

Type:

str

size

Object size in bytes.

Type:

int

last_modified

Timestamp of last modification (datetime from boto3). Typing kept generic to avoid a hard dependency in the public API.

Type:

Any

etag

Entity tag (usually an MD5 for non‑multipart uploads).

Type:

str

etag: str
key: str
last_modified: Any
size: int

s6.utils.stage_decorator module

Decorator for pipeline stage functions with context key requirements.

exception s6.utils.stage_decorator.PipelineStageError

Bases: Exception

Base exception for pipeline stage validation errors.

s6.utils.stage_decorator.pipeline_stage(require: List[str], produce: List[str]) → Callable[[...], Any]

Decorator for pipeline stage functions that operate on a context object.

Parameters:

  • require – List of dot-separated paths that must exist in context before running.

  • produce – List of dot-separated paths that must exist in context after running.

s6.utils.streamer module

Streaming utilities for multi-camera capture and transport.

This module includes: - Frame — lightweight JPEG container with header + concatenated payloads. - VideoStreamServer and FastAPIVideoStreamServer — capture

frames, stream over TCP, and expose simple HTTP endpoints for a composite PNG and capture specs.

class s6.utils.streamer.FastAPIVideoStreamServer(device_ids: List[int], capture_width: int = 1280, capture_height: int = 960, capture_fps: int = 50, capture_auto_exposure: int = 1, capture_exposure: int = -9)

Bases: VideoStreamServer

Video stream server exposing simple FastAPI endpoints.

Adds HTTP endpoints: - GET /api/current_frame: PNG composite of the latest frames. - GET /api/spec: capture specification JSON.

async get_current_frame()

Return an image/png response with the latest composite frame.

async get_spec()

Return a JSON response with the capture specifications.

async run()

Start FastAPI in a thread and run the TCP streaming server.

class s6.utils.streamer.Frame(*frames, timestamp)

Bases: object

Container for one or more encoded images with a timestamp.

Frames are serialized as a fixed-size header (see Frame.Header) followed by the concatenated JPEG payloads for each image. Images are encoded with OpenCV’s JPEG encoder at JPEG_QUALITY.

frames

One or more images, typically grayscale (H, W) arrays.

Type:

tuple[np.ndarray, …]

timestamp

Capture time as a UNIX timestamp in seconds.

Type:

float

class Header(crc: int, timestamp: float, reserved: bytes, offsets: list)

Bases: object

Fixed-size header for Frame encoding.

Fields

crcint

Placeholder for checksums (unused, defaults to 0).

timestampfloat

Capture time (seconds since epoch).

reservedbytes

8-byte reserved payload.

offsetslist[int]

Byte sizes of each encoded image payload, up to 8 entries.

crc: int
classmethod decode(header: bytes) → Header

Decode a header from bytes and strip padding zeros in offsets.

encode() → bytes

Encode the header to bytes, padding offsets to length 8.

classmethod nominal_length() → int

Return the encoded header length in bytes.

offsets: list
reserved: bytes
timestamp: float
JPEG_QUALITY = 80
PAYLOAD_SIZE_ENCODING = 'q'
static decode(byte_data: bytes) → Frame

Decode a header+payload byte sequence into a Frame.

Notes

Images are decoded in grayscale mode (cv2.IMREAD_GRAYSCALE).

encode() → bytes

Encode all images and return header+payload bytes.

encode_image(image: ndarray, quality: int = 80) → bytes

Encode a single image as JPEG and return raw bytes.

class s6.utils.streamer.LocalCameraStream(device_ids: List[int], program=None)

Bases: object

Iterate over local cameras in a fixed order and yield Frame.

Parameters:

  • device_ids (list[int]) – OS camera indices to open.

  • program (callable | None) – Optional function applied to each cv2.VideoCapture to set props.

release()

Release all camera capture handles.

class s6.utils.streamer.VideoStreamClient(host='localhost', port=8022)

Bases: object

TCP client that receives encoded Frame messages.

Parameters:

  • host (str, default "localhost") – Server hostname or IP.

  • port (int, default 8022) – Server TCP port.

run()

Blocking demo loop that displays frames in an OpenCV window.

socket_receiver()

Background loop that reads from the socket and enqueues frames.

start()

Start the receiver thread and connect to the server.

stop()

Stop the receiver thread and close the connection.

class s6.utils.streamer.VideoStreamServer(device_ids: List[int], capture_width: int = 1280, capture_height: int = 960, capture_fps: int = 49, capture_auto_exposure: int = 1, capture_exposure: int = -9)

Bases: object

Capture from multiple devices and stream frames over TCP.

Parameters:

  • device_ids (list[int]) – OS camera indices to open.

  • capture_width (int) – Desired capture resolution.

  • capture_height (int) – Desired capture resolution.

  • capture_fps (int) – Target frames per second.

  • capture_auto_exposure (int) – OpenCV value for auto exposure mode.

  • capture_exposure (int) – OpenCV exposure value when auto exposure is disabled.

async capture_frames()

Open devices, capture frames, update cache, and broadcast to clients.

async handle_new_client(reader, writer)

Accept a TCP client and stream encoded frames via a queue.

async monitor()

Periodically log the number of connected TCP clients.

program_camera(cap: VideoCapture)

Apply capture properties to a cv2.VideoCapture handle.

async run()

Override in subclasses to start servers and capture loop.

s6.utils.streamer.take_snapshot(server_url='http://localhost:8000')

Fetch one composite frame and split it into per-camera images.

Parameters:

server_url (str, default “http://localhost:8000”) – Base URL of a running FastAPIVideoStreamServer.

Returns:

List of color images ordered across the horizontal composite.

Return type:

list[np.ndarray]

Raises:

Exception – If fetching or decoding fails.

s6.utils.sync_controller module

Client for an ESP32 PWM sync controller.

Provides SyncControllerClient to query and set duty cycles for three outputs (d0, d1, d2) over a simple HTTP API exposed by an ESP32.

Endpoints

  • GET /status → {"d0": float, "d1": float, "d2": float}

  • GET /set?d0=...&d1=...&d2=... → updates duties and returns status

The client caches the last fetched status and refreshes on demand.

class s6.utils.sync_controller.SyncControllerClient(base_url, timeout=5)

Bases: object

Small HTTP client for the ESP32 Sync Controller.

Parameters:

  • base_url (str) – Base URL of the device, e.g., http://192.168.0.50.

  • timeout (float, default 5) – Per‑request timeout in seconds.

property d0

Current duty cycle for d0 in [0.0, 1.0].

property d1

Current duty cycle for d1 in [0.0, 1.0].

property d2

Current duty cycle for d2 in [0.0, 1.0].

refresh()

Force a status refresh from the device.

Raises:

requests.RequestException – If the HTTP request fails.

set_d0(value)

Shortcut for set_duty('d0', value).

set_d1(value)

Shortcut for set_duty('d1', value).

set_d2(value)

Shortcut for set_duty('d2', value).

set_duty(pin, value)

Set a single duty cycle while preserving the others.

Parameters:

  • pin ({'d0', 'd1', 'd2'}) – Which output to update.

  • value (float) – Duty cycle in [0.0, 1.0].

Returns:

JSON response returned by the device after updating.

Return type:

dict

Raises:

  • ValueError – If pin is invalid or value is out of range.

  • requests.RequestException – If the HTTP call fails or returns a non‑2xx status.

property status

Return cached status, fetching from the device if needed.

Returns:

Mapping with keys "d0", "d1", "d2" and float values in [0.0, 1.0].

Return type:

dict

s6.utils.tensor_registry module

Registry and decorator for custom torch.Tensor conversion.

Provides a decorator to register classes that implement a to_torch() method, and a utility to check registration status.

s6.utils.tensor_registry.is_torch_convertible_type(obj_or_cls)

Return True if the given object or class has been registered for torch conversion.

s6.utils.tensor_registry.register_torch_converter(cls)

Class decorator to register a custom type for automatic torch conversion.

The class must implement a to_torch(self) -> torch.Tensor method.

Usage:

@register_torch_converter class Vector2D:

def __init__(self, x, y):

self.x = x self.y = y

def to_torch(self):

import torch return torch.tensor([self.x, self.y], dtype=torch.float)

s6.utils.test_camera_identifier module

class s6.utils.test_camera_identifier.TestCameraIDIdentifier(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

test_call_identify_and_reorder()
test_check_region_has_marker()
test_erase_marker_region_rect()
test_from_json_and_not_found()
test_region_has_marker_rect()
test_region_has_marker_rect_empty()
test_region_matches_spec()

s6.utils.test_datapipe module

class s6.utils.test_datapipe.TestStructuredDatasetTorch(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

tearDown()

Hook method for deconstructing the test fixture after testing it.

test_balance_two_directories()
test_invalid_datakey_raises()
test_multiple_keys_extraction_and_tensor()
test_non_numeric_keeps_value()
test_shuffle_reproducible()
test_single_key_extraction_tensor()

s6.utils.test_datastore module

class s6.utils.test_datastore.DummyModel(*, a: int, b: str)

Bases: BaseModel

a: int
b: str
class s6.utils.test_datastore.TestStructuredDataset(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

tearDown()

Hook method for deconstructing the test fixture after testing it.

test_delete_one_basic()
test_delete_one_image_cleanup()
test_getitem_and_len()
test_multiple_writes_and_counter()
test_write_and_iterate_large_array()
test_write_and_iterate_simple()
test_write_and_iterate_small_array()
test_write_and_restore_pydantic_model()

s6.utils.test_entrypoint_loader module

class s6.utils.test_entrypoint_loader.TestEntryPointLoader(methodName='runTest')

Bases: TestCase

test_build_parser_parses_list_and_commands()
test_exec_script_calls_execv()
test_exec_script_no_shebang()
test_get_module_functions_returns_callables()
test_handle_list_error()
test_handle_list_nested_module()
test_handle_list_shell_script()
test_handle_list_top_level()
test_init_sets_base_dir()
test_invoke_module_with_and_without_main()
test_list_commands_excludes_specified_prefixes()
test_list_commands_filters_files_and_dirs()
test_load_module()
test_resolve_command_path_module_and_script_and_errors()
test_run_dispatches_correctly()
test_run_with_no_commands_prints_help()

s6.utils.test_infra module

class s6.utils.test_infra.DummyStorageClient

Bases: GatewayClient

Dummy client for testing storage-related upload/download and serialization.

create(record: Dict[str, Any], collection: str | None = None) → str

Insert a document into the (default) collection and return its ID.

download_file(filename)
get(doc_id: str, collection: str | None = None) → Dict[str, Any]

Fetch a single document by ID from the (default) collection.

upload_file(filename, fobj, content_type)
class s6.utils.test_infra.M(*, x: int)

Bases: BaseModel

x: int
class s6.utils.test_infra.TestGatewayClient(methodName='runTest')

Bases: TestCase

make_response(status=200, json_data=None, content=b'data')
setUp()

Hook method for setting up the test fixture before exercising it.

test_create()
test_delete()
test_delete_file()
test_delete_many()
test_download_file()
test_get()
test_post_file()
test_query()
test_update()
test_upload_file()
class s6.utils.test_infra.TestGatewayClientSerializeDeserialize(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

test_deserialize_list_dict()
test_deserialize_model()
test_deserialize_numpy()
test_deserialize_primitive()
test_serialize_list_dict()
test_serialize_model()
test_serialize_numpy()
test_serialize_primitive()
class s6.utils.test_infra.TestGatewayClientStorage(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

test_upload_download_nested()
test_upload_download_numpy()
test_upload_download_primitives()
test_upload_download_pydantic()
class s6.utils.test_infra.TestGatewayConnect(methodName='runTest')

Bases: TestCase

test_connect_contextmanager()

s6.utils.test_profiler module

class s6.utils.test_profiler.TestProfiler(methodName='runTest')

Bases: TestCase

tearDown()

Hook method for deconstructing the test fixture after testing it.

test_enable_disable_logging()

Verify that Profiler.enable_logging() and disable_logging() correctly update the verbosity flag.

test_export_metrics_file()

Test that export_metrics() writes a combined JSON file containing session metadata, remarks, and frame trace data.

test_export_trace_file()
Test the export() method to ensure:

  • A JSON file is written containing a “traceEvents” list and “displayTimeUnit”.

  • A log message is produced that includes the file name.

test_frame_context(mock_perf)
Test the frame() context manager:

  • Ensures that a frame records its start and end times.

  • Checks that trace events generated within the frame are captured.

  • Verifies that the frame is appended to the internal frames list.

test_metadata_collection()

Test that metadata() correctly accumulates session metadata.

test_timer_debug_log()

Test that when logging is set to DEBUG level, the timer context generates a debug log.

test_timer_events(mock_perf)
Test that using the Profiler.trace() context manager produces:

  • A “begin” event with phase “B” when entering.

  • An “end” event with phase “E” when exiting,

and that metadata and timing are recorded correctly.

test_trace_remark_mapping_nested_and_export()

Test nested remark mappings and their inclusion in export_metadata.

test_trace_remark_mapping_simple()

Test that providing a remark to Profiler.trace populates _remark_mapping with the correct full scope path and remark.

s6.utils.test_r2 module

class s6.utils.test_r2.FakeS3Client

Bases: object

delete_object(Bucket, Key, **kwargs)
download_file(bucket, key, filename, *args, **kwargs)
head_object(Bucket, Key, **kwargs)
put_object(Bucket, Key, Body, **kwargs)
upload_file(filename, bucket, key, *args, **kwargs)
class s6.utils.test_r2.FakeSession

Bases: object

client(service_name, **kwargs)
last = None
class s6.utils.test_r2.TestR2Client(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

tearDown()

Hook method for deconstructing the test fixture after testing it.

test_download_directory_creates_files()
test_init_fallbacks_to_aws_env_credentials()
test_init_reads_r2_env_credentials()
test_upload_directory_no_overwrite_raises()
test_upload_directory_with_overwrite_uploads_all()
test_upload_file_and_bytes_no_overwrite()

s6.utils.test_stage_decorator module

s6.utils.test_stage_decorator.test_get_path_nested_dict_list()
s6.utils.test_stage_decorator.test_pipeline_stage_invalid_index()
s6.utils.test_stage_decorator.test_pipeline_stage_missing_produce()
s6.utils.test_stage_decorator.test_pipeline_stage_missing_require()
s6.utils.test_stage_decorator.test_pipeline_stage_success()
s6.utils.test_stage_decorator.test_pipeline_stage_template_key_missing_produce()
s6.utils.test_stage_decorator.test_pipeline_stage_template_key_missing_require()
s6.utils.test_stage_decorator.test_pipeline_stage_template_key_success()

s6.utils.test_streamer module

class s6.utils.test_streamer.DummyCap(idx, frame)

Bases: object

isOpened()
read()
release()
class s6.utils.test_streamer.FakeSocket(message)

Bases: object

close()
connect(addr)
recv(bufsize)
class s6.utils.test_streamer.TestFrameEncodeDecode(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

tearDown()

Hook method for deconstructing the test fixture after testing it.

test_multi_frame_roundtrip()
test_single_frame_roundtrip()
class s6.utils.test_streamer.TestFrameHeader(methodName='runTest')

Bases: TestCase

test_header_encode_decode()
class s6.utils.test_streamer.TestLocalCameraStream(methodName='runTest')

Bases: TestCase

test_iterate_and_release(mock_vcap)
class s6.utils.test_streamer.TestTakeSnapshot(methodName='runTest')

Bases: TestCase

test_take_snapshot(mock_get)
class s6.utils.test_streamer.TestVideoStreamClient(methodName='runTest')

Bases: TestCase

test_socket_receiver(mock_socket_ctor)
test_start_stop_no_data(mock_socket_ctor)

s6.utils.test_sync_controller module

class s6.utils.test_sync_controller.FakeResponse(status=200, json_data=None)

Bases: object

Minimal fake response for testing SyncControllerClient.

json()
raise_for_status()
class s6.utils.test_sync_controller.TestConvenienceSetters(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

test_set_d0(mock_set)
test_set_d1(mock_set)
test_set_d2(mock_set)
class s6.utils.test_sync_controller.TestFetchStatus(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

test_fetch_status_defaults_missing_keys(mock_get)
test_fetch_status_success(mock_get)
class s6.utils.test_sync_controller.TestItemAccess(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

test_getitem_invalid()
test_getitem_valid()
test_properties_return_current_status()
test_setitem_calls_set_duty(mock_set)
test_setitem_invalid()
class s6.utils.test_sync_controller.TestSetDuty(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

test_set_duty_calls_api_and_updates_status(mock_get)
test_set_duty_invalid_pin()
test_set_duty_invalid_value()
class s6.utils.test_sync_controller.TestStatusProperty(methodName='runTest')

Bases: TestCase

test_refresh_forces_new_fetch()
test_status_property_caches()
test_status_property_returns_copy()
class s6.utils.test_sync_controller.TestSyncControllerClientInit(methodName='runTest')

Bases: TestCase

test_init_no_trailing_slash()
test_init_strips_trailing_slashes()

s6.utils.test_waypoint_utils module

class s6.utils.test_waypoint_utils.TestWaypointUtils(methodName='runTest')

Bases: TestCase

setUp()

Hook method for setting up the test fixture before exercising it.

tearDown()

Hook method for deconstructing the test fixture after testing it.

test_parse_file_dispatch_jsonl()
test_parse_file_dispatch_txt()
test_parse_file_unsupported_extension()
test_parse_jsonl_basic()
test_parse_jsonl_invalid_json()
test_parse_jsonl_invalid_values()
test_parse_jsonl_missing_keys()
test_parse_txt_basic()
test_parse_txt_invalid_line()

s6.utils.waypoint_utils module

Waypoint parsing and two-axis path generation utilities.

This module provides:

All distances and offsets are expressed in axis distance units (e.g., mm) as used by s6.robotic.stepper. Generators yield s6.schema.primitives.Vector2D where indicated.

class s6.utils.waypoint_utils.PathDriver(rpc_host: str = '10.20.0.2', rpc_port: int = 8000)

Bases: object

High-level two-axis driver for grid/spiral/random patterns.

Uses s6.robotic.client.RoboticClient to send motion commands to the running REST API. Maintains a simple internal target position used as the origin for relative patterns.

circular_spiral(step_size: float = 0.1, max_radius: float | None = None, dtheta: float = 0.05)

Yield an outward circular (Archimedean) spiral.

Parameters:

  • step_size (float, default 0.1) – Radial growth per revolution (b in r = b * theta).

  • max_radius (float, optional) – Maximum radius at which to stop emission.

  • dtheta (float, default 0.05) – Angular increment in radians per step.

Yields:

Vector2D – Each visited coordinate along the spiral.

property current: Tuple[float, float]

Return the current target coordinates (x, y) in axis units.

grid(nx: int, ny: int, dx: float, dy: float)

Yield a grid sweep of nx × ny points, spaced by dx, dy.

Starting at the current position, this generates a serpentine scan over a rectangle centered at the origin of the grid. Motion commands are relative; axes are not zeroed.

Parameters:

  • nx (int) – Number of samples along x and y.

  • ny (int) – Number of samples along x and y.

  • dx (float) – Spacing between adjacent samples in axis units.

  • dy (float) – Spacing between adjacent samples in axis units.

Yields:

Vector2D – Each visited coordinate in grid order.

list_waypoints()

Fetch the list of named waypoints from the server.

Returns:

Each item contains keys name and positions.

Return type:

list of dict

move(x: float, y: float) → None

Move to an absolute position by zeroing then applying offsets.

Parameters:

  • x (float) – Target coordinates in axis distance units. Values are rounded to integers for the underlying client calls.

  • y (float) – Target coordinates in axis distance units. Values are rounded to integers for the underlying client calls.

random(r: float, repeat: int, waypoint: str | None = None, max_step_size: Vector2D | Tuple[float, float] | None = (0.5, 0.5))

Randomly explore a disc of radius r around a named waypoint.

The driver first moves to waypoint. For each trial, it samples a random offset inside the disc and either moves in micro-steps (bounded by max_step_size per axis) or performs a single jump. It then returns to the waypoint before the next trial.

Parameters:

  • r (float) – Disc radius in axis units.

  • repeat (int) – Number of random targets to sample.

  • waypoint (str, optional) – Name of a waypoint to use as the center. Must exist on the server.

  • max_step_size (Vector2D or (float, float), default (0.5, 0.5)) – Maximum per-step delta along x and y when micro-stepping. If None, performs a single jump to each target.

Yields:

Vector2D – When micro-stepping, yields the cumulative offset from center after each micro-step (forward and return). When jumping, yields the full random offset.

Raises:

ValueError – If the waypoint is not found on the server.

spiral(step_size: float = 1.0, max_radius: float | None = None)

Yield an outward square spiral from the current position.

Parameters:

  • step_size (float, default 1.0) – Distance per step along each leg.

  • max_radius (float, optional) – If provided, stop once the Euclidean distance from the start exceeds this radius.

Yields:

Vector2D – Each visited coordinate.

to_waypoint(name: str) → None

Move to a named waypoint defined in configuration.

Parameters:

name (str) – Waypoint identifier. The internal current is not modified.

zero() → None

Zero both axes and reset the internal position to (0, 0).

class s6.utils.waypoint_utils.WaypointUtils

Bases: object

Parse waypoint files in plain-text or JSONL formats.

Each waypoint is a pair (x, y) representing a 2D offset/position.

classmethod parse_file(file_path: str) → List[Tuple[float, float]]

Parse a waypoint file by extension.

Parameters:

file_path (str) – Path to a file ending with .txt or .jsonl.

Returns:

Sequence of (x, y) pairs.

Return type:

list of tuple[float, float]

Raises:

ValueError – If the file extension is not supported.

classmethod parse_jsonl(file_path: str) → List[Tuple[float, float]]

Parse a JSONL file of objects with {"x": ..., "y": ...}.

Empty lines are ignored. Each line must parse to a JSON object with numeric x and y fields.

Parameters:

file_path (str) – Path to a JSON lines file.

Returns:

Sequence of (x, y) pairs.

Return type:

list of tuple[float, float]

Raises:

ValueError – If a line cannot be parsed as JSON or lacks required fields.

classmethod parse_txt(file_path: str) → List[Tuple[float, float]]

Parse a whitespace/comma-delimited text file of x y pairs.

Each non-empty, non-comment line contains two numbers separated by whitespace or a comma. Lines beginning with # and empty lines are ignored.

Parameters:

file_path (str) – Path to a text file.

Returns:

Sequence of (x, y) pairs.

Return type:

list of tuple[float, float]

Raises:

ValueError – If a line has invalid format or non-numeric values.

Module contents