IoT多设备模拟：Python实现容器填充度差异化递增逻辑

Got it, let's tackle this problem step by step. You need to simulate a gradual fill level increase for 50 IoT devices—each with their own timeline to hit 100%—while sending JSON updates every 20 seconds. Here's a practical, flexible implementation that fits your needs perfectly:

The idea is to track each device's unique "fill trend" (how many days it takes to go from 0% to 100%) and calculate its current fill level based on how much time has passed since it started. We'll add a touch of randomness to make the sensor data feel more realistic, and persist device state so the simulation doesn't reset if your script restarts.

import time
import json
import random
from datetime import datetime, timedelta
import os

# Configuration
NUM_DEVICES = 50
UPDATE_INTERVAL = 20  # Seconds
STATE_FILE = "device_states.json"
MIN_TREND_DAYS = 1
MAX_TREND_DAYS = 7  # Adjust to control how spread out fill levels are

def initialize_devices():
    """Create initial state for all devices if no state file exists"""
    devices = {}
    for i in range(NUM_DEVICES):
        device_id = f"master_device_{i:03d}"
        # Assign a random trend (days to reach 100%)
        trend_days = random.uniform(MIN_TREND_DAYS, MAX_TREND_DAYS)
        # Record start time (UTC for consistency)
        start_time = datetime.utcnow().isoformat()
        devices[device_id] = {
            "start_time": start_time,
            "trend_days": trend_days,
            "last_fill_level": 0.0
        }
    # Save initial state to file
    with open(STATE_FILE, "w") as f:
        json.dump(devices, f, indent=2)
    return devices

def load_device_states():
    """Load existing device state from file, or initialize if missing"""
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            return json.load(f)
    else:
        return initialize_devices()

def calculate_fill_level(device_state):
    """Calculate current fill level based on elapsed time and trend"""
    start_time = datetime.fromisoformat(device_state["start_time"])
    current_time = datetime.utcnow()
    elapsed_days = (current_time - start_time).total_seconds() / (24 * 3600)
    trend_days = device_state["trend_days"]
    
    # Linear fill calculation
    fill_level = (elapsed_days / trend_days) * 100
    # Cap at 100% and add small random noise for realism
    fill_level = min(fill_level, 100.0)
    fill_level += random.uniform(-2.0, 2.0)
    # Ensure we don't go below 0 or above 100
    fill_level = max(0.0, min(fill_level, 100.0))
    return round(fill_level, 1)

def send_iot_message(device_id, fill_level, trend_days):
    """Simulate sending JSON message to IoT infrastructure"""
    # Replace this with your actual IoT send logic
    message = {
        "device_id": device_id,
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "sensor_readings": {
            # Add your other simulated sensor values here
            "temperature": round(random.uniform(20.0, 25.0), 1),
            "humidity": round(random.uniform(30.0, 60.0), 1),
            "fill_level": fill_level
        },
        "trend_days": round(trend_days, 1)
    }
    print(f"Sending message for {device_id}:")
    print(json.dumps(message, indent=2))
    # Example: If using MQTT, you'd publish here: client.publish(topic, json.dumps(message))

def main():
    devices = load_device_states()
    print(f"Initialized {NUM_DEVICES} devices with random fill trends")
    
    while True:
        for device_id, state in devices.items():
            fill_level = calculate_fill_level(state)
            # Update state with latest fill level (optional, but useful for debugging)
            state["last_fill_level"] = fill_level
            send_iot_message(device_id, fill_level, state["trend_days"])
        
        # Save updated state (in case script restarts)
        with open(STATE_FILE, "w") as f:
            json.dump(devices, f, indent=2)
        
        time.sleep(UPDATE_INTERVAL)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nSimulation stopped by user")

• Persistent Device State: The device_states.json file stores each device's start time and trend, so if your script restarts, it picks up right where it left off instead of resetting all fill levels to 0.
• Unique Trends for Each Device: Every device gets a random trend between 1-7 days (adjust MIN_TREND_DAYS and MAX_TREND_DAYS to tweak this). This means after 24 hours, a device with a 1-day trend will be at ~100%, while one with a 7-day trend will be at ~14%—exactly the varying fill levels you want.
• Realistic Noise: We add ±2% random noise to the fill level so it doesn't look perfectly linear (real sensors rarely have perfectly smooth data).
• Easy Integration: The send_iot_message function is where you'd plug in your actual IoT transport (MQTT, HTTP API, etc.)—right now it just prints the message for testing.

• Fill Level Capping: Once a device hits 100%, it stays there (no overfilling).
• Restart Resilience: The state file ensures you don't lose progress if the script crashes or is restarted.
• Non-Negative Fill: The calculation ensures fill levels never drop below 0% even with random noise.

Run the script, and you'll see JSON messages printed every 20 seconds for each device. After a few hours, you'll notice distinct fill levels across devices based on their individual trends.

内容的提问来源于stack exchange，提问作者CoolDive