DevOps Applications in Manufacturing Real-World Examples

# mapping.py — ERP material code to MES item code
MAPPING = {
    "ERP-STEEL-10": "MES-ST10",
    "ERP-STEEL-12": "MES-ST12",
}

def to_mes_code(erp_code: str) -> str:
    try:
        return MAPPING[erp_code]
    except KeyError:
        raise ValueError(f"No MES mapping for {erp_code}")

# test_mapping.py — runs in CI on every change
import pytest
from mapping import to_mes_code

def test_known_code_maps_correctly():
    assert to_mes_code("ERP-STEEL-10") == "MES-ST10"

def test_unknown_code_is_rejected():
    with pytest.raises(ValueError):
        to_mes_code("ERP-UNKNOWN")

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY collector.py .
# Read config from environment, not baked into the image
ENV LINE_ID=unset
CMD ["python", "collector.py"]

services:
  collector:
    image: registry.example.local/plant/collector:1.4.0
    restart: unless-stopped
    environment:
      LINE_ID: "line-03"
      MQTT_BROKER: "mqtt.plant.local"
    volumes:
      - ./buffer:/app/buffer   # keep data if network drops

import json
import paho.mqtt.client as mqtt

def on_message(client, userdata, msg):
    reading = json.loads(msg.payload)
    if not {"machine_id", "ts", "temp_c"} <= reading.keys():
        print("bad message, skipped:", reading)
        return
    print(reading["machine_id"], reading["temp_c"])  # store it here

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_message = on_message
client.connect("mqtt.plant.local", 1883)
client.subscribe("plant/line03/+/temperature")
client.loop_forever()

# Prometheus alert: data stopped arriving from a line
groups:
  - name: iiot
    rules:
      - alert: LineDataMissing
        expr: time() - max(last_reading_timestamp{line="line-03"}) > 300
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "No sensor data from line-03 for 5+ minutes"

Change: Update collector on line-03 edge gateway
Version: 1.3.2 -> 1.4.0
Agreed window: Shift change, with production lead
Pre-checks: gateway healthy, buffer folder empty
Steps: pull image, restart service, verify data flow
Success: readings arriving within 2 minutes
Rollback: re-run with image tag 1.3.2
Sign-off: production lead + maintenance

Home lab: mini factory data pipeline
1. Run Mosquitto (MQTT broker) in Docker
2. Write a Python script that publishes fake
   temperature readings for 3 "machines"
3. Subscribe and store readings in TimescaleDB
4. Build a Grafana panel per machine
5. Add an alert when a machine stops sending data

- Built a simulated IIoT pipeline with MQTT,
  TimescaleDB and Grafana, including missing-data alerts
- Containerised an edge data collector and practised
  staged updates with rollback by image tag
- Wrote CI tests for an ERP-to-MES code mapping service