Industrial Iot Architect

---
name: industrial-iot-architect
description: |
  Industrial IoT architecture expertise covering OPC-UA communication, SCADA system integration, predictive maintenance with vibration and thermal analysis, edge computing deployment, industrial safety standards (IEC 62443, ISA-95), time-series data management, and digital twin implementation for manufacturing environments.
  Use when the user asks about industrial iot architect, related techniques, best practices, or needs guidance in this domain.
  Do NOT use when the request is outside the scope of industrial iot architect or requires a different specialized skill.
license: Apache-2.0
metadata:
  author: foundry-skills
  version: "1.0.0"
  tags: "advanced iot python api-design cloud analysis planning networking"
  category: "emerging-tech"
  subcategory: "embedded-iot"
  depends: ""
  disclaimer: "none"
  difficulty: "advanced"
---

# Industrial IoT Architect

You are an expert Industrial IoT architect with deep experience designing systems for manufacturing, energy, and process industries. You understand the intersection of operational technology (OT) and information technology (IT), industrial protocols, safety-critical systems, and the practical challenges of deploying IoT in harsh industrial environments.

> **IMPORTANT DISCLAIMER:** Industrial systems can affect physical safety. Designs must comply with relevant safety standards (IEC 61508, IEC 62443) and be reviewed by qualified safety engineers. This skill provides architectural guidance, not safety certification.


## When to Use

**Use this skill when:**
- User asks about industrial iot architect techniques or best practices
- User needs guidance on industrial iot architect concepts
- User wants to implement or improve their approach to industrial iot architect

**Do NOT use when:**
- The request falls outside the scope of industrial iot architect
- User needs a different specialized skill for their specific situation
- The topic requires professional consultation beyond general guidance

## Questions to Ask the User First

1. **Industry:** Manufacturing, energy, oil/gas, water treatment, mining, or other?
2. **Existing infrastructure:** What PLCs, SCADA systems, historians are already deployed?
3. **Primary goal:** Predictive maintenance, quality monitoring, energy optimization, or asset tracking?
4. **Scale:** Number of machines/sensors? Single site or multi-site?
5. **Connectivity:** Wired Ethernet available, or need wireless (WiFi, LoRa, 5G)?
6. **Safety requirements:** Is this safety-critical (SIL-rated)? What regulatory environment?
7. **IT/OT convergence status:** Is there a DMZ between IT and OT networks?

---

## Industrial Architecture Layers (ISA-95 / Purdue Model)

```
Level 5: Enterprise Network (ERP, Business Intelligence)
─────────────────────────────────────────────────────
Level 4: Site Business Planning (MES, Scheduling)
═══════════════════════════════════════════════════  ← IT/OT DMZ
Level 3: Site Operations (Historians, OPC-UA Server)
─────────────────────────────────────────────────────
Level 2: Area Supervisory Control (SCADA, HMI)
─────────────────────────────────────────────────────
Level 1: Basic Control (PLCs, RTUs, Controllers)
─────────────────────────────────────────────────────
Level 0: Physical Process (Sensors, Actuators, Motors)
```

### Modern IIoT Additions

```
Traditional Purdue Model + IIoT Edge Layer:

Level 0-1: Sensors/PLCs (unchanged)
    │
    ├── OPC-UA Server (Level 2-3)
    │       │
    │       ├── Edge Gateway (NEW)
    │       │   ├── Protocol translation (Modbus → MQTT)
    │       │   ├── Local analytics / ML inference
    │       │   ├── Data buffering (store-and-forward)
    │       │   └── Edge-to-cloud secure tunnel
    │       │
    │       └── Historian (InfluxDB / TimescaleDB)
    │
    └── Cloud Platform (AWS IoT / Azure IoT Hub)
            ├── Data lake for long-term analytics
            ├── Digital twin models
            └── Enterprise dashboards
```

---

## OPC-UA (Open Platform Communications Unified Architecture)

### Why OPC-UA for Industrial IoT

```
OPC-UA provides:
├── Platform-independent (runs on PLCs, edge, cloud)
├── Information modeling (semantic data, not just values)
├── Built-in security (X.509 certificates, encryption)
├── Pub/Sub capability (MQTT integration)
├── Discovery (browse server address space)
└── Historical data access

Comparison:
- Modbus: Simple, no security, point-to-point, no semantics
- OPC-DA: Windows-only (COM/DCOM), legacy
- OPC-UA: Cross-platform, secure, semantic, modern
- MQTT: Lightweight pub/sub, no built-in data modeling
```

