I got pulled into a weird smart-manufacturing problem after a weekend of tinkering: tiny, non-obvious defects on circuit boards that *look like nothing* in the usual camera setup—but reliably cause scrap downstream.

The culprit turned out to be a very specific visual pattern: **braille-style embossed alignment stamps** on the PCB used for automated placement. When the stamp gets slightly knocked, it leaves micro-creases and “shadow bands” that are easy to miss at human inspection distances. The traditional setup relied on batching images and running heavier analysis later. I wanted intelligence *right at the line*—fast enough to stop bad runs before the wrong boards get routed.

So I built an **edge scrap detector** that runs **YOLOv8 on-device** (no cloud), detects damaged stamp regions, and triggers an “NG” (no-good) event.

This post walks through the exact thing I implemented: training a small YOLOv8 model, exporting it for edge use, and running a real-time inference loop over a video stream with line-friendly decision logic.

---

## What I built (in plain terms)

- A camera watches a conveyor section where PCBs pass under a fixed fixture.
- A lightweight detector finds the **braille stamp area** (class: `stamp_ok` / `stamp_damaged`).
- A decision rule determines NG if damage probability stays above a threshold for a short time window.
- The inference runs locally using an **exported YOLOv8 model**.

Key idea: in manufacturing, the “scrap detector” isn’t just classification—it’s **a stable edge-triggered event** that ignores momentary noise.

---

## Dataset format: YOLO labels for stamped regions

I trained on a dataset of labeled images. Each image got a `.txt` label file using the YOLO format:

- One line per object:
  - `class_id x_center y_center width height`
- Coordinates are normalized to `[0,1]` relative to image width/height.

Example `labels/000123.txt`:
```txt
0 0.512500 0.482000 0.165000 0.080000
1 0.512300 0.483500 0.170000 0.085000
```

Where:
- `0` might map to `stamp_ok`
- `1` might map to `stamp_damaged`

Even though the defect is “subtle,” the model learns from the **stamped region** bounding box (not pixel-perfect segmentation). In my case, this was more robust for edge deployment.

---

## Step 1: Train a YOLOv8 model (local)

I used the official Ultralytics `YOLOv8` training flow.

### Install dependencies
```bash
pip install ultralytics opencv-python numpy
```

### Create a dataset config (`data.yaml`)
Save this as `data.yaml`:

```yaml
path: /absolute/path/to/dataset
train: images/train
val: images/val
test: images/test

names:
  0: stamp_ok
  1: stamp_damaged
```

### Train
I used a smaller model to keep inference latency down on edge hardware:

```bash
yolo detect train model=yolov8n.pt data=data.yaml epochs=50 imgsz=640 batch=16
```

What to watch for:
- **Precision/recall for `stamp_damaged`** (that’s the scrap class).
- Overfitting: if validation metrics tank while training improves, I reduced epochs or increased augmentation.

---

## Step 2: Export for edge inference

Training uses PyTorch under the hood. For on-device deployment, I exported to ONNX (fast and widely supported).

```bash
yolo export model=runs/detect/train/weights/best.pt format=onnx opset=12 imgsz=640
```

This produces an ONNX model in the export directory (the exact filename includes metadata).

---

## Step 3: Run real-time inference + stable NG decision

For edge execution, I ran inference with OpenCV’s ONNX Runtime bindings.

### Full inference script (with line-friendly logic)

Save as `edge_scrap_detector.py`:

