ppak10/Agentic-SLS-Telemetry

1#!/usr/bin/env python32"""Build the `ticks` config: one row per 10 Hz telemetry tick per build.3 4Reads upstream flat exports from the containing recorder repo:5  builds.jsonl, telemetry/{build_id}.parquet, frames.jsonl, position_hf/{build_id}.parquet6 7For each build with telemetry, emits `data/ticks/{build_id:03d}.parquet`8(zero-padded so lexical sort matches numeric build_id):9  - One row per unique telemetry timestamp (the 10 Hz tick from /state/snapshot)10  - Wide-format sensor columns named "{sensor_id}.{kind}" (~64 columns)11  - Denormalized build context (build_id, job_name, ..., print_profile_name)12  - frame_chamber / frame_galvo / frame_thermal: nearest frame path in13    [tick_ts - 100ms, tick_ts], null when no frame fell in that window14    (frame_thermal is the legacy bedmatrix GIF heatmap; recorder stopped15    producing it after 2026-07-12, so it is null for builds captured since)16  - bedmatrix: nearest IR temperature matrix in the same 100ms window, as a17    struct {width, height, values (row-major °C), path}; null when none.18    This is the raw 32x24 bed-surface temperature grid the recorder now streams19    as JSON in place of the old rendered frame_thermal heatmap.20  - position_hf_burst: list of {ts_offset_ms, x, y, z1, z2, r, has_homed}21    for all position_hf events in the same 100ms window22 23Usage:24    uv run scripts/ticks/01_extract.py            # all builds with telemetry25    uv run scripts/ticks/01_extract.py 13 26      # specific build ids26"""27import json28import sys29from datetime import timedelta30from pathlib import Path31 32import polars as pl33import pyarrow as pa34import pyarrow.parquet as pq35 36sys.path.insert(0, str(Path(__file__).parent.parent))37from _lib import (EXPORTS_DIR, DATA_DIR, TELEMETRY_DIR, POSITION_DIR,38                  FRAMES_BUFFER, iter_jsonl, load_build_to_profile_name)39 40 41OUTPUT_DIR = DATA_DIR / "ticks"42WINDOW = timedelta(milliseconds=100)  # 10 Hz tick interval43FRAME_KINDS = ("chamber", "galvo", "thermal")44# The IR temperature matrix. Same underlying sensor the legacy frame_thermal GIF45# was rendered from, but streamed as a raw {width, height, values} JSON grid.46# Attached like a frame (nearest-before-tick), embedded as a numeric struct.47BEDMATRIX_KIND = "bedmatrix"48# frame_* path strings are relative to this directory in the upstream recorder repo.49FRAMES_DIR = FRAMES_BUFFER  # loose frames for builds not yet zip-delivered50# HF Image feature wire format. Both fields nullable; the struct itself is null when no frame.51IMAGE_STRUCT_TYPE = pa.struct([52    pa.field("bytes", pa.binary()),53    pa.field("path",  pa.string()),54])55# Bedmatrix numeric struct. `values` is row-major, length width*height (32*24=768),56# temperatures in °C as float32 (firmware sends ~0.01°C resolution; f32 is ample).57# The whole struct is null when no matrix fell in the tick window.58BEDMATRIX_STRUCT_TYPE = pa.struct([59    pa.field("width",  pa.int32()),60    pa.field("height", pa.int32()),61    pa.field("values", pa.list_(pa.float32())),62    pa.field("path",   pa.string()),63])64# Streaming chunk size in rows. Tuned so peak embedded payload per chunk stays65# under ~1 GB (thermal frames dominate at ~315 KB each).66CHUNK_ROWS = 2_00067 68 69def load_builds_index() -> dict[int, dict]:70    rows = list(iter_jsonl(EXPORTS_DIR / "builds.jsonl"))71    return {r["id"]: r for r in rows}72 73 74def load_frames_for_build(build_id: int) -> pl.DataFrame:75    """Read just the frame rows for one build out of the upstream frames.jsonl."""76    rows = [r for r in iter_jsonl(EXPORTS_DIR / "frames.jsonl")77            if r.get("build_id") == build_id]78    if not rows:79        return pl.DataFrame(schema={"ts": pl.Datetime("us", "UTC"),80                                    "kind": pl.String, "path": pl.String})81    df = pl.DataFrame(rows).select(82        pl.col("ts").str.to_datetime(time_unit="us", time_zone="UTC"),83        pl.col("kind"),84        pl.col("path"),85    )86    return df87 88 89def pivot_telemetry(tel: pl.DataFrame) -> pl.DataFrame:90    """Wide-pivot (ts, sensor_id, kind, value) → one row per ts with91    {sensor_id}.{kind} columns. Duplicate samples within a tick collapse via first."""92    tel = tel.with_columns(93        col_name=pl.col("sensor_id") + "." + pl.col("kind")94    )95    return tel.pivot(96        on="col_name", index="ts", values="value", aggregate_function="first"97    ).sort("ts")98 99 100def attach_frames(wide: pl.DataFrame, frames: pl.DataFrame) -> pl.DataFrame:101    """For each frame kind, attach the nearest path within WINDOW prior to tick_ts."""102    for kind in FRAME_KINDS:103        f = (104            frames.filter(pl.col("kind") == kind)105            .select(pl.col("ts"), pl.col("path").alias(f"frame_{kind}"))106            .sort("ts")107        )108        if f.height == 0:109            wide = wide.with_columns(pl.lit(None, dtype=pl.String).alias(f"frame_{kind}"))110            continue111        wide = wide.sort("ts").join_asof(112            f, on="ts", strategy="backward", tolerance=WINDOW113        )114    return wide115 116 117def attach_bedmatrix(wide: pl.DataFrame, frames: pl.DataFrame) -> pl.DataFrame:118    """Attach the nearest bedmatrix path within WINDOW prior to tick_ts as a119    string column `bedmatrix` (parsed into a numeric struct at write time,120    exactly like the frame_* path columns)."""121    f = (122        frames.filter(pl.col("kind") == BEDMATRIX_KIND)123        .select(pl.col("ts"), pl.col("path").alias("bedmatrix"))124        .sort("ts")125    )126    if f.height == 0:127        return wide.with_columns(pl.lit(None, dtype=pl.String).alias("bedmatrix"))128    return wide.sort("ts").join_asof(129        f, on="ts", strategy="backward", tolerance=WINDOW130    )131 132 133def attach_position_hf(wide: pl.DataFrame, build_id: int) -> pl.DataFrame:134    """Append a `position_hf_burst` column: list of structs of position_hf events135    in (tick_ts - WINDOW, tick_ts]. Empty list when no events in window or no parquet."""136    pos_path = POSITION_DIR / f"{build_id:03d}.parquet"137    burst_dtype = pl.List(138        pl.Struct({139            "ts_offset_ms": pl.Float64,140            "x":  pl.Float64, "y":  pl.Float64,141            "z1": pl.Float64, "z2": pl.Float64,142            "r":  pl.Float64, "has_homed": pl.Boolean,143        })144    )145    if not pos_path.exists():146        return wide.with_columns(pl.lit([], dtype=burst_dtype).alias("position_hf_burst"))147 148    pos = pl.read_parquet(pos_path).sort("ts")149    pos_records = pos.to_dicts()150    ticks = wide["ts"].to_list()151 152    bursts: list[list[dict]] = []153    pos_lo = 0154    for tick_ts in ticks:155        t_start = tick_ts - WINDOW156        while pos_lo < len(pos_records) and pos_records[pos_lo]["ts"] <= t_start:157            pos_lo += 1158        j = pos_lo159        burst: list[dict] = []160        while j < len(pos_records) and pos_records[j]["ts"] <= tick_ts:161            rec = pos_records[j]162            burst.append({163                "ts_offset_ms": (rec["ts"] - tick_ts).total_seconds() * 1000.0,164                "x":  rec.get("x"),  "y":  rec.get("y"),165                "z1": rec.get("z1"), "z2": rec.get("z2"),166                "r":  rec.get("r"),  "has_homed": rec.get("has_homed"),167            })168            j += 1169        bursts.append(burst)170 171    return wide.with_columns(pl.Series("position_hf_burst", bursts, dtype=burst_dtype))172 173 174def denormalize_build(wide: pl.DataFrame, build_row: dict,175                      profile_name_lookup: dict[int, str]) -> pl.DataFrame:176    """Prepend build-context columns to every row. Cheap in parquet thanks to177    dictionary encoding (every row in this file has the same value)."""178    bid = build_row["id"]179    return wide.with_columns(180        pl.lit(bid).alias("build_id"),181        pl.lit(build_row.get("job_name")).alias("job_name"),182        pl.lit(build_row.get("started_at")).alias("started_at"),183        pl.lit(build_row.get("ended_at")).alias("ended_at"),184        pl.lit(build_row.get("phase")).alias("phase"),185        pl.lit(profile_name_lookup.get(bid)).alias("print_profile_name"),186        pl.lit(None, dtype=pl.String).alias("inova_session_id"),187    )188 189 190def _embed_frame_column(paths: list[str | None]) -> pa.Array:191    """For one chunk's worth of paths, read the image bytes from disk and192    return a StructArray with HF Image shape. Missing path → null struct;193    missing-on-disk → null struct (warn-and-continue)."""194    raw_bytes: list[bytes | None] = []195    for p in paths:196        if p is None:197            raw_bytes.append(None)198            continue199        try:200            raw_bytes.append((FRAMES_DIR / p).read_bytes())201        except FileNotFoundError:202            raw_bytes.append(None)203    bytes_array = pa.array(raw_bytes, type=pa.binary())204    path_array  = pa.array(paths,     type=pa.string())205    # Mask the whole struct as null when there is no path. Children stay null too.206    mask = pa.array([p is None for p in paths], type=pa.bool_())207    return pa.StructArray.from_arrays(208        [bytes_array, path_array],209        fields=[pa.field("bytes", pa.binary()), pa.field("path", pa.string())],210        mask=mask,211    )212 213 214def _embed_bedmatrix_column(paths: list[str | None]) -> pa.Array:215    """For one chunk's paths, read each bedmatrix JSON from disk and return a216    StructArray of {width, height, values, path}. Missing path, missing-on-disk,217    or unparseable JSON → null struct (warn-free continue, same as frames)."""218    widths:  list[int | None]         = []219    heights: list[int | None]         = []220    values:  list[list[float] | None] = []221    kept:    list[str | None]         = []222    for p in paths:223        if p is None:224            widths.append(None); heights.append(None); values.append(None); kept.append(None)225            continue226        try:227            d = json.loads((FRAMES_DIR / p).read_bytes())228            widths.append(d.get("width"))229            heights.append(d.get("height"))230            values.append([float(v) for v in d["values"]])231            kept.append(p)232        except (FileNotFoundError, KeyError, ValueError, TypeError):233            widths.append(None); heights.append(None); values.append(None); kept.append(None)234    mask = pa.array([p is None for p in kept], type=pa.bool_())235    return pa.StructArray.from_arrays(236        [237            pa.array(widths,  type=pa.int32()),238            pa.array(heights, type=pa.int32()),239            pa.array(values,  type=pa.list_(pa.float32())),240            pa.array(kept,    type=pa.string()),241        ],242        fields=list(BEDMATRIX_STRUCT_TYPE),243        mask=mask,244    )245 246 247def _make_output_schema(wide_arrow_schema: pa.Schema) -> pa.Schema:248    """Replace frame_* string fields with HF Image structs and the bedmatrix249    path string with its numeric struct."""250    new_fields = []251    image_field_names = {f"frame_{k}" for k in FRAME_KINDS}252    for field in wide_arrow_schema:253        if field.name in image_field_names:254            new_fields.append(pa.field(field.name, IMAGE_STRUCT_TYPE))255        elif field.name == "bedmatrix":256            new_fields.append(pa.field("bedmatrix", BEDMATRIX_STRUCT_TYPE))257        else:258            new_fields.append(field)259    return pa.schema(new_fields)260 261 262def _embed_chunk(chunk: pa.Table, output_schema: pa.Schema) -> pa.Table:263    """Swap frame_* and bedmatrix string columns for embedded structs in this chunk."""264    image_names = {f"frame_{k}" for k in FRAME_KINDS}265    arrays = []266    for field in output_schema:267        if field.name in image_names:268            arrays.append(_embed_frame_column(chunk[field.name].to_pylist()))269        elif field.name == "bedmatrix":270            arrays.append(_embed_bedmatrix_column(chunk[field.name].to_pylist()))271        else:272            arrays.append(chunk[field.name].combine_chunks())273    return pa.Table.from_arrays(arrays, schema=output_schema)274 275 276def process_build(build_id: int, builds_index: dict[int, dict],277                  profile_name_lookup: dict[int, str]) -> Path | None:278    tel_path = TELEMETRY_DIR / f"{build_id:03d}.parquet"279    if not tel_path.exists():280        return None281 282    tel = pl.read_parquet(tel_path)283    if tel.is_empty():284        return None285 286    wide = pivot_telemetry(tel)287    frames = load_frames_for_build(build_id)288    wide = attach_frames(wide, frames)289    wide = attach_bedmatrix(wide, frames)290    wide = attach_position_hf(wide, build_id)291    wide = denormalize_build(wide, builds_index[build_id], profile_name_lookup)292 293    # Put build context + ts first, then sensors, then frames + bedmatrix + burst.294    leading = ["build_id", "ts", "job_name", "started_at", "ended_at", "phase",295               "print_profile_name", "inova_session_id"]296    frame_cols = [f"frame_{k}" for k in FRAME_KINDS]297    trailing = frame_cols + ["bedmatrix", "position_hf_burst"]298    middle = [c for c in wide.columns if c not in leading and c not in trailing]299    wide = wide.select(leading + middle + trailing)300 301    # Stream-write: build target schema (with image structs), then iterate302    # CHUNK_ROWS-sized slices, embedding bytes per chunk to bound memory.303    base_table = wide.to_arrow()304    output_schema = _make_output_schema(base_table.schema)305    out_path = OUTPUT_DIR / f"{build_id:03d}.parquet"306    with pq.ParquetWriter(out_path, output_schema, compression="zstd") as writer:307        for i in range(0, base_table.num_rows, CHUNK_ROWS):308            chunk = base_table.slice(i, CHUNK_ROWS)309            writer.write_table(_embed_chunk(chunk, output_schema))310    return out_path311 312 313def main():314    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)315    builds_index = load_builds_index()316    profile_name_lookup = load_build_to_profile_name()317 318    args = [int(a) for a in sys.argv[1:]]319    targets = args or sorted(int(p.stem) for p in TELEMETRY_DIR.glob("*.parquet"))320 321    for bid in targets:322        if bid not in builds_index:323            print(f"build {bid}: not in builds.jsonl, skipping")324            continue325        out = process_build(bid, builds_index, profile_name_lookup)326        if out is None:327            print(f"build {bid}: no telemetry parquet, skipping")328            continue329        rows = pl.scan_parquet(out).select(pl.len()).collect().item()330        size = out.stat().st_size331        print(f"build {bid}: wrote {rows:,} ticks → {out.relative_to(Path.cwd())}  ({size:,} bytes)")332 333 334if __name__ == "__main__":335    main()336