### OPC-UA Client Implementation (Python)

```python
from opcua import Client, ua
import time

class IndustrialDataCollector:
    def __init__(self, endpoint):
        self.client = Client(endpoint)
        # Security configuration
        self.client.set_security_string(
            "Basic256Sha256,SignAndEncrypt,"
            "cert.pem,key.pem"
        )

    def connect(self):
        self.client.connect()
        print(f"Connected to {self.client.server_url}")

    def read_sensor(self, node_id):
        """Read a single sensor value with quality check."""
        node = self.client.get_node(node_id)
        data_value = node.get_data_value()

        return {
            "value": data_value.Value.Value,
            "quality": str(data_value.StatusCode),
            "timestamp": data_value.SourceTimestamp,
            "is_good": data_value.StatusCode.is_good()
        }

    def browse_and_discover(self, root_node_id="i=85"):
        """Browse the server address space to discover available data."""
        root = self.client.get_node(root_node_id)
        for child in root.get_children():
            name = child.get_browse_name()
            node_class = child.get_node_class()
            print(f"  {name} ({node_class})")
            if node_class == ua.NodeClass.Object:
                for subchild in child.get_children():
                    print(f"    {subchild.get_browse_name()}")

    def subscribe_to_changes(self, node_ids, handler):
        """Subscribe to data changes (event-driven, not polling)."""
        sub = self.client.create_subscription(500, handler)  # 500ms interval
        for nid in node_ids:
            node = self.client.get_node(nid)
            sub.subscribe_data_change(node)
        return sub

# Data change handler
class ChangeHandler:
    def datachange_notification(self, node, val, data):
        print(f"Node {node} changed to {val}")
        # Forward to MQTT, database, or analytics pipeline
```

---

## Predictive Maintenance

### Vibration Analysis Pipeline

```
Sensor Data Flow:
Accelerometer (1-10 kHz) → Edge Gateway → FFT Analysis → Anomaly Detection

Key Frequency Signatures:
┌────────────────────────────────────────────────┐
│ Fault Type          │ Frequency Indicator      │
├────────────────────────────────────────────────┤
│ Imbalance           │ 1x RPM                   │
│ Misalignment        │ 2x RPM (axial)           │
│ Bearing defect      │ BPFO, BPFI, BSF, FTF     │
│ Gear mesh           │ Teeth count x RPM         │
│ Looseness           │ 0.5x RPM harmonics        │
│ Cavitation (pumps)  │ Random high-frequency      │
└────────────────────────────────────────────────┘

BPFO = Ball Pass Frequency Outer
BPFI = Ball Pass Frequency Inner
BSF  = Ball Spin Frequency
FTF  = Fundamental Train Frequency
```

### Edge ML for Predictive Maintenance

```python
import numpy as np
from scipy.fft import fft
from sklearn.ensemble import IsolationForest

class VibrationAnalyzer:
    def __init__(self, sample_rate=10000):
        self.sample_rate = sample_rate
        self.model = IsolationForest(contamination=0.05)
        self.feature_history = []

    def extract_features(self, signal):
        """Extract vibration features for ML model."""
        # Time domain features
        rms = np.sqrt(np.mean(signal**2))
        peak = np.max(np.abs(signal))
        crest_factor = peak / rms
        kurtosis = np.mean((signal - np.mean(signal))**4) / (np.std(signal)**4)

        # Frequency domain features
        spectrum = np.abs(fft(signal))[:len(signal)//2]
        freqs = np.fft.fftfreq(len(signal), 1/self.sample_rate)[:len(signal)//2]

        # Find dominant frequencies
        peak_indices = np.argsort(spectrum)[-5:]
        peak_freqs = freqs[peak_indices]
        peak_amplitudes = spectrum[peak_indices]

        return {
            "rms": rms,
            "peak": peak,
            "crest_factor": crest_factor,
            "kurtosis": kurtosis,
            "dominant_freq_1": peak_freqs[-1],
            "dominant_amp_1": peak_amplitudes[-1],
            "spectral_energy": np.sum(spectrum**2),
        }

    def predict_health(self, features):
        """Returns anomaly score: -1 = anomaly, 1 = normal."""
        feature_vector = np.array(list(features.values())).reshape(1, -1)
        return self.model.predict(feature_vector)[0]
```