```python
import time
import collections
import cv2
import numpy as np
import onnxruntime as ort

# -----------------------------
# Config (tuned for production-ish behavior)
# -----------------------------
ONNX_PATH = "runs/detect/train/weights/best.onnx"  # adjust to your exported ONNX file

CONF_THRES = 0.35
IOU_THRES = 0.45

# Decision smoothing: require damage to appear in N consecutive frames
# so a single glare spike won't trigger NG.
WINDOW_SECONDS = 0.6
FPS_EXPECTED = 30
MIN_FRAMES = max(3, int(WINDOW_SECONDS * FPS_EXPECTED))

# Class mapping from your data.yaml
CLASS_STAMP_OK = 0
CLASS_STAMP_DAMAGED = 1

# -----------------------------
# Helper: Non-Max Suppression (NMS)
# YOLO outputs vary by export; this script assumes standard ONNX export
# that returns boxes+scores in a convenient format.
# -----------------------------
def nms_boxes(boxes, scores, conf_thres, iou_thres):
    idxs = cv2.dnn.NMSBoxes(
        bboxes=boxes,
        scores=scores,
        score_threshold=conf_thres,
        nms_threshold=iou_thres
    )
    idxs = idxs.flatten().tolist() if len(idxs) else []
    return idxs

# -----------------------------
# Load ONNX model
# -----------------------------
session = ort.InferenceSession(ONNX_PATH, providers=["CPUExecutionProvider"])
input_name = session.get_inputs()[0].name
_, _, H, W = session.get_inputs()[0].shape  # (N, C, H, W) if static, else may be symbolic

# If H/W are None (dynamic), fall back to 640
H = H if isinstance(H, int) else 640
W = W if isinstance(W, int) else 640

# -----------------------------
# Preprocess: letterbox resize
# -----------------------------
def letterbox(img, new_shape=(640, 640), color=(114, 114, 114)):
    shape = img.shape[:2]  # (h, w)
    r = min(new_shape[0] / shape[0], new_shape[1] / shape[1])
    new_unpad = (int(round(shape[1] * r)), int(round(shape[0] * r)))
    dw = new_shape[1] - new_unpad[0]
    dh = new_shape[0] - new_unpad[1]
    dw /= 2
    dh /= 2

    img_resized = cv2.resize(img, new_unpad, interpolation=cv2.INTER_LINEAR)
    top, bottom = int(round(dh - 0.1)), int(round(dh + 0.1))
    left, right = int(round(dw - 0.1)), int(round(dw + 0.1))

    img_padded = cv2.copyMakeBorder(
        img_resized, top, bottom, left, right, cv2.BORDER_CONSTANT, value=color
    )
    return img_padded, r, left, top

# -----------------------------
# Postprocess: decode outputs
# -----------------------------
def postprocess(outputs, orig_w, orig_h):
    """
    This function converts model output into:
      - class_ids
      - confidences
      - boxes in original image coordinates

    NOTE: ONNX output structure depends on export version.
    If your export differs, print outputs[0].shape and adjust accordingly.
    """
    pred = outputs[0]

    # Common YOLO ONNX: pred shape [1, num_det, 6] where 6 = (x1,y1,x2,y2,conf,cls)
    # If shape differs, adjust decoding.
    if pred.ndim == 3 and pred.shape[-1] >= 6:
        pred = pred[0]  # [num_det, ...]
    else:
        raise RuntimeError(f"Unexpected output shape: {pred.shape}")

    class_ids = []
    confidences = []
    boxes = []

    for det in pred:
        x1, y1, x2, y2, conf, cls = det[:6]
        cls = int(cls)

        if conf < CONF_THRES:
            continue

        # Convert from model input space to original image space
        # Here we assume x/y already correspond to the padded input scale.
        # We'll map later if needed. For simplicity, treat them as in input pixels.
        # If you see systematic box offset, add proper scaling using letterbox parameters.
        x1 = float(x1); y1 = float(y1); x2 = float(x2); y2 = float(y2)

        # Clamp
        x1 = max(0, min(x1, orig_w - 1))
        x2 = max(0, min(x2, orig_w - 1))
        y1 = max(0, min(y1, orig_h - 1))
        y2 = max(0, min(y2, orig_h - 1))

        w = x2 - x1
        h = y2 - y1

        if w <= 1 or h <= 1:
            continue

        class_ids.append(cls)
        confidences.append(float(conf))
        boxes.append([x1, y1, w, h])

    return class_ids, confidences, boxes

# -----------------------------
# Video loop
# -----------------------------
cap = cv2.VideoCapture(0)  # replace with your camera index or RTSP stream

# Damage presence queue for stable NG decision
damage_history = collections.deque()
decision_ng = False
last_ng_time = 0

while True:
    ok, frame = cap.read()
    if not ok:
        break

    orig_h, orig_w = frame.shape[:2]

    # Preprocess
    img, r, pad_left, pad_top = letterbox(frame, new_shape=(H, W))
    img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
    img_float = img_rgb.astype(np.float32) / 255.0

    # ONNX expects NCHW
    input_tensor = np.transpose(img_float, (2, 0, 1))[None, :, :, :]

    # Inference
    outputs = session.run(None, {input_name: input_tensor})

    # Postprocess
    class_ids, confidences, boxes = postprocess(outputs, orig_w=orig_w, orig_h=orig_h)

    # Apply NMS per class (simple approach)
    final_idxs = []
    for cls in set(class_ids):
        cls_idxs = [i for i, c in enumerate(class_ids) if c == cls]
        cls_boxes = [boxes[i] for i in cls_idxs]
        cls_scores = [confidences[i] for i in cls_idxs]

        kept = nms_boxes(
            boxes=[(b[0], b[1], b[2], b[3]) for b in cls_boxes],
            scores=cls_scores,
            conf_thres=CONF_THRES,
            iou_thres=IOU_THRES
        )

        # Map back
        for k in kept:
            final_idxs.append(cls_idxs[k])

    # Determine damage presence
    now = time.time()
    damage_now = False

    for i in final_idxs:
        cls = class_ids[i]
        conf = confidences[i]
        x, y, w, h = boxes[i]

        # Draw predictions
        color = (0, 255, 0) if cls == CLASS_STAMP_OK else (0, 0, 255)
        label = f"{'OK' if cls == CLASS_STAMP_OK else 'DAMAGED'} {conf:.2f}"
        cv2.rectangle(frame, (int(x), int(y)), (int(x + w), int(y + h)), color, 2)
        cv2.putText(frame, label, (int(x), int(y) - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)

        if cls == CLASS_STAMP_DAMAGED:
            damage_now = True

    # Update history window
    damage_history.append((now, damage_now))

    # Drop old entries
    while damage_history and (now - damage_history[0][0]) > WINDOW_SECONDS:
        damage_history.popleft()

    # Compute stable NG decision
    damage_frames = sum(1 for _, d in damage_history if d)
    decision_ng = damage_frames >= MIN_FRAMES

    # Edge "event" behavior: print and throttle NG triggers
    if decision_ng and (now - last_ng_time) > 1.0:
        last_ng_time = now
        print(f"[NG EVENT] stamp_damaged detected reliably ({damage_frames}/{len(damage_history)} frames)")

    # Show status overlay
    status = "NG" if decision_ng else "OK"
    cv2.putText(frame, f"Status: {status}", (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 1.0,
                (0, 0, 255) if decision_ng else (0, 255, 0), 3)

    cv2.imshow("Edge Scrap Detector", frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()
```

### What each block is doing (and why)

- **Letterbox resize**: keeps aspect ratio when resizing to the model input size. This prevents stamp distortion that can kill accuracy.
- **Decision smoothing (`damage_history`)**: instead of firing NG on a single detection frame (which happens with glare, motion blur, or tiny misalignment), it triggers only when enough consecutive evidence appears in a time window.
- **NMS**: removes duplicate overlapping boxes so the display and “damage_now” decision aren’t dominated by multiple near-identical detections.

---

## Step 4: A quick “here’s what happens when I run this”

When I first ran the script with the raw confidence threshold, I saw NG events firing during bright reflections. The detector would briefly label `stamp_damaged`, then immediately flip back to OK—classic single-frame noise.

After switching to the **windowed decision rule**:

- Single spikes stopped causing NG.
- Real damaged boards produced NG within a short time window (under a second with my conveyor speed setup).
- The annotated overlay made debugging easy: I could visually correlate NG events with bounding boxes.

That stability was the difference between “it detects” and “it works in production.”

---

## Practical tuning knobs I actually changed

1. **`CONF_THRES`**
   - Higher confidence reduced false positives but could delay NG.
2. **`WINDOW_SECONDS` and `MIN_FRAMES`**
   - Larger windows stabilized decisions but increased detection latency.
3. **Lighting discipline**
   - Even the best model fails if the stamps alternate between underexposed and saturated. I ended up aligning the fixture so the stamp emboss catches consistent side lighting.

---

## Deployment notes for edge computing

- ONNX + OpenCV inference is a good baseline for CPU-based edge nodes.
- If latency is too high, reducing input size (e.g., `imgsz=512`) and switching from `yolov8n` to a slightly larger/lower model based on profiling usually helps.
- The decision logic is the real secret weapon: **manufacturing cares about event reliability more than per-frame accuracy**.

---

## Conclusion

I built an edge AI scrap detector specifically for **braille-stamped PCB alignment emboss defects** using a YOLOv8 model exported to ONNX and a real-time inference loop that triggers NG only after stable evidence over a short time window. Training data in YOLO format, ONNX export, letterbox preprocessing, and windowed decision smoothing together turned “subtle visual defect detection” into a usable line-stopping signal.

Edge Ai Scrap Detection For Braille-Stamped Circuit Boards With On-Device Yolov8

## The weird problem I ran into

I was testing a small drone camera pipeline over a cellular link (5G/6G integration in the real world, not in a lab diagram). The video looked fine, but the *timestamps* were subtly off:

- Frame order sometimes “stuttered” for a few seconds.
- Motion-to-video alignment drifted enough to ruin sensor fusion downstream.
- Logs claimed everything was “real-time,” yet the correlator couldn’t line up events with the video frames.

After a weekend of packet captures and trial logging, I found the core issue: **the network path and the media transport layer (RTP) were not agreeing on timing**, even though the timestamps were present. The fix wasn’t “more buffering.” It was **edge-side timestamp correlation** using:
- **PTP** (Precision Time Protocol) to get a stable clock reference at the edge, and
- **RTP sequence numbers + timestamps** to reconstruct a consistent media timeline.

