simulator.py

import paho.mqtt.client as mqtt
import json
import time
import random
import os
from datetime import datetime
from faker import Faker

fake = Faker()

# MQTT Configuration
MQTT_BROKER = os.getenv('MQTT_BROKER', 'localhost')
MQTT_PORT = int(os.getenv('MQTT_PORT', '1883'))
MQTT_USERNAME = os.getenv('MQTT_USERNAME', None)
MQTT_PASSWORD = os.getenv('MQTT_PASSWORD', None)
MQTT_CLIENT_ID = f"iiot-simulator-{fake.uuid4()[:8]}"

# UNS Namespace Configuration
UNS_NAMESPACE = "UNS/manufacturing/plant1/area1/machine1"

# Tag definitions with initial values and ranges
TAGS = {
    "temperature": {
        "description": "Machine body temperature",
        "unit": "°C",
        "min": 20,
        "max": 95,
        "current": 45,
        "trend": random.uniform(-0.5, 0.5)
    },
    "pressure": {
        "description": "Hydraulic pressure",
        "unit": "PSI",
        "min": 0,
        "max": 350,
        "current": 150,
        "trend": random.uniform(-2, 2)
    },
    "vibration": {
        "description": "Bearing vibration level",
        "unit": "mm/s",
        "min": 0,
        "max": 20,
        "current": 5,
        "trend": random.uniform(-0.1, 0.1)
    },
    "motor_speed": {
        "description": "Motor RPM",
        "unit": "RPM",
        "min": 0,
        "max": 3000,
        "current": 1500,
        "trend": random.uniform(-50, 50)
    },
    "power_consumption": {
        "description": "Electrical power consumption",
        "unit": "kW",
        "min": 0,
        "max": 50,
        "current": 25,
        "trend": random.uniform(-1, 1)
    },
    "humidity": {
        "description": "Ambient humidity",
        "unit": "%",
        "min": 20,
        "max": 80,
        "current": 50,
        "trend": random.uniform(-0.5, 0.5)
    },
    "machine_status": {
        "description": "Machine operating status",
        "unit": "0=stopped, 1=running",
        "min": 0,
        "max": 1,
        "current": 1,
        "trend": 0
    },
    "bearing_temperature": {
        "description": "Bearing temperature",
        "unit": "°C",
        "min": 30,
        "max": 120,
        "current": 65,
        "trend": random.uniform(-0.3, 0.3)
    },
    "hydraulic_oil_temp": {
        "description": "Hydraulic oil temperature",
        "unit": "°C",
        "min": 20,
        "max": 85,
        "current": 55,
        "trend": random.uniform(-0.4, 0.4)
    },
    "production_rate": {
        "description": "Production rate",
        "unit": "units/hour",
        "min": 0,
        "max": 500,
        "current": 250,
        "trend": random.uniform(-10, 10)
    }
}


def on_connect(client, userdata, flags, rc):
    """Callback when client connects to broker."""
    if rc == 0:
        print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Connected to MQTT broker at {MQTT_BROKER}:{MQTT_PORT}")
    else:
        print(f"Failed to connect, return code {rc}")


def on_disconnect(client, userdata, rc):
    """Callback when client disconnects from broker."""
    if rc != 0:
        print(f"Unexpected disconnection. Return code: {rc}")


def on_publish(client, userdata, mid):
    """Callback when message is published."""
    pass


def update_tag_value(tag_name, tag_data):
    """Update tag value with realistic trending."""
    current = tag_data["current"]
    trend = tag_data["trend"]
    min_val = tag_data["min"]
    max_val = tag_data["max"]

    # Update current value with trend
    current += trend

    # Add some random noise for realism
    if random.random() < 0.3:  # 30% chance to change trend direction
        tag_data["trend"] = random.uniform(
            (min_val - current) * 0.01,
            (max_val - current) * 0.01
        )

    # Clamp to min/max
    current = max(min_val, min(max_val, current))

    tag_data["current"] = current
    return current


def generate_payload(tag_name, value):
    """Generate MQTT payload for a tag."""
    tag_data = TAGS[tag_name]

    # For integer tags like machine_status and production_rate
    if isinstance(tag_data["current"], int) or tag_name == "machine_status":
        if tag_name == "machine_status":
            value = int(random.choices([0, 1], weights=[10, 90])[0])
        else:
            value = int(value)
    else:
        value = round(value, 2)

    payload = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "tag": tag_name,
        "value": value,
        "unit": tag_data["unit"],
        "description": tag_data["description"]
    }

    return json.dumps(payload)


def publish_tags(client):
    """Publish all tag values to MQTT broker."""
    for tag_name in TAGS:
        topic = f"{UNS_NAMESPACE}/{tag_name}"
        value = update_tag_value(tag_name, TAGS[tag_name])
        payload = generate_payload(tag_name, value)

        result = client.publish(topic, payload, qos=1)

        if result.rc == mqtt.MQTT_ERR_SUCCESS:
            print(f"[{datetime.now().strftime('%H:%M:%S')}] Published {tag_name}: {value} to {topic}")
        else:
            print(f"Failed to publish to {topic}: {mqtt.error_string(result.rc)}")


def main():
    """Main loop to continuously publish tag data."""
    # Create MQTT client
    client = mqtt.Client(client_id=MQTT_CLIENT_ID)
    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.on_publish = on_publish

    # Set credentials if provided
    if MQTT_USERNAME and MQTT_PASSWORD:
        client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)

    # Retry connection with exponential backoff
    max_retries = 10
    retry_delay = 2
    for attempt in range(max_retries):
        try:
            print(f"Connecting to MQTT broker at {MQTT_BROKER}:{MQTT_PORT}... (attempt {attempt + 1}/{max_retries})")
            client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
            break
        except (ConnectionRefusedError, OSError) as e:
            if attempt < max_retries - 1:
                print(f"Connection failed, retrying in {retry_delay}s...")
                time.sleep(retry_delay)
                retry_delay = min(retry_delay * 1.5, 10)
            else:
                raise

    try:
        client.loop_start()

        # Give broker time to connect
        time.sleep(2)

        # Publish tags every second
        print(f"Starting to publish {len(TAGS)} tags every second...")
        print(f"Namespace: {UNS_NAMESPACE}\n")

        while True:
            publish_tags(client)
            time.sleep(1)

    except KeyboardInterrupt:
        print("\nShutting down...")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()