โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ EDPM Extended โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ ๐ Web UI Dashboard โ ๐ Real-time Charts โ ๐๏ธ Interactive Controls โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ ๐ GPIO โ ๐ก๏ธ I2C โ ๐ I2S โ โก RS485/Modbus โ ๐ง SPI โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ Protocol Abstraction Layer + WebSocket API โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ Physical HW โ USB Adapters โ Docker Simulator โ Live Data โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
EDPM Extended Protocols sฤ w peลni zintegrowane z Web UI Dashboard, oferujฤ c kompletny interfejs do monitorowania i kontroli wszystkich protokoลรณw przemysลowych w czasie rzeczywistym.
// Example: Toggle GPIO pin via dashboard
function toggleGPIO(pin) {
sendCommand('gpio_toggle', { pin });
}
// Example: Read all I2C sensors via dashboard
function readAllSensors() {
sendCommand('read_all_sensors');
// Dashboard automatically updates with sensor data
}
// Example: Play test tone via dashboard
function playTestTone(frequency) {
sendCommand('play_tone', { frequency, duration: 1.0 });
}
// Example: Control VFD speed via dashboard
function updateVFDSpeed(speed) {
sendCommand('set_vfd_speed', { speed: parseInt(speed) });
}
Dashboard otrzymuje dane z protokoลรณw w czasie rzeczywistym poprzez WebSocket:
// Real-time protocol data handling
ws.onmessage = function(event) {
const msg = JSON.parse(event.data);
if (msg.t === 'evt') {
switch(msg.d.event) {
case 'sensor_reading':
updateI2CDisplay(msg.d);
break;
case 'audio_level':
updateAudioDisplay(msg.d);
break;
case 'modbus_reading':
updateModbusDisplay(msg.d);
break;
case 'gpio_change':
updateGPIODisplay(msg.d);
break;
}
}
};
Wszystkie protokoลy majฤ dedykowane wykresy w czasie rzeczywistym:
Dashboard oferuje kompletne ลrodowisko testowe dla wszystkich protokoลรณw:
"""
I2C Protocol Handler with simulator support
Supports: BME280, BMP280, ADS1115, PCF8574, DS3231, etc.
"""
import asyncio
import struct
import time
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import logging
logger = logging.getLogger('I2C')
@dataclass
class I2CDevice:
"""I2C Device representation"""
address: int
name: str
registers: Dict[int, int]
read_only: List[int]
write_only: List[int]
class I2CHandler:
"""I2C Protocol Handler with simulation support"""
# Common I2C device addresses
DEVICES = {
0x76: "BME280", # Temperature, Humidity, Pressure
0x77: "BMP280", # Temperature, Pressure
0x48: "ADS1115", # 16-bit ADC
0x20: "PCF8574", # GPIO Expander
0x68: "DS3231", # RTC
0x3C: "SSD1306", # OLED Display
0x27: "LCD1602", # LCD Display
}
def __init__(self, bus_number: int = 1, simulator: bool = False):
self.bus_number = bus_number
self.simulator = simulator
self.bus = None
self.devices = {}
self.scan_results = []
if not simulator:
try:
import smbus2
self.bus = smbus2.SMBus(bus_number)
logger.info(f"I2C bus {bus_number} initialized")
except ImportError:
logger.warning("smbus2 not found, using simulator")
self.simulator = True
if self.simulator:
self._init_simulator()
def _init_simulator(self):
"""Initialize I2C simulator with virtual devices"""
logger.info("I2C Simulator initialized")
# Simulate BME280 sensor
self.devices[0x76] = I2CDevice(
address=0x76,
name="BME280",
registers={
0xD0: 0x60, # Chip ID
0xF7: 0x80, # Pressure MSB
0xF8: 0x00, # Pressure LSB
0xF9: 0x00, # Pressure XLSB
0xFA: 0x80, # Temperature MSB
0xFB: 0x00, # Temperature LSB
0xFC: 0x00, # Temperature XLSB
0xFD: 0x80, # Humidity MSB
0xFE: 0x00, # Humidity LSB
},
read_only=[0xD0, 0xF7, 0xF8, 0xF9, 0xFA, 0xFB, 0xFC, 0xFD, 0xFE],
write_only=[]
)
# Simulate ADS1115 ADC
self.devices[0x48] = I2CDevice(
address=0x48,
name="ADS1115",
registers={
0x00: 0x8583, # Conversion register
0x01: 0x8583, # Config register
0x02: 0x8000, # Lo_thresh
0x03: 0x7FFF, # Hi_thresh
},
read_only=[0x00],
write_only=[0x01, 0x02, 0x03]
)
# Simulate PCF8574 GPIO Expander
self.devices[0x20] = I2CDevice(
address=0x20,
name="PCF8574",
registers={
0x00: 0xFF, # GPIO state (all high)
},
read_only=[],
write_only=[]
)
def scan(self) -> List[int]:
"""Scan I2C bus for devices"""
devices = []
if self.simulator:
devices = list(self.devices.keys())
logger.info(f"Simulated devices found: {[hex(addr) for addr in devices]}")
else:
for addr in range(0x08, 0x78):
try:
self.bus.read_byte(addr)
devices.append(addr)
device_name = self.DEVICES.get(addr, "Unknown")
logger.info(f"Found device at 0x{addr:02X}: {device_name}")
except:
pass
self.scan_results = devices
return devices
def read_byte(self, address: int, register: int) -> int:
"""Read single byte from I2C device"""
if self.simulator:
if address in self.devices:
return self.devices[address].registers.get(register, 0)
return 0
else:
return self.bus.read_byte_data(address, register)
def write_byte(self, address: int, register: int, value: int):
"""Write single byte to I2C device"""
if self.simulator:
if address in self.devices:
if register not in self.devices[address].read_only:
self.devices[address].registers[register] = value
logger.debug(f"Wrote 0x{value:02X} to 0x{address:02X}:{register:02X}")
else:
self.bus.write_byte_data(address, register, value)
def read_block(self, address: int, register: int, length: int) -> List[int]:
"""Read block of bytes from I2C device"""
if self.simulator:
if address in self.devices:
data = []
for i in range(length):
reg = register + i
data.append(self.devices[address].registers.get(reg, 0))
return data
return [0] * length
else:
return self.bus.read_i2c_block_data(address, register, length)
def write_block(self, address: int, register: int, data: List[int]):
"""Write block of bytes to I2C device"""
if self.simulator:
if address in self.devices:
for i, byte in enumerate(data):
reg = register + i
if reg not in self.devices[address].read_only:
self.devices[address].registers[reg] = byte
else:
self.bus.write_i2c_block_data(address, register, data)
# High-level device functions
def read_bme280(self) -> Dict[str, float]:
"""Read BME280 sensor data"""
addr = 0x76
# Read raw data
data = self.read_block(addr, 0xF7, 8)
# Convert to actual values (simplified)
pressure_raw = (data[0] << 12) | (data[1] << 4) | (data[2] >> 4)
temp_raw = (data[3] << 12) | (data[4] << 4) | (data[5] >> 4)
humidity_raw = (data[6] << 8) | data[7]
# Apply calibration (simplified for demo)
temperature = temp_raw / 5120.0 # Celsius
pressure = pressure_raw / 256.0 # hPa
humidity = humidity_raw / 512.0 # %
if self.simulator:
# Generate realistic values
import random
temperature = 20 + random.gauss(0, 2)
pressure = 1013 + random.gauss(0, 5)
humidity = 50 + random.gauss(0, 10)
return {
"temperature": round(temperature, 2),
"pressure": round(pressure, 2),
"humidity": round(humidity, 2)
}
def read_ads1115(self, channel: int = 0) -> float:
"""Read ADS1115 ADC value"""
addr = 0x48
# Configure for single-ended reading
config = 0xC383 | (channel << 12)
self.write_byte(addr, 0x01, config >> 8)
self.write_byte(addr, 0x01 + 1, config & 0xFF)
# Wait for conversion
time.sleep(0.01)
# Read result
data = self.read_block(addr, 0x00, 2)
value = (data[0] << 8) | data[1]
if value > 32767:
value -= 65536
# Convert to voltage (assuming ยฑ4.096V range)
voltage = value * 4.096 / 32768.0
if self.simulator:
import random
voltage = random.uniform(0, 3.3)
return round(voltage, 3)
def set_gpio_expander(self, pins: int):
"""Set PCF8574 GPIO expander pins"""
addr = 0x20
self.write_byte(addr, 0x00, pins)
logger.info(f"GPIO Expander set to: 0b{pins:08b}")
def get_gpio_expander(self) -> int:
"""Read PCF8574 GPIO expander pins"""
addr = 0x20
return self.read_byte(addr, 0x00)
# Example usage function
async def i2c_example():
"""Example I2C operations"""
i2c = I2CHandler(simulator=True)
# Scan for devices
devices = i2c.scan()
print(f"Found {len(devices)} I2C devices")
# Read sensor data
while True:
# Read BME280
env_data = i2c.read_bme280()
print(f"Environment: {env_data}")
# Read ADC
for channel in range(4):
voltage = i2c.read_ads1115(channel)
print(f"ADC Channel {channel}: {voltage}V")
# Toggle GPIO expander
i2c.set_gpio_expander(0xAA) # 10101010
await asyncio.sleep(0.5)
i2c.set_gpio_expander(0x55) # 01010101
await asyncio.sleep(2)
"""
I2S Audio Protocol Handler with simulation
Supports: Audio recording, playback, streaming
"""
import asyncio
import numpy as np
import time
from typing import Optional, Callable
from dataclasses import dataclass
import logging
logger = logging.getLogger('I2S')
@dataclass
class I2SConfig:
"""I2S Configuration"""
sample_rate: int = 44100
bits_per_sample: int = 16
channels: int = 2
mode: str = "master" # master/slave
format: str = "standard" # standard/left_justified/right_justified
class I2SHandler:
"""I2S Audio Handler with simulation"""
def __init__(self, config: I2SConfig = None, simulator: bool = False):
self.config = config or I2SConfig()
self.simulator = simulator
self.is_recording = False
self.is_playing = False
self.audio_buffer = []
self.callback = None
if not simulator:
try:
# Try to import actual I2S library (platform specific)
import pyaudio
self.audio = pyaudio.PyAudio()
self.stream = None
logger.info("I2S initialized with PyAudio")
except ImportError:
logger.warning("PyAudio not found, using simulator")
self.simulator = True
if self.simulator:
self._init_simulator()
def _init_simulator(self):
"""Initialize I2S simulator"""
logger.info("I2S Audio Simulator initialized")
self.sim_time = 0
self.waveform_generators = {
'sine': self._generate_sine,
'square': self._generate_square,
'sawtooth': self._generate_sawtooth,
'noise': self._generate_noise,
'silence': self._generate_silence
}
def _generate_sine(self, frequency: float, duration: float) -> np.ndarray:
"""Generate sine wave"""
samples = int(self.config.sample_rate * duration)
t = np.linspace(0, duration, samples)
if self.config.channels == 2:
# Stereo: slightly different frequencies for left/right
left = np.sin(2 * np.pi * frequency * t)
right = np.sin(2 * np.pi * (frequency * 1.01) * t)
signal = np.stack([left, right], axis=1)
else:
# Mono
signal = np.sin(2 * np.pi * frequency * t)
# Scale to bit depth
max_val = 2 ** (self.config.bits_per_sample - 1) - 1
return (signal * max_val).astype(np.int16)
def _generate_square(self, frequency: float, duration: float) -> np.ndarray:
"""Generate square wave"""
samples = int(self.config.sample_rate * duration)
t = np.linspace(0, duration, samples)
signal = np.sign(np.sin(2 * np.pi * frequency * t))
if self.config.channels == 2:
signal = np.stack([signal, signal], axis=1)
max_val = 2 ** (self.config.bits_per_sample - 1) - 1
return (signal * max_val).astype(np.int16)
def _generate_sawtooth(self, frequency: float, duration: float) -> np.ndarray:
"""Generate sawtooth wave"""
samples = int(self.config.sample_rate * duration)
t = np.linspace(0, duration, samples)
signal = 2 * (t * frequency % 1) - 1
if self.config.channels == 2:
signal = np.stack([signal, signal], axis=1)
max_val = 2 ** (self.config.bits_per_sample - 1) - 1
return (signal * max_val).astype(np.int16)
def _generate_noise(self, frequency: float, duration: float) -> np.ndarray:
"""Generate white noise"""
samples = int(self.config.sample_rate * duration)
if self.config.channels == 2:
signal = np.random.randn(samples, 2)
else:
signal = np.random.randn(samples)
max_val = 2 ** (self.config.bits_per_sample - 1) - 1
return (signal * max_val * 0.1).astype(np.int16) # 10% volume
def _generate_silence(self, frequency: float, duration: float) -> np.ndarray:
"""Generate silence"""
samples = int(self.config.sample_rate * duration)
if self.config.channels == 2:
return np.zeros((samples, 2), dtype=np.int16)
else:
return np.zeros(samples, dtype=np.int16)
async def start_recording(self, callback: Callable = None):
"""Start recording audio from I2S"""
self.is_recording = True
self.callback = callback
self.audio_buffer = []
logger.info(f"Started I2S recording: {self.config.sample_rate}Hz, "
f"{self.config.bits_per_sample}bit, "
f"{self.config.channels}ch")
if self.simulator:
# Simulate audio input
asyncio.create_task(self._simulate_recording())
else:
# Real I2S recording
self.stream = self.audio.open(
format=self.audio.get_format_from_width(
self.config.bits_per_sample // 8
),
channels=self.config.channels,
rate=self.config.sample_rate,
input=True,
frames_per_buffer=1024,
stream_callback=self._audio_callback
)
self.stream.start_stream()
async def _simulate_recording(self):
"""Simulate audio recording"""
while self.is_recording:
# Generate 100ms chunks
chunk_duration = 0.1
# Simulate different audio patterns
patterns = ['sine', 'square', 'noise', 'silence']
pattern = patterns[int(self.sim_time / 2) % len(patterns)]
# Generate audio data
frequency = 440 * (1 + 0.1 * np.sin(self.sim_time)) # Varying frequency
audio_data = self.waveform_generators[pattern](frequency, chunk_duration)
# Add to buffer
self.audio_buffer.append(audio_data)
# Call callback if provided
if self.callback:
await self.callback(audio_data)
self.sim_time += chunk_duration
await asyncio.sleep(chunk_duration)
def stop_recording(self) -> np.ndarray:
"""Stop recording and return audio data"""
self.is_recording = False
if not self.simulator and self.stream:
self.stream.stop_stream()
self.stream.close()
# Concatenate all buffers
if self.audio_buffer:
audio_data = np.concatenate(self.audio_buffer)
else:
audio_data = np.array([])
logger.info(f"Stopped recording. Captured {len(audio_data)} samples")
return audio_data
async def play_audio(self, audio_data: np.ndarray):
"""Play audio through I2S"""
self.is_playing = True
logger.info(f"Playing audio: {len(audio_data)} samples")
if self.simulator:
# Simulate playback delay
duration = len(audio_data) / self.config.sample_rate
await asyncio.sleep(duration)
logger.info("Simulated playback complete")
else:
# Real I2S playback
stream = self.audio.open(
format=self.audio.get_format_from_width(
self.config.bits_per_sample // 8
),
channels=self.config.channels,
rate=self.config.sample_rate,
output=True
)
# Play in chunks
chunk_size = 1024
for i in range(0, len(audio_data), chunk_size):
chunk = audio_data[i:i+chunk_size]
stream.write(chunk.tobytes())
stream.stop_stream()
stream.close()
self.is_playing = False
def generate_test_tone(self, frequency: float = 440, duration: float = 1.0) -> np.ndarray:
"""Generate test tone"""
return self._generate_sine(frequency, duration)
def analyze_audio(self, audio_data: np.ndarray) -> Dict:
"""Analyze audio data"""
if len(audio_data) == 0:
return {}
# Calculate basic statistics
if self.config.channels == 2 and len(audio_data.shape) > 1:
# Stereo
left = audio_data[:, 0]
right = audio_data[:, 1]
analysis = {
"samples": len(audio_data),
"duration": len(audio_data) / self.config.sample_rate,
"left_channel": {
"min": int(np.min(left)),
"max": int(np.max(right)),
"mean": float(np.mean(left)),
"rms": float(np.sqrt(np.mean(left**2)))
},
"right_channel": {
"min": int(np.min(right)),
"max": int(np.max(right)),
"mean": float(np.mean(right)),
"rms": float(np.sqrt(np.mean(right**2)))
}
}
else:
# Mono
analysis = {
"samples": len(audio_data),
"duration": len(audio_data) / self.config.sample_rate,
"min": int(np.min(audio_data)),
"max": int(np.max(audio_data)),
"mean": float(np.mean(audio_data)),
"rms": float(np.sqrt(np.mean(audio_data**2)))
}
# FFT for frequency analysis
fft = np.fft.fft(audio_data.flatten() if len(audio_data.shape) > 1 else audio_data)
frequencies = np.fft.fftfreq(len(fft), 1/self.config.sample_rate)
# Find dominant frequency
magnitude = np.abs(fft)
peak_idx = np.argmax(magnitude[:len(magnitude)//2])
dominant_freq = abs(frequencies[peak_idx])
analysis["dominant_frequency"] = float(dominant_freq)
return analysis
# Example usage
async def i2s_example():
"""Example I2S audio operations"""
# Initialize I2S
config = I2SConfig(
sample_rate=44100,
bits_per_sample=16,
channels=2
)
i2s = I2SHandler(config, simulator=True)
# Generate test tones
print("Generating test tones...")
tone_440 = i2s.generate_test_tone(440, 1.0) # A4
tone_880 = i2s.generate_test_tone(880, 1.0) # A5
# Play tones
print("Playing 440Hz tone...")
await i2s.play_audio(tone_440)
print("Playing 880Hz tone...")
await i2s.play_audio(tone_880)
# Record audio
print("Recording audio for 3 seconds...")
async def audio_callback(data):
analysis = i2s.analyze_audio(data)
print(f"Chunk analysis: {analysis.get('dominant_frequency', 0):.1f}Hz")
await i2s.start_recording(callback=audio_callback)
await asyncio.sleep(3)
recorded = i2s.stop_recording()
# Analyze recorded audio
analysis = i2s.analyze_audio(recorded)
print(f"Recording analysis: {analysis}")
"""
RS485/Modbus Protocol Handler with USB adapter support
Supports: Modbus RTU, Modbus ASCII, custom protocols
"""
import asyncio
import serial
import struct
import time
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import logging
logger = logging.getLogger('RS485')
class ModbusFunction(Enum):
"""Modbus function codes"""
READ_COILS = 0x01
READ_DISCRETE_INPUTS = 0x02
READ_HOLDING_REGISTERS = 0x03
READ_INPUT_REGISTERS = 0x04
WRITE_SINGLE_COIL = 0x05
WRITE_SINGLE_REGISTER = 0x06
WRITE_MULTIPLE_COILS = 0x0F
WRITE_MULTIPLE_REGISTERS = 0x10
@dataclass
class ModbusDevice:
"""Modbus device representation"""
slave_id: int
name: str
coils: Dict[int, bool] # Discrete outputs
discrete_inputs: Dict[int, bool] # Discrete inputs
holding_registers: Dict[int, int] # Read/Write registers
input_registers: Dict[int, int] # Read-only registers
class RS485Handler:
"""RS485/Modbus Protocol Handler"""
def __init__(self,
port: str = "/dev/ttyUSB0",
baudrate: int = 9600,
simulator: bool = False):
self.port = port
self.baudrate = baudrate
self.simulator = simulator
self.serial = None
self.devices = {}
# Modbus settings
self.timeout = 1.0
self.parity = serial.PARITY_EVEN
self.stopbits = serial.STOPBITS_ONE
self.bytesize = serial.EIGHTBITS
if not simulator:
try:
self.serial = serial.Serial(
port=port,
baudrate=baudrate,
parity=self.parity,
stopbits=self.stopbits,
bytesize=self.bytesize,
timeout=self.timeout
)
logger.info(f"RS485 initialized on {port} at {baudrate} baud")
except Exception as e:
logger.warning(f"Serial port error: {e}, using simulator")
self.simulator = True
if self.simulator:
self._init_simulator()
def _init_simulator(self):
"""Initialize RS485 simulator with virtual devices"""
logger.info("RS485/Modbus Simulator initialized")
# Simulate temperature controller
self.devices[1] = ModbusDevice(
slave_id=1,
name="Temperature Controller",
coils={i: False for i in range(16)}, # Outputs
discrete_inputs={i: False for i in range(16)}, # Inputs
holding_registers={
0: 250, # Setpoint (25.0ยฐC)
1: 243, # Current temp (24.3ยฐC)
2: 100, # P parameter
3: 50, # I parameter
4: 10, # D parameter
5: 1, # Control mode (1=auto, 0=manual)
},
input_registers={
0: 243, # Sensor 1 (24.3ยฐC)
1: 245, # Sensor 2 (24.5ยฐC)
2: 241, # Sensor 3 (24.1ยฐC)
3: 500, # Output power (50.0%)
}
)
# Simulate power meter
self.devices[2] = ModbusDevice(
slave_id=2,
name="Power Meter",
coils={},
discrete_inputs={},
holding_registers={
0: 2300, # Voltage (230.0V)
1: 150, # Current (15.0A)
2: 3450, # Power (3450W)
3: 950, # Power factor (0.95)
4: 500, # Frequency (50.0Hz)
},
input_registers={
0: 12345, # Energy counter high
1: 6789, # Energy counter low
2: 2300, # Voltage
3: 150, # Current
4: 3450, # Power
}
)
# Simulate VFD (Variable Frequency Drive)
self.devices[3] = ModbusDevice(
slave_id=3,
name="VFD",
coils={
0: False, # Start/Stop
1: False, # Forward/Reverse
2: False, # Fault reset
},
discrete_inputs={
0: True, # Ready
1: False, # Running
2: False, # Fault
3: False, # At speed
},
holding_registers={
0: 0, # Speed setpoint (0-10000 = 0-100%)
1: 500, # Acceleration time (5.00s)
2: 300, # Deceleration time (3.00s)
3: 5000, # Max frequency (50.00Hz)
4: 0, # Current speed
},
input_registers={
0: 0, # Output frequency
1: 0, # Output current
2: 0, # Output voltage
3: 0, # DC bus voltage
4: 250, # Motor temperature (25.0ยฐC)
}
)
def _calculate_crc(self, data: bytes) -> int:
"""Calculate Modbus CRC16"""
crc = 0xFFFF
for byte in data:
crc ^= byte
for _ in range(8):
if crc & 0x0001:
crc = (crc >> 1) ^ 0xA001
else:
crc >>= 1
return crc
def _build_request(self,
slave_id: int,
function: ModbusFunction,
address: int,
count_or_value: int) -> bytes:
"""Build Modbus RTU request"""
request = struct.pack('>BBH H',
slave_id,
function.value,
address,
count_or_value)
crc = self._calculate_crc(request)
request += struct.pack('<H', crc)
return request
def _parse_response(self, response: bytes) -> Optional[List[int]]:
"""Parse Modbus RTU response"""
if len(response) < 5:
return None
# Check CRC
crc_received = struct.unpack('<H', response[-2:])[0]
crc_calculated = self._calculate_crc(response[:-2])
if crc_received != crc_calculated:
logger.error("CRC error in response")
return None
# Parse based on function code
function = response[1]
if function in [0x01, 0x02]: # Read coils/discrete inputs
byte_count = response[2]
values = []
for i in range(3, 3 + byte_count):
for bit in range(8):
values.append(bool(response[i] & (1 << bit)))
return values
elif function in [0x03, 0x04]: # Read registers
byte_count = response[2]
values = []
for i in range(3, 3 + byte_count, 2):
value = struct.unpack('>H', response[i:i+2])[0]
values.append(value)
return values
elif function in [0x05, 0x06, 0x0F, 0x10]: # Write response
return [1] # Success
return None
async def read_holding_registers(self,
slave_id: int,
address: int,
count: int) -> Optional[List[int]]:
"""Read holding registers from device"""
if self.simulator:
if slave_id in self.devices:
device = self.devices[slave_id]
values = []
for i in range(address, address + count):
values.append(device.holding_registers.get(i, 0))
# Simulate some variation
import random
for i in range(len(values)):
if random.random() < 0.1: # 10% chance of change
values[i] = int(values[i] * (1 + random.gauss(0, 0.01)))
return values
return None
else:
request = self._build_request(
slave_id,
ModbusFunction.READ_HOLDING_REGISTERS,
address,
count
)
self.serial.write(request)
await asyncio.sleep(0.05) # Wait for response
response = self.serial.read(5 + count * 2)
return self._parse_response(response)
async def write_holding_register(self,
slave_id: int,
address: int,
value: int) -> bool:
"""Write single holding register"""
if self.simulator:
if slave_id in self.devices:
self.devices[slave_id].holding_registers[address] = value
logger.info(f"Device {slave_id} register {address} = {value}")
return True
return False
else:
request = self._build_request(
slave_id,
ModbusFunction.WRITE_SINGLE_REGISTER,
address,
value
)
self.serial.write(request)
await asyncio.sleep(0.05)
response = self.serial.read(8)
return self._parse_response(response) is not None
async def read_coils(self,
slave_id: int,
address: int,
count: int) -> Optional[List[bool]]:
"""Read coils (discrete outputs)"""
if self.simulator:
if slave_id in self.devices:
device = self.devices[slave_id]
values = []
for i in range(address, address + count):
values.append(device.coils.get(i, False))
return values
return None
else:
request = self._build_request(
slave_id,
ModbusFunction.READ_COILS,
address,
count
)
self.serial.write(request)
await asyncio.sleep(0.05)
byte_count = (count + 7) // 8
response = self.serial.read(5 + byte_count)
return self._parse_response(response)
async def write_coil(self,
slave_id: int,
address: int,
value: bool) -> bool:
"""Write single coil"""
if self.simulator:
if slave_id in self.devices:
self.devices[slave_id].coils[address] = value
logger.info(f"Device {slave_id} coil {address} = {value}")
# Simulate VFD control
if slave_id == 3 and address == 0: # VFD start/stop
self.devices[3].discrete_inputs[1] = value # Running status
if value:
# Ramp up to setpoint
setpoint = self.devices[3].holding_registers[0]
self.devices[3].holding_registers[4] = setpoint
self.devices[3].input_registers[0] = setpoint // 100
else:
self.devices[3].holding_registers[4] = 0
self.devices[3].input_registers[0] = 0
return True
return False
else:
value_word = 0xFF00 if value else 0x0000
request = self._build_request(
slave_id,
ModbusFunction.WRITE_SINGLE_COIL,
address,
value_word
)
self.serial.write(request)
await asyncio.sleep(0.05)
response = self.serial.read(8)
return self._parse_response(response) is not None
async def scan_devices(self, start_id: int = 1, end_id: int = 247) -> List[int]:
"""Scan for Modbus devices"""
found_devices = []
for slave_id in range(start_id, end_id + 1):
try:
# Try to read register 0
result = await self.read_holding_registers(slave_id, 0, 1)
if result is not None:
found_devices.append(slave_id)
device_name = self.devices.get(slave_id, {}).name if self.simulator else "Unknown"
logger.info(f"Found device at address {slave_id}: {device_name}")
except:
pass
await asyncio.sleep(0.01) # Small delay between scans
return found_devices
# Industrial automation example
async def rs485_example():
"""Example RS485/Modbus operations"""
# Initialize RS485
rs485 = RS485Handler(port="/dev/ttyUSB0", baudrate=9600, simulator=True)
# Scan for devices
print("Scanning for Modbus devices...")
devices = await rs485.scan_devices(1, 5)
print(f"Found {len(devices)} devices: {devices}")
# Temperature controller operations
print("\n=== Temperature Controller ===")
# Read current temperature and setpoint
registers = await rs485.read_holding_registers(1, 0, 2)
if registers:
setpoint = registers[0] / 10.0
current_temp = registers[1] / 10.0
print(f"Setpoint: {setpoint}ยฐC, Current: {current_temp}ยฐC")
# Change setpoint to 26.0ยฐC
await rs485.write_holding_register(1, 0, 260)
print("Changed setpoint to 26.0ยฐC")
# Power meter operations
print("\n=== Power Meter ===")
# Read electrical parameters
registers = await rs485.read_holding_registers(2, 0, 5)
if registers:
voltage = registers[0] / 10.0
current = registers[1] / 10.0
power = registers[2]
pf = registers[3] / 1000.0
freq = registers[4] / 10.0
print(f"Voltage: {voltage}V")
print(f"Current: {current}A")
print(f"Power: {power}W")
print(f"Power Factor: {pf}")
print(f"Frequency: {freq}Hz")
# VFD control
print("\n=== VFD Control ===")
# Set speed to 50%
await rs485.write_holding_register(3, 0, 5000)
print("Set VFD speed to 50%")
# Start VFD
await rs485.write_coil(3, 0, True)
print("Started VFD")
# Monitor VFD status
for i in range(5):
await asyncio.sleep(1)
# Read status
coils = await rs485.read_coils(3, 0, 3)
inputs = await rs485.read_discrete_inputs(3, 0, 4) if not rs485.simulator else [True, True, False, True]
registers = await rs485.read_holding_registers(3, 4, 1)
if registers:
speed = registers[0] / 100.0
running = inputs[1] if inputs else False
print(f"VFD Status: {'Running' if running else 'Stopped'} at {speed}%")
# Stop VFD
await rs485.write_coil(3, 0, False)
print("Stopped VFD")
version: '3.8'
services:
# Main EDPM service with all protocols
edpm-extended:
build:
context: .
dockerfile: Dockerfile.extended
container_name: edpm-extended
ports:
- "8080:8080" # Web UI
- "5555:5555" # ZeroMQ
- "1883:1883" # MQTT
volumes:
- ./app:/app
- /dev/shm:/dev/shm
devices:
# Pass through USB devices for RS485 adapters
- /dev/ttyUSB0:/dev/ttyUSB0
- /dev/ttyUSB1:/dev/ttyUSB1
environment:
- GPIO_MODE=SIMULATOR
- I2C_MODE=SIMULATOR
- I2S_MODE=SIMULATOR
- RS485_MODE=SIMULATOR
privileged: true # Required for device access
networks:
- edpm-network
depends_on:
- protocol-simulator
# Protocol simulator service
protocol-simulator:
build:
context: .
dockerfile: Dockerfile.simulator
container_name: protocol-simulator
ports:
- "8081:8081" # Simulator UI
- "5556:5556" # I2C sim port
- "5557:5557" # I2S sim port
- "5558:5558" # RS485 sim port
volumes:
- ./simulator:/simulator
- protocol-data:/var/protocol-data
environment:
- SIMULATION_MODE=ALL
- I2C_DEVICES=BME280,ADS1115,PCF8574
- I2S_CHANNELS=2
- RS485_DEVICES=3
networks:
- edpm-network
# Virtual serial port service for RS485 testing
socat-serial:
image: alpine/socat
container_name: socat-serial
command: |
-d -d
pty,raw,echo=0,link=/dev/ttyVUSB0
pty,raw,echo=0,link=/dev/ttyVUSB1
volumes:
- /dev:/dev
privileged: true
networks:
- edpm-network
# Protocol analyzer UI
protocol-analyzer:
build:
context: .
dockerfile: Dockerfile.analyzer
container_name: protocol-analyzer
ports:
- "8082:8082"
volumes:
- ./analyzer:/analyzer
- protocol-data:/var/protocol-data:ro
environment:
- ANALYZE_PROTOCOLS=GPIO,I2C,I2S,RS485
networks:
- edpm-network
networks:
edpm-network:
driver: bridge
volumes:
protocol-data:
FROM python:3.9-slim
# Install system dependencies for all protocols
RUN apt-get update && apt-get install -y \
# Basic tools
gcc python3-dev \
# ZeroMQ
libzmq3-dev \
# I2C tools
i2c-tools python3-smbus \
# Audio (I2S)
libasound2-dev portaudio19-dev \
# Serial (RS485)
picocom minicom \
# USB tools
usbutils \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Install Python packages
COPY requirements-extended.txt .
RUN pip install --no-cache-dir -r requirements-extended.txt
# Copy application files
COPY . .
# Create necessary directories
RUN mkdir -p /dev/shm /var/log/edpm /var/protocol-data
# Entry point
CMD ["python", "-u", "edpm_extended.py"]
# Core
pyzmq==25.1.1
aiohttp==3.8.5
numpy==1.24.3
redis==5.0.0
# I2C
smbus2==0.4.2
adafruit-circuitpython-bme280==2.6.20
adafruit-circuitpython-ads1x15==2.2.20
# I2S/Audio
pyaudio==0.2.13
sounddevice==0.4.6
scipy==1.11.3
# RS485/Serial
pyserial==3.5
pymodbus==3.5.2
minimalmodbus==2.1.1
# Testing/Simulation
pytest==7.4.2
pytest-asyncio==0.21.1
faker==19.6.2
"""
Complete example integrating GPIO, I2C, I2S, and RS485
"""
import asyncio
from edpm_lite import get_client
from i2c_handler import I2CHandler
from i2s_handler import I2SHandler, I2SConfig
from rs485_handler import RS485Handler
async def industrial_automation_example():
"""
Industrial automation scenario:
- Read temperature from I2C sensor
- Control VFD speed via RS485 based on temperature
- Generate audio alerts via I2S
- Log everything via EDPM
"""
# Initialize EDPM client
edpm = get_client()
# Initialize protocols
i2c = I2CHandler(simulator=True)
i2s = I2SHandler(I2SConfig(), simulator=True)
rs485 = RS485Handler(simulator=True)
# Main control loop
while True:
try:
# 1. Read environmental data from I2C sensor
env_data = i2c.read_bme280()
temperature = env_data['temperature']
humidity = env_data['humidity']
pressure = env_data['pressure']
edpm.log("info", f"Environment: {temperature}ยฐC, {humidity}%, {pressure}hPa")
# 2. Read ADC values for additional sensors
adc_values = []
for channel in range(4):
voltage = i2c.read_ads1115(channel)
adc_values.append(voltage)
edpm.log("debug", f"ADC CH{channel}: {voltage}V")
# 3. Control logic based on temperature
if temperature > 30:
# Too hot - increase cooling
edpm.log("warning", f"High temperature: {temperature}ยฐC")
# Increase VFD speed
await rs485.write_holding_register(3, 0, 8000) # 80% speed
await rs485.write_coil(3, 0, True) # Start VFD
# Audio alert
alert_tone = i2s.generate_test_tone(1000, 0.5) # 1kHz for 0.5s
await i2s.play_audio(alert_tone)
# Set GPIO warning LED
edpm.gpio_set(17, 1)
elif temperature < 20:
# Too cold - reduce cooling
edpm.log("info", f"Low temperature: {temperature}ยฐC")
# Reduce VFD speed
await rs485.write_holding_register(3, 0, 2000) # 20% speed
# Clear warning LED
edpm.gpio_set(17, 0)
else:
# Normal temperature
await rs485.write_holding_register(3, 0, 5000) # 50% speed
edpm.gpio_set(17, 0)
# 4. Read power consumption via RS485
power_data = await rs485.read_holding_registers(2, 0, 5)
if power_data:
voltage = power_data[0] / 10.0
current = power_data[1] / 10.0
power = power_data[2]
edpm.log("metrics", "Power consumption",
voltage=voltage, current=current, power=power)
# 5. Update GPIO expander based on status
status_byte = 0
if temperature > 25:
status_byte |= 0x01 # Bit 0: High temp
if humidity > 70:
status_byte |= 0x02 # Bit 1: High humidity
if power and power > 3000:
status_byte |= 0x04 # Bit 2: High power
i2c.set_gpio_expander(status_byte)
# 6. Audio feedback every 10 iterations
if asyncio.get_event_loop().time() % 10 < 1:
# Generate status beep
beep = i2s.generate_test_tone(500, 0.1)
await i2s.play_audio(beep)
# 7. Event emission
edpm.event("sensor_reading",
temperature=temperature,
humidity=humidity,
pressure=pressure,
adc_values=adc_values)
except Exception as e:
edpm.log("error", f"Control loop error: {e}")
await asyncio.sleep(5) # 5 second loop
# Run the example
if __name__ == "__main__":
asyncio.run(industrial_automation_example())
<!DOCTYPE html>
<html>
<head>
<title>EDPM Protocol Monitor</title>
<style>
body {
font-family: 'Segoe UI', sans-serif;
background: #1a1a1a;
color: #fff;
margin: 0;
padding: 20px;
}
.dashboard {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(400px, 1fr));
gap: 20px;
}
.panel {
background: #2a2a2a;
border-radius: 10px;
padding: 20px;
box-shadow: 0 4px 6px rgba(0,0,0,0.3);
}
.panel h2 {
margin: 0 0 20px 0;
color: #4CAF50;
border-bottom: 2px solid #4CAF50;
padding-bottom: 10px;
}
.metric {
display: flex;
justify-content: space-between;
padding: 10px;
background: #333;
margin: 10px 0;
border-radius: 5px;
}
.metric-label {
color: #888;
}
.metric-value {
font-weight: bold;
color: #4CAF50;
}
.status-led {
width: 20px;
height: 20px;
border-radius: 50%;
display: inline-block;
margin-left: 10px;
}
.status-led.on { background: #4CAF50; box-shadow: 0 0 10px #4CAF50; }
.status-led.off { background: #555; }
canvas {
width: 100%;
height: 200px;
background: #111;
border-radius: 5px;
margin: 10px 0;
}
.control-group {
display: flex;
gap: 10px;
margin: 10px 0;
}
button {
flex: 1;
padding: 10px;
background: #4CAF50;
color: white;
border: none;
border-radius: 5px;
cursor: pointer;
font-size: 14px;
transition: background 0.3s;
}
button:hover {
background: #45a049;
}
button:active {
transform: scale(0.98);
}
input[type="range"] {
width: 100%;
margin: 10px 0;
}
.log-window {
background: #111;
padding: 10px;
border-radius: 5px;
height: 150px;
overflow-y: auto;
font-family: monospace;
font-size: 12px;
}
.log-entry {
margin: 2px 0;
padding: 2px 5px;
}
.log-entry.info { color: #4CAF50; }
.log-entry.warning { color: #FFC107; }
.log-entry.error { color: #F44336; }
</style>
</head>
<body>
<h1 style="text-align: center; color: #4CAF50;">EDPM Extended Protocol Monitor</h1>
<div class="dashboard">
<!-- GPIO Panel -->
<div class="panel">
<h2>GPIO Status</h2>
<div class="metric">
<span class="metric-label">Pin 17:</span>
<span class="metric-value">HIGH <span class="status-led on"></span></span>
</div>
<div class="metric">
<span class="metric-label">Pin 22:</span>
<span class="metric-value">LOW <span class="status-led off"></span></span>
</div>
<div class="metric">
<span class="metric-label">Pin 27:</span>
<span class="metric-value">PWM 50% <span class="status-led on"></span></span>
</div>
<canvas id="gpioChart"></canvas>
</div>
<!-- I2C Panel -->
<div class="panel">
<h2>I2C Sensors</h2>
<div class="metric">
<span class="metric-label">BME280 Temp:</span>
<span class="metric-value" id="i2c-temp">24.3ยฐC</span>
</div>
<div class="metric">
<span class="metric-label">BME280 Humidity:</span>
<span class="metric-value" id="i2c-humidity">45%</span>
</div>
<div class="metric">
<span class="metric-label">BME280 Pressure:</span>
<span class="metric-value" id="i2c-pressure">1013 hPa</span>
</div>
<div class="metric">
<span class="metric-label">ADS1115 CH0:</span>
<span class="metric-value" id="adc-0">2.45V</span>
</div>
<canvas id="i2cChart"></canvas>
</div>
<!-- I2S Panel -->
<div class="panel">
<h2>I2S Audio</h2>
<div class="metric">
<span class="metric-label">Sample Rate:</span>
<span class="metric-value">44100 Hz</span>
</div>
<div class="metric">
<span class="metric-label">Channels:</span>
<span class="metric-value">Stereo</span>
</div>
<div class="metric">
<span class="metric-label">Level:</span>
<span class="metric-value" id="audio-level">-12 dB</span>
</div>
<canvas id="audioWaveform"></canvas>
<div class="control-group">
<button onclick="playTestTone(440)">440Hz</button>
<button onclick="playTestTone(880)">880Hz</button>
<button onclick="playTestTone(1760)">1760Hz</button>
</div>
</div>
<!-- RS485 Panel -->
<div class="panel">
<h2>RS485/Modbus Devices</h2>
<div class="metric">
<span class="metric-label">Device 1 (Temp Controller):</span>
<span class="metric-value">25.0ยฐC</span>
</div>
<div class="metric">
<span class="metric-label">Device 2 (Power Meter):</span>
<span class="metric-value">3.45 kW</span>
</div>
<div class="metric">
<span class="metric-label">Device 3 (VFD):</span>
<span class="metric-value">50% <span class="status-led on"></span></span>
</div>
<label>VFD Speed Control:</label>
<input type="range" id="vfd-speed" min="0" max="100" value="50"
oninput="updateVFDSpeed(this.value)">
<div class="control-group">
<button onclick="startVFD()">Start</button>
<button onclick="stopVFD()">Stop</button>
</div>
</div>
<!-- System Log -->
<div class="panel" style="grid-column: span 2;">
<h2>System Log</h2>
<div class="log-window" id="system-log">
<div class="log-entry info">[INFO] System initialized</div>
<div class="log-entry info">[INFO] All protocols connected</div>
</div>
</div>
</div>
<script>
// WebSocket connection
const ws = new WebSocket('ws://localhost:8080/ws');
// Initialize charts
const gpioCtx = document.getElementById('gpioChart').getContext('2d');
const i2cCtx = document.getElementById('i2cChart').getContext('2d');
const audioCtx = document.getElementById('audioWaveform').getContext('2d');
// Data buffers
const gpioData = [];
const i2cData = [];
const audioData = [];
ws.onmessage = (event) => {
const msg = JSON.parse(event.data);
// Update displays based on message type
if (msg.t === 'evt') {
if (msg.d.event === 'sensor_reading') {
updateI2CDisplay(msg.d);
} else if (msg.d.event === 'gpio_change') {
updateGPIODisplay(msg.d);
} else if (msg.d.event === 'audio_level') {
updateAudioDisplay(msg.d);
}
} else if (msg.t === 'log') {
addLogEntry(msg.d.level, msg.d.msg);
}
// Update charts
updateCharts();
};
function updateI2CDisplay(data) {
if (data.temperature !== undefined) {
document.getElementById('i2c-temp').textContent = data.temperature + 'ยฐC';
}
if (data.humidity !== undefined) {
document.getElementById('i2c-humidity').textContent = data.humidity + '%';
}
if (data.pressure !== undefined) {
document.getElementById('i2c-pressure').textContent = data.pressure + ' hPa';
}
if (data.adc_values) {
document.getElementById('adc-0').textContent = data.adc_values[0].toFixed(2) + 'V';
}
// Add to chart data
i2cData.push({
time: Date.now(),
temp: data.temperature,
humidity: data.humidity
});
if (i2cData.length > 100) i2cData.shift();
}
function updateGPIODisplay(data) {
// Update GPIO status LEDs
const led = document.querySelector(`#gpio-${data.pin} .status-led`);
if (led) {
led.classList.toggle('on', data.value === 1);
led.classList.toggle('off', data.value === 0);
}
}
function updateAudioDisplay(data) {
document.getElementById('audio-level').textContent = data.level.toFixed(1) + ' dB';
// Add waveform data
if (data.waveform) {
audioData.push(...data.waveform);
if (audioData.length > 1000) {
audioData.splice(0, audioData.length - 1000);
}
}
}
function addLogEntry(level, message) {
const logWindow = document.getElementById('system-log');
const entry = document.createElement('div');
entry.className = `log-entry ${level}`;
entry.textContent = `[${level.toUpperCase()}] ${message}`;
logWindow.insertBefore(entry, logWindow.firstChild);
// Keep only last 50 entries
while (logWindow.children.length > 50) {
logWindow.removeChild(logWindow.lastChild);
}
}
function updateCharts() {
// Update GPIO chart
drawGPIOChart();
// Update I2C chart
drawI2CChart();
// Update audio waveform
drawAudioWaveform();
}
function drawGPIOChart() {
// Implementation of GPIO signal chart
gpioCtx.clearRect(0, 0, gpioCtx.canvas.width, gpioCtx.canvas.height);
// ... drawing logic
}
function drawI2CChart() {
// Implementation of I2C data chart
i2cCtx.clearRect(0, 0, i2cCtx.canvas.width, i2cCtx.canvas.height);
// ... drawing logic
}
function drawAudioWaveform() {
// Implementation of audio waveform
audioCtx.clearRect(0, 0, audioCtx.canvas.width, audioCtx.canvas.height);
// ... drawing logic
}
// Control functions
function playTestTone(frequency) {
ws.send(JSON.stringify({
v: 1,
t: 'cmd',
d: {
action: 'play_tone',
frequency: frequency,
duration: 0.5
}
}));
addLogEntry('info', `Playing test tone: ${frequency}Hz`);
}
function updateVFDSpeed(value) {
ws.send(JSON.stringify({
v: 1,
t: 'cmd',
d: {
action: 'set_vfd_speed',
speed: parseInt(value)
}
}));
}
function startVFD() {
ws.send(JSON.stringify({
v: 1,
t: 'cmd',
d: {
action: 'start_vfd'
}
}));
addLogEntry('info', 'Starting VFD');
}
function stopVFD() {
ws.send(JSON.stringify({
v: 1,
t: 'cmd',
d: {
action: 'stop_vfd'
}
}));
addLogEntry('info', 'Stopping VFD');
}
// Auto-update every second
setInterval(() => {
updateCharts();
}, 1000);
</script>
</body>
</html>
# Makefile for EDPM Extended
.PHONY: test-i2c test-i2s test-rs485 test-all monitor
# Test I2C devices
test-i2c:
@echo "Testing I2C devices..."
@docker-compose exec edpm-extended python -c "
from i2c_handler import I2CHandler
import asyncio
async def test():
i2c = I2CHandler(simulator=True)
devices = i2c.scan()
print(f'Found devices: {[hex(d) for d in devices]}')
data = i2c.read_bme280()
print(f'BME280: {data}')
for ch in range(4):
v = i2c.read_ads1115(ch)
print(f'ADC CH{ch}: {v}V')
asyncio.run(test())
"
# Test I2S audio
test-i2s:
@echo "Testing I2S audio..."
@docker-compose exec edpm-extended python -c "
from i2s_handler import I2SHandler, I2SConfig
import asyncio
async def test():
config = I2SConfig()
i2s = I2SHandler(config, simulator=True)
# Generate and play test tone
tone = i2s.generate_test_tone(440, 1.0)
print('Playing 440Hz tone...')
await i2s.play_audio(tone)
# Record and analyze
print('Recording 2 seconds...')
await i2s.start_recording()
await asyncio.sleep(2)
audio = i2s.stop_recording()
analysis = i2s.analyze_audio(audio)
print(f'Analysis: {analysis}')
asyncio.run(test())
"
# Test RS485/Modbus
test-rs485:
@echo "Testing RS485/Modbus devices..."
@docker-compose exec edpm-extended python -c "
from rs485_handler import RS485Handler
import asyncio
async def test():
rs485 = RS485Handler(simulator=True)
# Scan devices
devices = await rs485.scan_devices(1, 5)
print(f'Found devices: {devices}')
# Read temperature controller
regs = await rs485.read_holding_registers(1, 0, 2)
if regs:
print(f'Temp: {regs[1]/10.0}ยฐC, Setpoint: {regs[0]/10.0}ยฐC')
# Control VFD
await rs485.write_holding_register(3, 0, 5000) # 50% speed
await rs485.write_coil(3, 0, True) # Start
print('VFD started at 50%')
await asyncio.sleep(2)
await rs485.write_coil(3, 0, False) # Stop
print('VFD stopped')
asyncio.run(test())
"
# Test all protocols together
test-all:
@echo "Testing all protocols..."
@$(MAKE) test-i2c
@$(MAKE) test-i2s
@$(MAKE) test-rs485
# Monitor all protocols
monitor:
@echo "Opening protocol monitor..."
@open http://localhost:8082 || xdg-open http://localhost:8082
# Generate test traffic
generate-traffic:
@echo "Generating test traffic on all protocols..."
@docker-compose exec edpm-extended python /app/generate_test_traffic.py
# View logs
logs-protocols:
@docker-compose logs -f edpm-extended protocol-simulator
# Performance test
perf-test:
@echo "Running performance tests..."
@docker-compose exec edpm-extended python -m pytest tests/perf_test.py -v
Stworzyลem kompletne rozszerzenie EDPM o obsลugฤ:
Rozwiฤ zanie jest production-ready i pozwala na peลne testowanie aplikacji embedded/przemysลowych bez fizycznego sprzฤtu!