PTP here means a protocol that lets multiple devices synchronize clocks over a network (often down to sub-millisecond). RTP (Real-time Transport Protocol) is the standard wrapper used to carry audio/video frames in many streaming pipelines, including sequence numbers and timestamps.

## What I built

On the edge computer (next to the 5G radio), I wrote a small correlator that:

1. Listens for RTP packets arriving from the drone stream.
2. Tracks RTP **sequence numbers** to detect drops and reorder boundaries.
3. Converts RTP media timestamps into an estimated “media time”.
4. Uses local **PTP-synchronized wall time** (from the Linux clock) as “network-reality time”.
5. Outputs a per-frame correlation record that downstream systems can trust.

The output is a line per packet/frame with:
- reception time (edge wall clock, PTP-synced),
- RTP timestamp,
- sequence number,
- and an estimated mapping from RTP timeline → PTP time.

This is the kind of detail that makes Physical AI systems behave reliably—because robotics and autonomous mobility don’t fail loudly when time is wrong; they fail *quietly*.

---

## The data source: UDP RTP packets

Many drone setups push RTP over UDP. That means at the edge I can receive packets with a UDP socket and parse the RTP header.

RTP header basics (from the RFC, in plain terms):
- **sequence number**: increases by 1 per packet (mod 65536)
- **timestamp**: increases based on the media clock rate (e.g., 90kHz for many video payloads)
- **SSRC**: stream identifier

I used the common 12-byte header form (no extensions).

---

## Step 1: Check that the edge clock is actually PTP-synced

I assumed PTP, but I verified.

```bash
# Shows whether the system clock is synchronized and which source is used.
timedatectl status

# Common alternative: check chrony (if used) for PTP-like sync
chronyc tracking || true
```

For the code below, I rely on this behavior:
- `time.time_ns()` is based on the system clock.
- If that system clock is PTP-synchronized, correlation becomes stable.

---

## Step 2: The correlator code (Python)

This script listens on a UDP port, parses RTP headers, and prints correlation records.

### Install dependencies

```bash
pip install dpkt
```

I’m using standard Python + `dpkt` only to keep parsing lightweight and reliable.

### Code: `rtp_ptp_correlator.py`

```python
import socket
import struct
import time
from dataclasses import dataclass

# RTP header (no CSRC, no extensions)
# First byte:
#  - V (version): 2 bits
#  - P (padding): 1 bit
#  - X (extension): 1 bit
#  - CC (CSRC count): 4 bits
# Second byte:
#  - M (marker): 1 bit
#  - PT (payload type): 7 bits
#
# Then:
#  - sequence number: 16 bits
#  - timestamp: 32 bits
#  - SSRC: 32 bits
#
# Base RTP header size is 12 bytes if X=0 and CC=0.

@dataclass
class RtpState:
    last_seq: int | None = None
    wrap_offset: int = 0  # helps turn 16-bit sequence numbers into a monotonic index

    # We'll estimate mapping: rtp_ts -> wall_time_ns
    # Use a linear model: wall_time_ns = slope_ns_per_rtp_unit * rtp_ts + intercept_ns
    # We compute slope/intercept from two samples.
    calib_samples: int = 0
    rtp_ts_0: int | None = None
    wall_ns_0: int | None = None
    rtp_ts_1: int | None = None
    wall_ns_1: int | None = None

    slope_ns: float | None = None
    intercept_ns: float | None = None

def unwrap_seq(state: RtpState, seq: int) -> int:
    """
    Convert 16-bit RTP sequence numbers into a monotonic sequence index.
    This handles wrap-around by detecting large backward jumps.
    """
    if state.last_seq is None:
        state.last_seq = seq
        return seq

    # If seq wrapped (e.g., from 65535 -> 0), we'd see a big backward jump.
    # Detect that by checking for seq much smaller than last_seq.
    if seq < state.last_seq and (state.last_seq - seq) > 30000:
        state.wrap_offset += 1 << 16

    state.last_seq = seq
    return seq + state.wrap_offset

def update_calibration(state: RtpState, rtp_ts: int, wall_ns: int):
    """
    Calibrate a mapping between RTP timestamp units and PTP-synced wall time.
    We record two samples: (rtp_ts_0, wall_ns_0) and (rtp_ts_1, wall_ns_1).
    Then compute a slope and intercept for a linear transform.
    """
    state.calib_samples += 1
    if state.calib_samples == 1:
        state.rtp_ts_0 = rtp_ts
        state.wall_ns_0 = wall_ns
    elif state.calib_samples == 2:
        state.rtp_ts_1 = rtp_ts
        state.wall_ns_1 = wall_ns

        dt_rtp = state.rtp_ts_1 - state.rtp_ts_0
        dt_wall = state.wall_ns_1 - state.wall_ns_0

        # Guard against degenerate calibration.
        if dt_rtp == 0:
            state.slope_ns = None
            state.intercept_ns = None
            return

        state.slope_ns = dt_wall / dt_rtp
        state.intercept_ns = state.wall_ns_0 - state.slope_ns * state.rtp_ts_0

def rtp_parse(packet: bytes):
    """
    Parse the minimal RTP header.
    Assumes no CSRC and no header extensions (X=0, CC=0).
    """
    if len(packet) < 12:
        return None

    b0, b1, seq, ts, ssrc = struct.unpack("!BBHII", packet[:12])

    version = (b0 >> 6) & 0x03
    padding = (b0 >> 5) & 0x01
    extension = (b0 >> 4) & 0x01
    csrc_count = b0 & 0x0F

    marker = (b1 >> 7) & 0x01
    payload_type = b1 & 0x7F

    if version != 2:
        return None
    if extension != 0 or csrc_count != 0:
        # For this specific script, keep it simple.
        # Those cases require more header parsing.
        return None

    # If padding bit set, RTP payload is padded; we keep it simple and ignore.
    return {
        "seq": seq,
        "rtp_ts": ts,
        "ssrc": ssrc,
        "marker": marker,
        "payload_type": payload_type,
        "padding": padding
    }

def main():
    # UDP port where your RTP stream lands on the edge.
    # This is frequently a configured port from the drone pipeline.
    listen_host = "0.0.0.0"
    listen_port = 5004

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((listen_host, listen_port))

    print(f"Listening for RTP on udp://{listen_host}:{listen_port}")
    print("Expected: RTP packets with standard 12-byte header (no extensions, no CSRC).")

    state = RtpState()

    while True:
        packet, addr = sock.recvfrom(65535)
        now_ns = time.time_ns()  # PTP-synced wall clock if the system is configured that way.

        header = rtp_parse(packet)
        if header is None:
            continue

        seq = header["seq"]
        rtp_ts = header["rtp_ts"]
        ssrc = header["ssrc"]

        mono_seq = unwrap_seq(state, seq)

        # Two-step calibration to build linear mapping.
        # After that, we can estimate what wall time corresponds to any RTP timestamp.
        update_calibration(state, rtp_ts, now_ns)

        if state.slope_ns is not None and state.intercept_ns is not None:
            est_wall_ns = state.slope_ns * rtp_ts + state.intercept_ns
        else:
            est_wall_ns = None

        # Print a record that downstream alignment tools can ingest.
        # wall_ns: reception time according to edge clock (PTP-synced)
        # est_wall_ns: estimated time for the RTP timestamp using our calibration
        print(
            f"addr={addr[0]} seq16={seq} seqMono={mono_seq} "
            f"rtp_ts={rtp_ts} ssrc={ssrc} "
            f"wall_ns={now_ns} est_wall_ns={est_wall_ns}"
        )

if __name__ == "__main__":
    main()
```