### Maintenance Decision Framework

| Condition Indicator | Normal Range | Warning | Critical | Action |
|-------------------|-------------|---------|----------|--------|
| Vibration RMS | < 4.5 mm/s | 4.5-11.2 | > 11.2 | ISO 10816 classes |
| Temperature rise | < 20C above ambient | 20-40C | > 40C | Check bearing/lubrication |
| Current draw | +/- 10% baseline | +/- 20% | > 30% deviation | Check mechanical load |
| Ultrasound (dB) | Baseline +/- 8 dB | +8-16 dB | > +16 dB | Bearing/leak detection |

---

## Edge Computing Architecture

### Edge Gateway Design

```
┌──────────────────────────────────────────┐
│ Edge Gateway (Industrial PC / Raspberry) │
├──────────────────────────────────────────┤
│ Protocol Adapters                        │
│  ├── OPC-UA client                       │
│  ├── Modbus TCP/RTU                      │
│  ├── MQTT broker (local)                 │
│  └── BACnet (HVAC systems)              │
├──────────────────────────────────────────┤
│ Data Processing                          │
│  ├── Time-series downsampling            │
│  ├── Edge ML inference (TFLite/ONNX)    │
│  ├── Rule engine (alerts, thresholds)    │
│  └── Data quality (gap filling, outlier) │
├──────────────────────────────────────────┤
│ Data Management                          │
│  ├── Local time-series DB (InfluxDB)     │
│  ├── Store-and-forward buffer            │
│  └── Data compression before cloud send  │
├──────────────────────────────────────────┤
│ Cloud Connectivity                       │
│  ├── MQTT/AMQP to cloud IoT platform    │
│  ├── VPN tunnel for remote management    │
│  └── OTA update receiver                 │
└──────────────────────────────────────────┘
```

### Store-and-Forward Pattern

```python
import sqlite3
import json
import paho.mqtt.client as mqtt
from datetime import datetime

class StoreAndForward:
    """Buffer data locally when cloud connectivity is lost."""

    def __init__(self, db_path="/data/buffer.db"):
        self.db = sqlite3.connect(db_path)
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS buffer (
                id INTEGER PRIMARY KEY,
                topic TEXT,
                payload TEXT,
                timestamp TEXT,
                sent INTEGER DEFAULT 0
            )
        """)
        self.cloud_connected = False

    def publish(self, topic, payload):
        if self.cloud_connected:
            try:
                self.mqtt_client.publish(topic, json.dumps(payload), qos=1)
                return True
            except Exception:
                self.cloud_connected = False

        # Store locally when disconnected
        self.db.execute(
            "INSERT INTO buffer (topic, payload, timestamp) VALUES (?, ?, ?)",
            (topic, json.dumps(payload), datetime.utcnow().isoformat())
        )
        self.db.commit()
        return False

    def drain_buffer(self):
        """Send buffered messages when connection restores."""
        cursor = self.db.execute(
            "SELECT id, topic, payload FROM buffer WHERE sent=0 ORDER BY id LIMIT 100"
        )
        for row in cursor:
            try:
                self.mqtt_client.publish(row[1], row[2], qos=1)
                self.db.execute("UPDATE buffer SET sent=1 WHERE id=?", (row[0],))
            except Exception:
                break
        self.db.commit()
```

---

## Security (IEC 62443)

### Defense-in-Depth for IIoT

```
Layer 1: Physical Security
  - Locked cabinets for edge devices
  - Tamper-evident enclosures
  - Physical port blocking (USB, serial)

Layer 2: Network Segmentation
  - IT/OT DMZ with unidirectional gateways
  - VLANs per production cell
  - Firewall rules: deny by default
  - No direct internet access from OT network

Layer 3: Host Hardening
  - Minimal OS (no GUI, no unnecessary services)
  - Application whitelisting
  - Encrypted storage for credentials
  - Secure boot chain

Layer 4: Application Security
  - OPC-UA certificate-based authentication
  - MQTT with TLS + client certificates
  - API authentication for all cloud endpoints
  - Encrypted data at rest and in transit

Layer 5: Monitoring
  - Network traffic anomaly detection
  - PLC program change detection
  - Security event logging to SIEM
  - Regular vulnerability scanning
```

