Selected Python exercises inspired by Industrial IoT (IIoT) patterns: MQTT/UNS-style naming, alarm routing, signal validation, and KPI calculations. The goal of this repo is clarity + realistic plant logic, not building a big framework.
Function-based exercises with an IIoT mindset:
- Type hints (basic), runtime checks, dicts,
**kwargs - MQTT/UNS-style topic building
- Report generation with dynamic keys (
approved_*,rejected_*) - Scrap rate calculation
- Alarm routing logic with practical runtime checks
MQTT Topic Normalizer (dirty input → clean UNS-style topic):
- Split topics into segments
- Normalize segments (strip/lower/space →
_) - Remove invalid segments (e.g. empty segments caused by
//) - Validate the final structure (
site/line/cell/tag) and allowed tags
✅ This exercise uses a full pipeline approach with map + lambda + filter for progressive improvement in data processing style.
Actionable Event Router (events → unified structure):
- Drop stale events (time-based filtering)
- Keep only actionable events:
- telemetry: value convertible to
float - alarms: dict payload with
sev >= 2
- telemetry: value convertible to
- Transform accepted events into a unified dict (ready for storage / dashboards)
✅ Also built using pipelines (map + filter + lambda) to keep the logic explicit and easy to reason about.
A more robust data pipeline that:
- Uses
map+lambdato transform process variables (including unit conversion and type validation) - Validates sensor values (avoids issues like
boolbehaving likeint) - Calculates scrap percentage and triggers alarms (e.g., HIGH_SCRAP)
- Handles edge cases such as zero production (avoids
ZeroDivisionError)
-
TopicBuilder- Class for building consistent UNS-style MQTT topics
- Includes topic normalization and cell updates without creating a new instance
-
AlarmEvent + AlarmPolicy- Separation between alarm data and policy decision logic
- Simple critical check and actionable filtering
-
ShiftReport- Shift production summary with total units, scrap rate, warnings, and dictionary export
Python 3.10+ recommended. No external dependencies.
python ex_funcs_iiot.py python ex_mqtt_topic_normalizer.py python ex_actionable_event_router.py python ex_map_lambda_kpi.py
This repository is a learning log and a small portfolio showcase focused on:
- clean Python fundamentals
- realistic IIoT data handling
- Consistent naming (MQTT / UNS-like conventions)
- Practical validation and robustness for “messy real-world data”
These are standalone exercises/scripts (not a package). The emphasis is on readable, practical code that scales in complexity over time.