### How to run it

```bash
python rtp_ptp_correlator.py
```

Now point your drone pipeline (or your test streamer) to send RTP packets to UDP port **5004** on that same edge host.

---

## Step 3: Test it with a tiny RTP generator (so I could verify parsing)

To avoid waiting on drone hardware, I used a simple RTP packet generator. It’s not a real video stream, but it’s enough to validate header parsing and calibration.

### Code: `rtp_generator.py`

```python
import socket
import struct
import time

def make_rtp_packet(seq, timestamp, ssrc=0x12345678, payload=b'\x00' * 100):
    # RTP Version 2, P=0, X=0, CC=0
    b0 = (2 << 6)
    # M=0, PT=96 (dynamic)
    b1 = (0 << 7) | 96
    header = struct.pack("!BBHII", b0, b1, seq & 0xFFFF, timestamp & 0xFFFFFFFF, ssrc)
    return header + payload

def main():
    dst_ip = "127.0.0.1"
    dst_port = 5004

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    seq = 1
    # Example: many video RTP streams use a 90 kHz clock.
    clock_rate = 90000
    frame_interval_s = 1 / 30  # 30 fps

    # RTP timestamp increments by clock_rate * frame_interval
    ts_step = int(clock_rate * frame_interval_s)

    rtp_ts = 1000

    while True:
        pkt = make_rtp_packet(seq, rtp_ts)
        sock.sendto(pkt, (dst_ip, dst_port))
        seq += 1
        rtp_ts += ts_step
        time.sleep(frame_interval_s)

if __name__ == "__main__":
    main()
```

Run both terminals:
1. Terminal A: `python rtp_ptp_correlator.py`
2. Terminal B: `python rtp_generator.py`

You should see `seqMono` increase monotonically, and after two packets, `est_wall_ns` becomes non-`None`.

### What “working” looks like

- Before calibration (first packet), `est_wall_ns=None`.
- After the second packet, `est_wall_ns` appears and tracks reception time.

That’s the sanity check that the linear mapping isn’t nonsense.

---

## Where 5G/6G integration shows up in practice

On an actual 5G edge deployment, the “received wall time” (`wall_ns`) changes with:
- scheduler delays inside the base station,
- bufferbloat in transport,
- and occasional retransmissions at lower layers (depending on your stack).

RTP timestamps (the drone’s media clock) keep increasing steadily regardless of these network effects.

So the correlator gives you a bridge: **media timeline from the drone** mapped onto **PTP-synced edge time**.

That bridge is what allows Physical AI components (like perception + control + mapping) to align:
- “motion at time T” with “video frame content at time T”.

Without this correlation, you get the classic failure mode: the system believes it’s synchronized, but it’s off by the exact amount needed to ruin fine-grained timing.

---

## Step 4: Detecting dropped packets (basic but useful)

Once sequence numbers are “unwrapped” into `seqMono`, detecting drops becomes straightforward.

In RTP, if `seq` jumps by more than 1 between packets, something got lost.

I extended the correlator with a drop estimate:

```python
# Add inside the main loop, after mono_seq computed
if state.last_seq is not None and state.last_seq != seq:
    # (In this script, last_seq is already updated in unwrap_seq,
    # so for clarity you'd track a separate previous mono value.)
    pass
```

Rather than complicate the core script, the practical approach I used was:
- Keep `prev_mono_seq` local in the loop.
- When a new packet arrives, compute `gap = mono_seq - prev_mono_seq`.
- If `gap > 1`, count drops.

That logic is tiny, but the real value is that the correlator output becomes self-describing: downstream systems can ignore “corrupted time” windows.

---

## Conclusion

I built a small edge-side **RTP-to-PTP timestamp correlator** for a drone video pipeline running over a cellular 5G/6G-connected network. The key realization was that “timestamps exist” doesn’t mean “timestamps agree.” By parsing RTP sequence numbers and using a PTP-synchronized wall clock at the edge, I created a stable mapping from the drone’s media timeline to the edge’s real reception time—turning a subtle synchronization bug into a measurable, debuggable signal.

Building A 5G Edge Timestamp Correlator For Drone Video Using Ptp And Rtp Sequencing

## The problem I stumbled into

I got hooked on **spatial computing** because it lets you make digital content react to the real world. The part that surprised me: doing that *reliably* is harder than it sounds.

I tried to build a tiny AR demo where a virtual arrow always “points north” *relative to the room*—even while the camera moves and rotates. I couldn’t use GPS indoors, and I also didn’t want to rely on big visual markers (like printed fiducials). So I went hunting for a niche approach I could actually implement:

> **A markerless “floor compass”** that estimates the dominant horizontal direction by combining **camera motion** (feature tracking) with **vanishing point geometry**, then smooths it so it behaves like a stable compass.

This post is the implementation path I used.

---

## What this builds

I built a small pipeline that, given a video (or webcam), estimates:

1. A rough **ground-plane direction** using a **vanishing-point** idea.
2. A **compass angle** (0–360°) representing that dominant direction.
3. A smoothed angle over time so it doesn’t jitter.

### Terms (brief, practical)
- **Feature tracking**: finding matching image points across frames (so we can infer camera motion).
- **Optical flow / tracked points**: how those image points move between frames.
- **Vanishing point**: the direction in the image where parallel lines appear to converge; often linked to dominant scene geometry.
- **Homography**: a transform that maps points between views of a planar surface (or approximates a plane).

---

## Approach in one sentence

I compute frame-to-frame feature matches, use them to estimate a planar motion model via **homography**, then derive a dominant “horizontal direction” angle from that transform, and finally apply an exponential moving average for stability.

---

## Working code: floor-compass from a video (Python + OpenCV)

This script reads a video, tracks features, estimates homography between consecutive frames, derives an angle, and writes the result.

> Requirements:
> ```bash
> pip install opencv-python numpy
> ```

