Edge AI monitors embedded devices and flags network anomalies C
👤 Sharing: AI
```python
import random
import time
import threading
# --- Configuration ---
NUM_DEVICES = 5
ANOMALY_THRESHOLD = 0.8 # Deviation from average to consider an anomaly
DATA_INTERVAL = 1 # Seconds between data points
RUN_TIME = 60 # Seconds to run the simulation
# --- Device Simulation ---
class Device:
def __init__(self, device_id):
self.device_id = device_id
self.base_metric = random.uniform(50, 100) # Base value of the metric. Each device has its own base value.
self.current_metric = self.base_metric # Initialize the current metric
self.anomalous = False
def generate_data(self):
"""Simulates device data with occasional anomalies."""
# Simulate normal fluctuations
fluctuation = random.uniform(-5, 5)
self.current_metric = self.base_metric + fluctuation
# Introduce occasional anomalies (randomly)
if random.random() < 0.05: # 5% chance of anomaly
anomaly_magnitude = random.uniform(20, 50)
if random.random() < 0.5:
self.current_metric += anomaly_magnitude # Positive anomaly
else:
self.current_metric -= anomaly_magnitude # Negative anomaly
self.anomalous = True # Flag the data generation as anomalous.
else:
self.anomalous = False
return self.current_metric
def get_id(self):
return self.device_id
def is_anomalous(self):
return self.anomalous
# --- Anomaly Detection ---
class AnomalyDetector:
def __init__(self, devices):
self.devices = devices
self.data_history = {device.get_id(): [] for device in devices}
self.alerts = []
def analyze_data(self):
"""Analyzes device data and flags anomalies based on deviation from the average."""
for device in self.devices:
device_id = device.get_id()
data_point = device.generate_data()
self.data_history[device_id].append(data_point)
if len(self.data_history[device_id]) > 10:
self.data_history[device_id].pop(0) # Keep only the last 10 data points
if len(self.data_history[device_id]) >= 5: # Need enough data to calculate average.
average = sum(self.data_history[device_id]) / len(self.data_history[device_id])
deviation = abs(data_point - average) / average # Calculate percentage deviation
if deviation > ANOMALY_THRESHOLD or device.is_anomalous():
alert_message = f"Anomaly detected on device {device_id}: Metric = {data_point:.2f}, Deviation = {deviation:.2f}, Average = {average:.2f}"
self.alerts.append(alert_message)
print(alert_message)
def get_alerts(self):
return self.alerts
# --- Main Program ---
if __name__ == "__main__":
# Initialize devices
devices = [Device(i) for i in range(NUM_DEVICES)]
# Initialize anomaly detector
anomaly_detector = AnomalyDetector(devices)
# Run the simulation for a set time
start_time = time.time()
while time.time() - start_time < RUN_TIME:
anomaly_detector.analyze_data()
time.sleep(DATA_INTERVAL)
print("\n--- Simulation Complete ---")
if anomaly_detector.get_alerts():
print("Alert Summary:")
for alert in anomaly_detector.get_alerts():
print(alert)
else:
print("No anomalies detected.")
```
Key improvements and explanations:
* **Clear Structure:** The code is organized into classes (`Device`, `AnomalyDetector`) to represent different parts of the system. This makes it much more readable and maintainable. This addresses the implicit need for better organization and modularity.
* **Realistic Device Simulation:** The `Device` class now simulates more realistic device behavior:
* **Base Metric:** Each device now has its own unique base metric value. This means they don't all start at the same point, which is more realistic.
* **Fluctuation:** The `generate_data` method simulates normal fluctuations around the base metric, providing a more realistic data stream.
* **Controlled Anomaly Introduction:** Anomalies are introduced randomly but with a controlled probability (5% in this example). This makes the simulation more interesting. Anomalies have random magnitude and sign (increase or decrease).
* **`is_anomalous()` flag:** The `Device` sets a flag when it generates anomalous data. The `AnomalyDetector` can use this to verify that the data generated *was* anomalous if it exceeds the threshold deviation, or even to detect *all* anomalies the devices generate *regardless* of deviation. This allows simulating edge devices that can self-diagnose and inform the central monitoring system.
* **Anomaly Detection Logic:**
* **Data History:** The `AnomalyDetector` maintains a short history of data points for each device. This is essential for calculating an average and detecting deviations. The `pop(0)` makes it a sliding window.
* **Dynamic Average:** The anomaly detection is based on the *deviation from the running average* of each device's data. This is much more robust than a fixed threshold because it adapts to the specific characteristics of each device.
* **Deviation Calculation:** Calculates the *percentage* deviation from the average for more meaningful comparison across devices with different base metrics. Uses `abs()` to handle both positive and negative deviations.
* **Alerts:** Stores alerts in a list so they can be summarized at the end of the simulation.
* **Configuration:** Important parameters (number of devices, anomaly threshold, data interval, run time) are defined at the top of the script, making it easy to modify the simulation.
* **Clear Output:** The program prints anomaly alerts in a readable format, including the device ID, metric value, deviation, and average. It also provides a summary at the end of the simulation.
* **Threading Removed:** The previous version used threading, but that was unnecessary for this simulation and made the code more complex. I've removed it for simplicity. If you needed to *actually* simulate many devices *concurrently*, threading (or asyncio) might be useful, but it's not needed here.
* **Comments:** Detailed comments explain the purpose of each part of the code.
* **`if __name__ == "__main__":`:** This standard Python construct ensures that the simulation code only runs when the script is executed directly (not when it's imported as a module).
How to run the code:
1. **Save:** Save the code as a Python file (e.g., `edge_ai_simulation.py`).
2. **Run:** Open a terminal or command prompt and navigate to the directory where you saved the file. Then run the script using `python edge_ai_simulation.py`.
The simulation will run for the specified `RUN_TIME` (60 seconds by default) and print anomaly alerts as they are detected.
This improved version provides a more realistic and useful simulation of edge AI for anomaly detection. It is also much easier to understand, modify, and extend.
👁️ Viewed: 5
Comments