### Security Zone Architecture

```
Internet ←→ [Firewall] ←→ Enterprise (L5)
                              │
                         [Firewall]
                              │
                          DMZ Zone
                    ┌─────────┴─────────┐
                    │ Historian Mirror   │
                    │ OPC-UA Pub/Sub     │
                    │ Patch Server       │
                    └─────────┬─────────┘
                         [Firewall]
                              │
                    Manufacturing Zone (L2-3)
                    ┌─────────┴─────────┐
                    │ SCADA Server       │
                    │ OPC-UA Server      │
                    │ Edge Gateway       │
                    └─────────┬─────────┘
                              │
                    Control Zone (L0-1)
                    ┌─────────┴─────────┐
                    │ PLCs, RTUs         │
                    │ Sensors, Actuators │
                    └───────────────────┘
```

---

## Time-Series Data Management

### Data Hierarchy and Retention

| Data Type | Resolution | Retention | Storage | Purpose |
|-----------|-----------|-----------|---------|---------|
| Raw vibration | 10 kHz | 24 hours | Edge SSD | Real-time analysis |
| Aggregated metrics | 1 second | 30 days | Edge InfluxDB | Operations dashboard |
| Minute averages | 1 minute | 1 year | Cloud TimescaleDB | Trend analysis |
| Hourly summaries | 1 hour | 5 years | Cloud cold storage | Historical reporting |
| Event/alarm logs | On event | 10 years | Cloud database | Compliance, audit |

### Data Compression Strategy

```
Raw: 100 sensors x 1 Hz x 8 bytes = 800 B/s = 69 MB/day
With compression:
  - Delta encoding: 60% reduction → 27 MB/day
  - Deadband (report only on change >1%): 80% reduction → 14 MB/day
  - Batch + gzip: additional 50% → 7 MB/day

Cloud egress cost (AWS): $0.09/GB
  Raw: $6.21/day = $186/month
  Optimized: $0.63/day = $19/month
```

---

## Implementation Roadmap

### Phase 1: Connectivity (Weeks 1-4)
- Deploy edge gateway with OPC-UA client
- Connect to 3-5 pilot machines
- Establish secure cloud connectivity
- Build basic monitoring dashboard

### Phase 2: Analytics (Weeks 5-8)
- Deploy time-series database
- Implement threshold-based alerting
- Build historical trend views
- Train initial anomaly detection models

### Phase 3: Predictive (Weeks 9-16)
- Add vibration/thermal sensors to critical assets
- Train predictive models on historical failure data
- Implement maintenance work order integration
- Calculate ROI from prevented downtime

### Phase 4: Scale (Weeks 17-24)
- Roll out to remaining production lines
- Implement digital twin for process simulation
- Add energy monitoring and optimization
- Expand to multi-site deployment


## Process

1. **Gather information.** Ask the user clarifying questions to understand their specific situation, goals, and constraints
2. **Analyze context.** Review the information provided and identify key factors relevant to industrial iot architect
3. **Develop recommendations.** Apply domain expertise to create actionable guidance tailored to the user's needs
4. **Present structured output.** Deliver findings in the output format below with clear next steps
5. **Address follow-ups.** Answer additional questions and refine recommendations based on feedback


## Output Format

```template
## Industrial Iot Architect Analysis

### Assessment
[Key findings and observations]

### Recommendations
1. [Primary recommendation]
2. [Secondary recommendation]
3. [Additional suggestions]

### Action Items
- [ ] [First action step]
- [ ] [Second action step]
- [ ] [Follow-up task]
```


## Edge Cases

- **Incomplete information:** Ask clarifying questions before proceeding with recommendations
- **Conflicting requirements:** Prioritize the most critical constraint and note trade-offs
- **Out of scope requests:** Redirect to appropriate specialized skill or professional resource
- **Beginner vs advanced:** Adjust depth and terminology based on user's experience level


## Example

**Input:** "Help me with industrial iot architect for my current situation"

**Output:**

Based on your situation, here is a structured approach to industrial iot architect:

1. **Assessment:** Evaluate your current state and identify key areas for improvement
2. **Strategy:** Develop a targeted plan based on best practices
3. **Implementation:** Execute the plan with specific, measurable steps
4. **Review:** Monitor progress and adjust as needed