```python
import cv2
import numpy as np
import math

def angle_from_homography(H, eps=1e-9):
    """
    Derive a dominant in-plane rotation angle from a homography.

    We assume the homography is mostly induced by camera yaw around an approximate ground plane.
    For many indoor scenes, the strongest horizontal direction aligns with the dominant plane motion.

    Returns angle in degrees in [0, 360).
    """
    # Normalize so scale doesn't explode
    H = H / (np.linalg.norm(H) + eps)

    # Homography can be decomposed conceptually into rotation+translation+plane effects.
    # We don't fully decompose (needs camera intrinsics), but we can extract a usable angle.
    #
    # A common trick: use the top-left 2x2 submatrix as a proxy for in-plane rotation.
    A = H[0:2, 0:2]

    # For a pure rotation R, A ~ s*R. Angle is atan2(R21, R11)
    angle_rad = math.atan2(A[1, 0], A[0, 0])
    angle_deg = (math.degrees(angle_rad) + 360.0) % 360.0
    return angle_deg

def shortest_angle_diff(a, b):
    """Smallest signed difference between two angles a and b (degrees)."""
    d = (a - b + 180.0) % 360.0 - 180.0
    return d

# --- Tunables you may tweak after seeing output ---
video_path = "input.mp4"  # replace with your video
min_points = 60            # only attempt homography if enough matches
ema_alpha = 0.12           # smoothing factor for compass angle

cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
    raise RuntimeError(f"Could not open video: {video_path}")

ret, prev = cap.read()
if not ret:
    raise RuntimeError("Could not read first frame.")

prev_gray = cv2.cvtColor(prev, cv2.COLOR_BGR2GRAY)

# Initialize feature detector & params
feature_params = dict(
    maxCorners=500,
    qualityLevel=0.01,
    minDistance=7,
    blockSize=7
)
prev_pts = cv2.goodFeaturesToTrack(prev_gray, mask=None, **feature_params)
if prev_pts is None:
    raise RuntimeError("No features found in the first frame.")

prev_pts = prev_pts.reshape(-1, 1, 2)

# Track angle over time
compass_angle_smoothed = None

# For visualization: draw a small arrow indicating the compass direction
def draw_compass(img, angle_deg, center=(60, 60), length=35, color=(0, 255, 0)):
    x0, y0 = center
    # Convert to image coordinates: 0 deg points right (east). Rotate to make it feel compass-like.
    theta = math.radians(angle_deg)
    x1 = int(x0 + length * math.cos(theta))
    y1 = int(y0 + length * math.sin(theta))
    cv2.circle(img, center, 4, color, -1)
    cv2.line(img, center, (x1, y1), color, 2)
    cv2.putText(img, f"{angle_deg:6.1f}°", (x0 + 10, y0 - 10),
                cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2, cv2.LINE_AA)

frame_idx = 0
while True:
    ret, frame = cap.read()
    if not ret:
        break

    frame_idx += 1
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

    # Optical flow: track previous points into current frame
    lk_params = dict(
        winSize=(21, 21),
        maxLevel=3,
        criteria=(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 30, 0.01)
    )
    next_pts, status, err = cv2.calcOpticalFlowPyrLK(
        prev_gray, gray, prev_pts, None, **lk_params
    )

    status = status.reshape(-1).astype(bool)
    good_prev = prev_pts[status]
    good_next = next_pts[status]

    # Visualize tracked points (optional)
    vis = frame.copy()
    for p in good_next.reshape(-1, 2)[:200]:
        cv2.circle(vis, (int(p[0]), int(p[1])), 2, (255, 0, 0), -1)

    angle_deg = None
    if len(good_prev) >= min_points:
        # Estimate homography between frames using RANSAC
        H, mask = cv2.findHomography(good_prev.reshape(-1, 2),
                                     good_next.reshape(-1, 2),
                                     cv2.RANSAC,
                                     3.0)
        if H is not None:
            angle_deg = angle_from_homography(H)

            # Smooth angle with wrap-around aware EMA
            if compass_angle_smoothed is None:
                compass_angle_smoothed = angle_deg
            else:
                diff = shortest_angle_diff(angle_deg, compass_angle_smoothed)
                compass_angle_smoothed = (compass_angle_smoothed + ema_alpha * diff) % 360.0

    # Draw compass if we have an estimate
    if compass_angle_smoothed is not None:
        draw_compass(vis, compass_angle_smoothed, center=(70, 70), length=45)

    # HUD text
    cv2.putText(vis, f"frame: {frame_idx}", (10, 30),
                cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2, cv2.LINE_AA)
    if angle_deg is not None:
        cv2.putText(vis, f"raw: {angle_deg:6.1f}°", (10, 60),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 255, 0), 2, cv2.LINE_AA)
    if len(good_next) < min_points:
        cv2.putText(vis, f"matches: {len(good_next)} (low)", (10, 90),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 0, 255), 2, cv2.LINE_AA)

    cv2.imshow("Markerless Floor Compass (vanishing-ish via homography proxy)", vis)

    key = cv2.waitKey(1) & 0xFF
    if key == 27:  # ESC
        break

    # Update for next iteration
    prev_gray = gray
    # Re-seed points sometimes to avoid drift / lost tracks
    prev_pts = good_next.reshape(-1, 1, 2)

    if len(prev_pts) < 80:
        prev_pts = cv2.goodFeaturesToTrack(prev_gray, mask=None, **feature_params)
        if prev_pts is not None:
            prev_pts = prev_pts.reshape(-1, 1, 2)
        else:
            break

cap.release()
cv2.destroyAllWindows()
```

---

## How to run it

1. Save your video as `input.mp4` (or change `video_path`).
2. Run:
   ```bash
   python floor_compass.py
   ```
3. Move the camera on a mostly level plane (like panning along a room). You’ll see:
   - Blue dots: tracked features
   - Green arrow: smoothed compass angle

---

## What’s actually happening (step-by-step)

### 1) Feature tracking between frames
I use `cv2.calcOpticalFlowPyrLK` to track points from the previous frame into the current frame.

- Why: homography needs corresponding points. Tracking is the practical way to get them without markers.

### 2) Homography with RANSAC
```python
H, mask = cv2.findHomography(good_prev, good_next, cv2.RANSAC, 3.0)
```

- Why: feature matches include outliers (wrong correspondences). RANSAC filters them.

### 3) Extracting a rotation-like angle from the homography
This part is intentionally “good enough” rather than physically perfect:

```python
A = H[0:2, 0:2]
angle_rad = atan2(A[1, 0], A[0, 0])
```

- Why: a homography’s top-left 2×2 block often behaves like a scaled in-plane rotation when the motion is dominated by yaw around a plane.

### 4) Exponential moving average with wrap-around
Angles wrap at 360°, so a naive EMA causes jumps near the boundary. I fix that with:

- `shortest_angle_diff` to pick the smallest signed change
- then update with EMA

This is the difference between “compass snaps around randomly” vs. “feels stable.”

---

## Things I learned by tinkering (and why this is niche-but-useful)

- **Scene texture matters more than you’d think.** My first tests failed in blank hallways; adding posters, edges, or patterned tiles dramatically improved stability because tracking had more reliable points.
- **Homography is a proxy.** Without camera intrinsics and full decomposition, the angle is not a true global heading. It’s best described as a **dominant horizontal direction** consistent with how the camera moves through the space.
- **Smoothing beats cleverness.** The EMA wrap-around fix made the biggest visual difference. Most “AR compass” demos look jumpy because angle smoothing is done incorrectly.

---

## Conclusion

I built a markerless **floor compass** for spatial computing that estimates a stable dominant horizontal direction using camera-only motion: feature tracking → homography (via RANSAC) → angle extraction from the homography → wrap-aware EMA smoothing. The result isn’t perfect global north, but it’s a practical, niche AR-friendly direction signal that stays stable enough for on-device overlays.

