## 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

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