description
Forklift-safety rule: detect a Material Handling Equipment unit (forklift / reach truck) TRAVELLING while its load (product package) is raised above the vehicle base — a classic tip-over / dropped-load hazard. Works purely on geometry over tracked boxes: associates the package box to its carrying MHE, measures the package's vertical offset vs the vehicle base (base_y_distance / increased_y_distance) and the vehicle's travel angle/velocity (degrees_threshold, prd_move_threshold); event confidence is a weighted blend of MHE/product confidence and geometry terms. An optional sklearn postprocess gate re-validates each candidate on the last-5 bbox features (fail-open if the service is down). Per-track, immediate emit.
use cases
- warehouse forklift safety — "no driving with raised forks/load" policy enforcement
- reach-truck operation audits in racking aisles
- any moving-vehicle-with-elevated-load hazard (adaptable classes via config)
limitations
- needs a detector emitting BOTH the vehicle classes and the load/package class
- camera should view the vehicle side-on enough that raised-load geometry is visible
- does not measure actual load height in metres — thresholds are relative to bbox geometry
config parameters
zone
| Parameter | Type | Default | Notes |
|---|---|---|---|
| cam_zones * | array of array of pair of integer | [] |
advanced
| Parameter | Type | Default | Notes |
|---|---|---|---|
| max_miss * | integer | 40 | |
| degrees_threshold * | number | 75 | |
| bbox_confidence * | number | 0.7 | |
| min_length_bboxes * | integer | 2 | |
| min_height_threshold * | number | 0.07 | |
| min_width_threshold * | number | 0.05 | |
| max_height_threshold * | number | 0.6 | |
| max_width_threshold * | number | 0.4 | |
| cam_resolution * | array of integer | [1920, 1080] | [width, height] (min items 2, max items 2) |
| target_class * | array of string | ["forklift", "reach_truck", "product_package"] | |
| prd_move_threshold * | number | 0.5 | |
| base_y_distance * | number | 0.8 | |
| increased_y_distance * | number | 1.5 | |
| MHE_target_class * | array of string | ["forklift", "reach_truck"] | |
| product_target_class * | array of string | ["product_package"] | |
| event_confidence_alpha * | number | 0.2 | |
| event_confidence_beta * | number | 0.3 | |
| event_confidence_gamma * | number | 0.3 | |
| event_confidence_delta * | number | 0.2 |
golden replay cases
| Case | Frames | Expected events |
|---|---|---|
| positive | 3 | 1 |
readme
mhe10
MHE10 — "moving while load at height": detect a forklift / reach truck travelling while carrying a product raised above the vehicle base.
A rule module (code + co-located module.yaml + config.schema.json + defaults.yaml) built on
celia-rule-sdk and registered as MHE10 via the celia.rules entry point. It is a faithful port of
old_srcs/eme-ai/cloud_app/internal/controller/lfvn/mhe10/camera.py: the conditions and confidence
formula are preserved line-for-line; only the plumbing changed (embedded ByteTrack → injected
Tracker, magic-key config → typed MHE10Config, requests.post postprocess → an injected validator,
event_checking(*args) -> tuple → check(tracks, frame_id) -> list[EventCandidate]).
How it plugs in
Discovered by a rule worker via the celia.rules entry point (MHE10Module); the worker feeds it
tracks and publishes the EvidenceRequests it returns. The algorithm: zone-filter tracks → classify
MHE vs product → map product↔MHE intersections → per-track gate → check (intersection, motion angle,
travel, lift threshold, optional ML gate) → weighted confidence.
Configuration
module.yaml: rule_code: MHE10, domain: warehouse, emission_strategy: immediate. Requires the
warehouse_detector model and tracking: bytetrack (native for offline replay).
Config keys (config.schema.json + defaults.yaml): bbox_confidence, min_length_bboxes,
degrees_threshold, {min,max}_{height,width}_threshold, prd_move_threshold, base_y_distance,
increased_y_distance, MHE_target_class, product_target_class, and the
event_confidence_{alpha,beta,gamma,delta} weights.
Postprocess ML gate
An optional sklearn classifier over the last-5 forklift/product boxes (15 features: vertical gaps +
box areas) confirms or rejects a candidate event. Fail-open: emit unless it explicitly returns
False. It runs in-process by default (the rule worker is already Python — no separate service):
pip install "./modules/rules[mhe10-validator]" # scikit-learn + joblib + numpy
# fetch the model and point the worker at it
wget https://minio.emagiceyes.rainscales.com/cloud-ai-models/infer/third_party_model/model.pkl \
-O /opt/celia/mhe10/model.pkl
MHE10_POSTPROCESS_MODEL_PATH=/opt/celia/mhe10/model.pkl # -> in-process SklearnValidator (default)
For the rare case where the model is heavy enough to deserve its own container, the same Protocol has
an HTTP variant — set MHE10_POSTPROCESS_URL instead (pip install "...[postprocess]"). With neither
env set, the gate is a fail-open no-op (AcceptAllValidator), which is what golden replay uses.
Testing
pytest modules/rules/mhe10 # 12 offline tests via the celia-replay harness
One positive golden replay (events diff == 0) plus per-branch negatives, the postprocess gate, and
edge cases. The golden case is synthetic; bit-fidelity vs the legacy worker still needs recorded real
Kafka fixtures (golden-replay, ADR-0008).