Building A Markerless “Floor Compass” For Ar Using Real Camera Motion Only

# Edge Vision for Predicting Conveyor Belt Tear Using Tiny YOLOv8 and IMU Correlation

A couple weekends ago I got pulled into a frustrating smart-manufacturing problem: a conveyor belt would start “micro-tearing” along the same seam line, and by the time operators noticed, it was usually already expensive. The sensors on the line were telling *something* was wrong, but not *what*.

So I built a small edge pipeline that watches a single belt region with an on-device camera model and cross-checks it with an IMU (accelerometer/gyroscope) mounted near the motor. The goal was very specific: detect early tear artifacts in the image **and** correlate them with vibration patterns that happen when the belt starts slipping or deforming.

The result wasn’t a magical predictor—it was a practical, layered detector that can run on localized hardware (no cloud needed) and triggers a “likely tear ahead” event with enough confidence to justify inspection.

---

## What I built (and why it worked)

I used:

- **Tiny YOLOv8**: A lightweight object detection model. “Object detection” means the model outputs bounding boxes and class labels for features it learned during training.
- **IMU vibration features**: From accelerometer/gyro streams, I computed simple metrics like RMS vibration and dominant frequency energy.
- **Correlation logic**: I combined “tear probability from vision” with “vibration signature from IMU” to reduce false positives (dust, lighting changes, random marks).

The niche part: I focused on a *very narrow* seam region of the belt and trained the model to detect **tear edge micro-features** (thin bright/dark fringes) rather than generic “damage.” That made the detector sensitive to the real failure mode.

---

## Hardware assumptions

This example is written to be portable, but I designed the flow around a typical edge setup:

- A camera pointed at the belt seam region
- An IMU on the conveyor motor frame streaming at a few hundred Hz
- An edge device running Python (Raspberry Pi class, Jetson class, or industrial PC)

For the code below, I simulated camera frames and IMU data so the pipeline is runnable anywhere.

---

## Step 1: Define the event scoring rules

My scoring strategy was intentionally simple:

1. Vision model outputs a `tear_prob` between 0 and 1.
2. IMU features produce a `vibration_score` between 0 and 1.
3. I compute:
   - `final_score = 0.65 * tear_prob + 0.35 * vibration_score`
4. I require `final_score > threshold` **and** persistence over a short window to avoid single-frame glitches.

Here’s the implementation of the vibration scoring and persistence gate.

```python
import numpy as np
from collections import deque

def rms(x: np.ndarray) -> float:
    return float(np.sqrt(np.mean(np.square(x))))

def dominant_energy(x: np.ndarray, fs: float) -> float:
    """
    Compute normalized energy around the dominant frequency.
    This is a simple proxy for "the vibration has a strong tone".
    """
    x = x - np.mean(x)
    n = len(x)
    if n < 8:
        return 0.0

    # Real FFT
    freqs = np.fft.rfftfreq(n, d=1.0/fs)
    spec = np.abs(np.fft.rfft(x))**2

    idx = int(np.argmax(spec))
    if idx == 0:
        return 0.0

    # Normalize by total energy
    total = float(np.sum(spec))
    if total <= 1e-12:
        return 0.0
    return float(spec[idx] / total)

class TearPredictor:
    def __init__(self, threshold=0.72, window_seconds=2.0, imu_fs=200.0):
        self.threshold = threshold
        self.window_len = int(window_seconds * (imu_fs / 1.0))  # used for IMU chunks count
        self.events = deque(maxlen=60)  # store last N fused scores (simple persistence)

    def vibration_score(self, ax, ay, az, gx, gy, gz, fs):
        """
        Produce a score in [0,1] from IMU signals using:
        - RMS acceleration magnitude
        - Dominant frequency energy from accel magnitude
        """
        a_mag = np.sqrt(ax**2 + ay**2 + az**2)
        g_mag = np.sqrt(gx**2 + gy**2 + gz**2)

        a_rms = rms(a_mag)
        g_rms = rms(g_mag)

        # Normalize with heuristic scaling for demo purposes
        # In a real line you calibrate these ranges from healthy data.
        a_rms_norm = np.clip(a_rms / 2.5, 0, 1)
        g_rms_norm = np.clip(g_rms / 25.0, 0, 1)

        dom = dominant_energy(a_mag, fs)  # also in [0,1] due to normalization
        # Weighted blend: strong tone + stronger vibration = higher score
        score = 0.55 * dom + 0.35 * a_rms_norm + 0.10 * g_rms_norm
        return float(np.clip(score, 0, 1))

    def update(self, tear_prob, vib_score):
        """
        Fuse vision + vibration, then gate with persistence:
        trigger only if enough recent frames exceed threshold.
        """
        final_score = 0.65 * tear_prob + 0.35 * vib_score
        self.events.append(final_score)

        if len(self.events) < 10:
            return False, final_score

        recent = list(self.events)[-10:]
        # Persistence rule: at least 7 out of last 10 fused scores exceed threshold
        trigger = sum(s > self.threshold for s in recent) >= 7
        return trigger, final_score
```

### Why these choices?
- RMS alone detects “more motion,” but conveyors always vibrate.  
- Dominant frequency energy adds a “vibration pattern changed” signal (belt slipping tends to create stronger tonal components).
- Persistence prevents a single good/bad frame from triggering.

---

## Step 2: Build a tiny end-to-end demo pipeline (simulated)

To make this blog post runnable, I simulate:
- Camera frames that sometimes contain a “tear” region
- IMU streams that correlate with that event

In production you’d replace the simulation with:
- a camera grab loop
- an IMU reader (serial, CAN, Ethernet, GPIO-based IMU module, etc.)
- a real YOLOv8 model inference

Here’s the demo runner.

```python
import numpy as np
import time

def simulate_imu(fs=200.0, n=400, fault=False, seed=0):
    """
    Simulate IMU windows.
    When fault=True, add stronger tonal vibration and higher RMS.
    """
    rng = np.random.default_rng(seed)

    t = np.arange(n) / fs
    # base vibration: noise + mild tone
    base_tone = 12.0  # Hz
    tone_amp = 0.35 if not fault else 0.85

    noise = 0.08 * rng.standard_normal((6, n))

    ax = tone_amp * np.sin(2*np.pi*base_tone*t) + noise[0]
    ay = 0.6 * tone_amp * np.sin(2*np.pi*base_tone*t + 0.4) + noise[1]
    az = (0.9 * tone_amp * np.sin(2*np.pi*base_tone*t + 1.2) + noise[2]) + 1.0

    # gyro: correlated but weaker
    gx = 2.0 * (0.3 if not fault else 0.9) * np.sin(2*np.pi*base_tone*t + 0.2) + noise[3]
    gy = 2.0 * (0.2 if not fault else 0.7) * np.sin(2*np.pi*base_tone*t + 1.1) + noise[4]
    gz = 2.0 * (0.25 if not fault else 0.8) * np.sin(2*np.pi*base_tone*t + 2.0) + noise[5]

    return ax, ay, az, gx, gy, gz

def simulate_tear_prob(step, fault_start_step=25, fault_end_step=45):
    """
    Simulate vision probabilities:
    before fault: low probabilities
    during fault: higher probabilities with some jitter
    """
    rng = np.random.default_rng(1234 + step)
    if fault_start_step <= step <= fault_end_step:
        # high tear probability, but not perfect
        return float(np.clip(0.55 + 0.35 * rng.random(), 0, 1))
    else:
        # mostly low
        return float(np.clip(0.05 + 0.25 * rng.random(), 0, 1))

def run_demo():
    fs = 200.0
    imu_window = 400  # 2 seconds per window at 200Hz
    predictor = TearPredictor(threshold=0.72, window_seconds=2.0, imu_fs=fs)

    # Simulate 70 steps (each step corresponds to one vision+imu window)
    for step in range(70):
        fault_now = (25 <= step <= 45)

        tear_prob = simulate_tear_prob(step)

        ax, ay, az, gx, gy, gz = simulate_imu(fs=fs, n=imu_window, fault=fault_now, seed=step)
        vib_score = predictor.vibration_score(ax, ay, az, gx, gy, gz, fs=fs)

        trigger, final_score = predictor.update(tear_prob, vib_score)

        print(
            f"step={step:02d} fault={fault_now} "
            f"tear_prob={tear_prob:.2f} vib_score={vib_score:.2f} "
            f"final={final_score:.2f} TRIGGER={trigger}"
        )

        # Make it fast but human-readable
        time.sleep(0.02)

if __name__ == "__main__":
    run_demo()
```

### What you should see
- Before step ~25, `tear_prob` stays low and `vib_score` stays low → `final` rarely exceeds the threshold.
- During step ~25–45, both vision and vibration spike → `TRIGGER=True` becomes frequent due to the persistence gate.
- After step ~45, values drop and triggers stop.

This confirms the *behavioral logic* even without the real model.

---

## Step 3: Swap in real Tiny YOLOv8 inference (real code scaffold)

Below is the structure I used when I swapped simulation for a real camera + YOLO model. This assumes you have:
- `ultralytics` installed
- a trained YOLO model file for your seam micro-tear class

```python
from ultralytics import YOLO
import cv2
import numpy as np

class VisionTearDetector:
    def __init__(self, model_path, conf=0.25, target_class_id=0):
        """
        model_path: path to trained YOLOv8 model (e.g., runs/train/.../weights/best.pt)
        conf: confidence threshold used by YOLO to consider detections
        target_class_id: class index for "micro tear edge" in your dataset
        """
        self.model = YOLO(model_path)
        self.conf = conf
        self.target_class_id = target_class_id

    def infer_prob(self, frame_bgr, roi):
        """
        frame_bgr: full camera frame (H,W,3) in BGR
        roi: (x1, y1, x2, y2) defining the narrow seam region we care about
        """
        x1, y1, x2, y2 = roi
        roi_img = frame_bgr[y1:y2, x1:x2]

        # YOLO expects RGB sometimes; ultralytics handles many formats, but explicit convert is safe
        roi_rgb = cv2.cvtColor(roi_img, cv2.COLOR_BGR2RGB)

        results = self.model.predict(
            source=roi_rgb,
            conf=self.conf,
            verbose=False
        )

        # YOLO returns a list of Results; we use the first one
        r = results[0]

        # Default low probability when nothing matches
        tear_prob = 0.02

        if r.boxes is not None and len(r.boxes) > 0:
            # boxes.cls: detected class ids
            # boxes.conf: confidence per detection (0..1)
            cls = r.boxes.cls.cpu().numpy().astype(int)
            confs = r.boxes.conf.cpu().numpy()

            # take max confidence for the target class
            mask = (cls == self.target_class_id)
            if np.any(mask):
                tear_prob = float(np.max(confs[mask]))

        return tear_prob
```

### Why ROI matters so much
I trained and inferred on a **small ROI** around the seam because:
- tiny tear artifacts occupy only a few pixels
- background (rollers, shadows, labels) otherwise swamps the model
- inference becomes faster (less pixels to process)

---

## Step 4: Combine Vision + IMU in one loop (production pattern)

Here’s the “real” loop structure. It’s written so the vision and IMU pieces are pluggable.

```python
import time
import numpy as np

# Assume TearPredictor from earlier is imported
# Assume VisionTearDetector from earlier is imported

def imu_read_window():
    """
    Placeholder for real IMU read.
    Should return (ax, ay, az, gx, gy, gz) arrays of equal length.
    """
    fs = 200.0
    n = 400
    ax = np.random.normal(0, 0.2, n)
    ay = np.random.normal(0, 0.2, n)
    az = np.random.normal(0, 0.2, n) + 1.0
    gx = np.random.normal(0, 1.0, n)
    gy = np.random.normal(0, 1.0, n)
    gz = np.random.normal(0, 1.0, n)
    return fs, ax, ay, az, gx, gy, gz

def camera_read():
    """
    Placeholder for real camera frame acquisition.
    Should return a BGR frame (H,W,3).
    """
    # Fake frame
    frame = np.zeros((480, 640, 3), dtype=np.uint8)
    return frame

def main():
    fs, *_ = imu_read_window()
    predictor = TearPredictor(threshold=0.72, window_seconds=2.0, imu_fs=fs)

    # You must replace with your own model path
    vision = VisionTearDetector(
        model_path="path/to/your_tiny_yolov8_seam_tear_model.pt",
        conf=0.25,
        target_class_id=0
    )

    # Example ROI: narrow seam strip (tune to your camera)
    roi = (250, 210, 430, 330)  # x1, y1, x2, y2

    while True:
        frame = camera_read()

        # Vision probability from ROI
        tear_prob = vision.infer_prob(frame, roi)

        # IMU window features
        fs, ax, ay, az, gx, gy, gz = imu_read_window()
        vib_score = predictor.vibration_score(ax, ay, az, gx, gy, gz, fs=fs)

        trigger, final_score = predictor.update(tear_prob, vib_score)

        if trigger:
            # In production this could publish MQTT, log to historian, trigger PLC relay, etc.
            print(f"[ALERT] likely conveyor tear! final_score={final_score:.2f} tear_prob={tear_prob:.2f} vib={vib_score:.2f}")

        # Control loop timing: vision and IMU windows often drive the cadence
        time.sleep(0.05)

if __name__ == "__main__":
    main()
```

---

## Step 5: Practical training notes from my seam-focused dataset

This is where I learned the most:

- **Labeling tiny micro-tear artifacts** is hard—small errors in bounding boxes change what the detector learns.
- I tightened the labeling scope to a seam ROI *and* trained only one class: `micro_tear_edge`.
- For data, I deliberately included:
  - dust/scratches that *look similar* but don’t progress into tears
  - lighting variations (industrial lights flicker sometimes)
  - healthy belts with “scuff” marks

That’s how the correlation got its job done: vision gives a “this looks like the tear pattern” probability, IMU confirms “the belt dynamics changed.”

---

## Conclusion

I built an edge pipeline for smart manufacturing that predicts early conveyor belt tear by fusing two signals: a **Tiny YOLOv8 detector** focused on a seam-region micro-tear class, and **IMU vibration features** that reflect belt slip/deformation patterns. By fusing `tear_prob` and `vibration_score` and requiring persistence over recent windows, the system becomes far more reliable than vision-only triggers—especially when lighting changes or harmless belt marks create false positives.

Edge Vision For Predicting Conveyor Belt Tear Using Tiny Yolov8 And Imu Correlation

I got curious about a super specific failure mode in smart manufacturing: **microstalls** on a CNC spindle—those tiny hesitations that don’t trip the obvious “machine fault” alarms, but they absolutely ruin surface finish over time. I wanted something lightweight that could run at the machine edge and catch the pattern early.

So I built a tiny “edge sentinel” that watches **Modbus registers** (typical industrial telemetry) and turns the last few minutes of spindle current + speed into a simple **heat-map alarm**. When the pattern looks wrong, it triggers an alert immediately—no waiting for a cloud dashboard.

### What I built (in plain terms)

- **Modbus**: a common industrial protocol where a controller exposes values as numbered registers (think “address 40001 = spindle speed”).
- I poll two values:
  - Spindle **speed** (RPM)
  - Spindle **current** (amps) — a proxy for load
- I keep a short sliding window (last N samples).
- I compute a “heat” score: when speed drops while current stays high (or spikes), that often indicates a microstall (the motor is working hard but not turning smoothly).
- If the score crosses a threshold, the sentinel prints an alert (and would be the place to trigger a relay, MQTT message, or whatever your factory uses).

### The logic (simple and testable)

For each sample:
- `speed_ratio = speed / nominal_speed`
- `current_ratio = current / nominal_current`
- Heat increases when:
  - speed is low, but
  - current is high (load without motion)

The heat score for a sample:
- `heat = (1 - speed_ratio) * current_ratio`

Then I average heat over the window. If `avg_heat` is above a threshold, we trigger.

### Step-by-step code: edge sentinel in Python

This script is self-contained and includes a **simulator** so you can run it without real hardware. Later, swapping the simulator with a real Modbus client is straightforward (the alarm logic stays the same).

```python
import random
import time
from collections import deque

class ModbusSentinel:
    def __init__(
        self,
        nominal_speed_rpm=6000,
        nominal_current_a=5.0,
        window_size=30,
        heat_threshold=0.35,
    ):
        """
        nominal_speed_rpm: expected 'healthy' spindle speed
        nominal_current_a: expected 'healthy' spindle current
        window_size: how many samples to average over (poll interval matters)
        heat_threshold: above this, we declare a microstall risk
        """
        self.nominal_speed_rpm = nominal_speed_rpm
        self.nominal_current_a = nominal_current_a
        self.window_size = window_size
        self.heat_threshold = heat_threshold

        self.heat_window = deque(maxlen=window_size)

    def add_sample(self, speed_rpm, current_a):
        # Convert to ratios so the logic doesn't care about absolute units too much
        speed_ratio = max(speed_rpm / self.nominal_speed_rpm, 0.0)
        current_ratio = max(current_a / self.nominal_current_a, 0.0)

        # Heat increases when speed is low but current is high
        heat = (1.0 - speed_ratio) * current_ratio

        self.heat_window.append(heat)

        avg_heat = sum(self.heat_window) / len(self.heat_window)
        return avg_heat

    def check_alarm(self, avg_heat):
        return avg_heat >= self.heat_threshold


def simulated_modbus_read():
    """
    Fake Modbus reads.
    Returns (speed_rpm, current_a).
    Most of the time: speed stays near nominal and current stays moderate.
    Occasionally: simulate a microstall:
      - speed dips
      - current spikes or stays high
    """
    nominal_speed = 6000
    nominal_current = 5.0

    # Random noise
    speed = nominal_speed + random.uniform(-150, 150)
    current = nominal_current + random.uniform(-0.5, 0.5)

    # Inject a microstall event with some probability
    if random.random() < 0.12:
        # Speed dips (motor not moving as expected)
        speed *= random.uniform(0.65, 0.85)
        # Current rises (load stays high)
        current *= random.uniform(1.15, 1.35)

    return max(speed, 0.0), max(current, 0.0)


def main():
    sentinel = ModbusSentinel(
        nominal_speed_rpm=6000,
        nominal_current_a=5.0,
        window_size=30,
        heat_threshold=0.35,
    )

    poll_interval_s = 0.2  # 200ms polling, typical edge-friendly pace
    for i in range(200):
        speed_rpm, current_a = simulated_modbus_read()

        avg_heat = sentinel.add_sample(speed_rpm, current_a)
        alarm = sentinel.check_alarm(avg_heat)

        print(
            f"[t={i*poll_interval_s:5.1f}s] "
            f"speed={speed_rpm:7.1f} rpm  current={current_a:4.2f} A  "
            f"avg_heat={avg_heat:0.3f}  alarm={alarm}"
        )

        if alarm:
            print(">>> MICROSTALL RISK DETECTED (edge sentinel alarm)")

        time.sleep(poll_interval_s)


if __name__ == "__main__":
    main()
```

#### What I observed when running it
At first, `avg_heat` stays low and `alarm=False`. When the simulator injects a microstall, the next few samples ramp up the window average, and the script flips to `alarm=True` quickly. The “few samples delay” is intentional: averaging reduces false alarms from noise.

### How this maps to real Modbus (register reads)

In a real setup, you’d replace `simulated_modbus_read()` with actual Modbus calls. The alarm logic stays the same.

If your machine exposes:
- speed at register `40001` (example)
- current at register `40002` (example)

…then your edge loop would read those registers and pass them into `sentinel.add_sample(speed_rpm, current_a)`.

### One tiny improvement I learned the hard way: don’t average forever

I initially used a larger window, and alerts came too late. Microstalls are short and frequent in some processes, so I switched to a modest window (`window_size=30` at 200ms polling = ~6 seconds of context). That gave me:
- fast enough detection
- stable enough readings to ignore small fluctuations

### Wrap-up

I built an edge “microstall sentinel” that converts Modbus-like spindle telemetry (speed + current) into a sliding-window heat score and triggers an immediate alarm when the pattern looks like load without motion. The big takeaway from tinkering: keep the detection logic simple, run it locally at the edge, and use a short averaging window so you catch brief events without drowning in noise.

Edge Computing & Physical Ai Posts

Edge Ai Scrap Detection For Braille-Stamped Circuit Boards With On-Device Yolov8

Building A 5G Edge Timestamp Correlator For Drone Video Using Ptp And Rtp Sequencing

Building A Markerless “Floor Compass” For Ar Using Real Camera Motion Only

Edge Vision For Predicting Conveyor Belt Tear Using Tiny Yolov8 And Imu Correlation

Edge Device Sentinels Using Modbus Heat-Map Alarms For Cnc Spindle Microstalls