Skip to content

Commit f3b9d70

Browse files
author
Tom Softreck
committed
update
1 parent aa1dd95 commit f3b9d70

2 files changed

Lines changed: 264 additions & 54 deletions

File tree

‎run_rtu_output.py‎

Lines changed: 145 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
"""
66

77
import os
8+
import sys
89
import logging
910
from flask import Flask, jsonify, request, render_template_string
1011
from modapi.api.rtu import ModbusRTU
@@ -121,48 +122,87 @@
121122
# Helper function to auto-detect RTU devices
122123
def auto_detect(ports):
123124
"""Auto-detect Modbus RTU device on specified ports"""
124-
if not ports:
125-
return None
126-
127-
# Try each port with common baudrates
128125
baudrates = [9600, 115200, 19200, 4800, 38400, 57600]
129-
unit_ids = [1, 2, 3, 0] # Include broadcast address 0
126+
unit_ids = [1, 2, 3, 4]
130127

131128
for port in ports:
132129
for baudrate in baudrates:
133130
for unit_id in unit_ids:
134131
try:
135-
client = ModbusRTU(port=port, baudrate=baudrate, timeout=0.5)
136-
if client.connect():
137-
# Try to read a register to verify connection
138-
response = client.read_holding_registers(0, 1, unit_id)
139-
if response is not None:
140-
logger.info(f"Found working configuration: {port}, {baudrate}, unit_id={unit_id}")
141-
client.disconnect()
142-
return {
143-
'port': port,
144-
'baudrate': baudrate,
145-
'unit_id': unit_id
146-
}
147-
148-
# Try reading coils if registers didn't work
149-
response = client.read_coils(0, 8, unit_id)
150-
if response is not None:
151-
logger.info(f"Found working configuration: {port}, {baudrate}, unit_id={unit_id}")
152-
client.disconnect()
153-
return {
154-
'port': port,
155-
'baudrate': baudrate,
156-
'unit_id': unit_id
157-
}
158-
132+
client = ModbusRTU(port, baudrate)
133+
logger.info(f"Testing configuration: port={port}, baudrate={baudrate}, unit_id={unit_id}")
134+
135+
# Próbuj odczytać rejestry
136+
logger.debug(f"Attempting to read holding registers with unit_id={unit_id}")
137+
response = client.read_holding_registers(0, 1, unit_id)
138+
if response is not None:
139+
logger.info(f"✅ Auto-detect success with holding registers: port={port}, baudrate={baudrate}, unit_id={unit_id}, response={response}")
140+
client.disconnect()
141+
return {
142+
'port': port,
143+
'baudrate': baudrate,
144+
'unit_id': unit_id
145+
}
146+
else:
147+
logger.debug(f"No response from holding registers with unit_id={unit_id}")
148+
149+
# Próbuj odczytać cewki
150+
logger.debug(f"Attempting to read coils with unit_id={unit_id}")
151+
response = client.read_coils(0, 8, unit_id)
152+
if response is not None:
153+
logger.info(f"✅ Auto-detect success with coils: port={port}, baudrate={baudrate}, unit_id={unit_id}, response={response}")
159154
client.disconnect()
155+
return {
156+
'port': port,
157+
'baudrate': baudrate,
158+
'unit_id': unit_id
159+
}
160+
else:
161+
logger.debug(f"No response from coils with unit_id={unit_id}")
162+
163+
client.disconnect()
160164
except Exception as e:
161-
logger.debug(f"Error testing {port} at {baudrate} with unit_id={unit_id}: {e}")
165+
# Log connection errors
166+
logger.debug(f"Error testing {port} at {baudrate} baud with unit_id={unit_id}: {e}")
162167

163168
logger.warning("No working configuration found")
164169
return None
165170

171+
def init_mock_mode():
172+
"""Initialize mock mode for testing without hardware"""
173+
global RTU_CONFIG
174+
print("🔧 Uruchamiam w trybie MOCK (bez rzeczywistego urządzenia)")
175+
RTU_CONFIG = {
176+
'port': 'MOCK',
177+
'baudrate': 9600,
178+
'unit_id': 1
179+
}
180+
logger.info(f"✅ Używam konfiguracji MOCK: {RTU_CONFIG}")
181+
182+
# Monkey patch ModbusRTU for mock mode
183+
def mock_read_coils(self, unit_id, address, count):
184+
logger.info(f"MOCK: Reading {count} coils from address {address} (unit_id={unit_id})")
185+
return [False] * count
186+
187+
def mock_write_single_coil(self, unit_id, address, value):
188+
logger.info(f"MOCK: Writing coil at address {address} to {value} (unit_id={unit_id})")
189+
return True
190+
191+
def mock_read_holding_registers(self, unit_id, address, count):
192+
logger.info(f"MOCK: Reading {count} registers from address {address} (unit_id={unit_id})")
193+
return [0] * count
194+
195+
ModbusRTU.read_coils = mock_read_coils
196+
ModbusRTU.write_single_coil = mock_write_single_coil
197+
ModbusRTU.read_holding_registers = mock_read_holding_registers
198+
199+
# Override connect and disconnect for mock mode
200+
ModbusRTU.connect = lambda self: True
201+
ModbusRTU.disconnect = lambda self: None
202+
203+
print("✅ Mock RTU device ready")
204+
return True
205+
166206
def init_rtu():
167207
"""Inicjalizuj RTU i znajdź działającą konfigurację"""
168208
global RTU_CONFIG
@@ -183,7 +223,33 @@ def init_rtu():
183223
manual_client = ModbusRTU('/dev/ttyACM0', 9600)
184224

185225
if manual_client.connect():
186-
success, result = manual_client.test_connection(1)
226+
# Implement test_connection directly
227+
result = {
228+
'port': '/dev/ttyACM0',
229+
'baudrate': 9600,
230+
'unit_id': 1,
231+
'success': False,
232+
'error': None
233+
}
234+
235+
try:
236+
# Try to read a register to verify connection
237+
response = manual_client.read_holding_registers(0, 1, 1)
238+
if response is not None:
239+
result['success'] = True
240+
else:
241+
result['error'] = "No response from device"
242+
243+
# Try reading coils if registers didn't work
244+
if not result['success']:
245+
response = manual_client.read_coils(0, 8, 1)
246+
if response is not None:
247+
result['success'] = True
248+
result['error'] = None
249+
except Exception as e:
250+
result['error'] = str(e)
251+
252+
success = result['success']
187253
if success:
188254
RTU_CONFIG = {
189255
'port': '/dev/ttyACM0',
@@ -212,7 +278,33 @@ def status():
212278

213279
# Test połączenia
214280
with ModbusRTU(RTU_CONFIG['port'], RTU_CONFIG['baudrate']) as client:
215-
success, result = client.test_connection(RTU_CONFIG['unit_id'])
281+
# Implement test_connection directly
282+
result = {
283+
'port': RTU_CONFIG['port'],
284+
'baudrate': RTU_CONFIG['baudrate'],
285+
'unit_id': RTU_CONFIG['unit_id'],
286+
'success': False,
287+
'error': None
288+
}
289+
290+
try:
291+
# Try to read a register to verify connection
292+
response = client.read_holding_registers(0, 1, RTU_CONFIG['unit_id'])
293+
if response is not None:
294+
result['success'] = True
295+
else:
296+
result['error'] = "No response from device"
297+
298+
# Try reading coils if registers didn't work
299+
if not result['success']:
300+
response = client.read_coils(0, 8, RTU_CONFIG['unit_id'])
301+
if response is not None:
302+
result['success'] = True
303+
result['error'] = None
304+
except Exception as e:
305+
result['error'] = str(e)
306+
307+
success = result['success']
216308

217309
return jsonify({
218310
'connected': success,
@@ -340,31 +432,30 @@ def get_register(address):
340432
print("🚀 RTU Output Server - Zastępuje problematyczny run_output.py")
341433
print("📡 Używa bezpośredniej komunikacji RTU zamiast PyModbus")
342434

343-
# Inicjalizuj RTU
344-
if init_rtu():
345-
print(f"✅ RTU skonfigurowane: {RTU_CONFIG['port']} @ {RTU_CONFIG['baudrate']} baud")
346-
print(f"🔧 Unit ID: {RTU_CONFIG['unit_id']}")
347-
print("🌐 Serwer dostępny na http://localhost:5005")
348-
print("📋 API endpoints:")
349-
print(" GET /status - status połączenia")
350-
print(" GET /coil/<addr> - odczyt cewki")
351-
print(" POST /coil/<addr> - zapis cewki")
352-
print(" GET /coils - odczyt wszystkich cewek")
353-
print(" GET /registers/<addr> - odczyt rejestru")
354-
print()
355-
356-
# Uruchom serwer Flask
357-
app.run(
358-
host='0.0.0.0',
359-
port=5005,
360-
debug=False, # Wyłącz debug w produkcji
361-
use_reloader=False # Zapobiega podwójnej inicjalizacji
362-
)
435+
# Import sys if not already imported
436+
import sys
437+
438+
# Check for mock mode
439+
mock_mode = "--mock" in sys.argv
440+
441+
if mock_mode:
442+
print("🧪 Wykryto flagę --mock, uruchamiam w trybie testowym bez sprzętu")
443+
init_success = init_mock_mode()
363444
else:
445+
# Inicjalizacja RTU
446+
print("🔌 Uruchamiam w trybie normalnym, szukam podłączonego sprzętu RTU")
447+
init_success = init_rtu()
448+
449+
if not init_success:
364450
print("❌ Nie można uruchomić serwera bez działającej konfiguracji RTU")
365451
print("🔍 Sprawdź:")
366452
print(" - Czy urządzenie jest podłączone do /dev/ttyACM0 lub /dev/ttyUSB0")
367-
print(" - Czy urządzenie jest włączone")
453+
print(" - Czy urządzenie jest włączone")
368454
print(" - Czy masz uprawnienia do portu szeregowego")
369455
print(" - Czy nie używa niestandardowej prędkości lub unit ID")
370-
exit(1)
456+
print("\n💡 Możesz uruchomić w trybie MOCK dla testów: python run_rtu_output.py --mock")
457+
sys.exit(1)
458+
459+
# Uruchom serwer
460+
print(f"✅ Uruchamiam serwer na http://localhost:5005/")
461+
app.run(host='0.0.0.0', port=5005, debug=False, use_reloader=False)

‎test_rtu_module.py‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
#!/usr/bin/env python3
2+
"""
3+
Test script for the refactored Modbus RTU module
4+
"""
5+
6+
import logging
7+
import sys
8+
from modapi.api.rtu import ModbusRTU, test_rtu_connection, create_rtu_client
9+
from modapi.api.rtu.utils import find_serial_ports, scan_for_devices
10+
from modapi.api.rtu.devices import WaveshareIO8CH, WaveshareAnalogInput8CH
11+
12+
# Configure logging
13+
logging.basicConfig(level=logging.INFO,
14+
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
15+
logger = logging.getLogger(__name__)
16+
17+
def test_imports():
18+
"""Test importing all components"""
19+
logger.info("✅ Successfully imported all components")
20+
return True
21+
22+
def test_find_ports():
23+
"""Test finding serial ports"""
24+
ports = find_serial_ports()
25+
logger.info(f"Found serial ports: {ports}")
26+
return len(ports) > 0
27+
28+
def test_connection():
29+
"""Test connection to RTU device"""
30+
port = '/dev/ttyACM0' # Default port
31+
success, result = test_rtu_connection(port)
32+
33+
if success:
34+
logger.info(f"✅ Connection successful: {result}")
35+
else:
36+
logger.warning(f"❌ Connection failed: {result}")
37+
38+
return success
39+
40+
def test_client_creation():
41+
"""Test creating RTU client"""
42+
try:
43+
client = create_rtu_client()
44+
logger.info("✅ Client created successfully")
45+
client.disconnect()
46+
return True
47+
except Exception as e:
48+
logger.error(f"❌ Failed to create client: {e}")
49+
return False
50+
51+
def test_auto_detect():
52+
"""Test auto-detection of RTU devices"""
53+
client = ModbusRTU()
54+
config = client.auto_detect()
55+
56+
if config:
57+
logger.info(f"✅ Auto-detection successful: {config}")
58+
return True
59+
else:
60+
logger.warning("❌ Auto-detection failed")
61+
return False
62+
63+
def test_device_classes():
64+
"""Test device-specific classes"""
65+
try:
66+
# Just test instantiation, don't connect to hardware
67+
io_device = WaveshareIO8CH(port=None)
68+
analog_device = WaveshareAnalogInput8CH(port=None)
69+
logger.info("✅ Device classes instantiated successfully")
70+
return True
71+
except Exception as e:
72+
logger.error(f"❌ Failed to instantiate device classes: {e}")
73+
return False
74+
75+
def main():
76+
"""Run all tests"""
77+
logger.info("Starting RTU module tests...")
78+
79+
tests = [
80+
("Import Test", test_imports),
81+
("Port Detection Test", test_find_ports),
82+
("Device Class Test", test_device_classes)
83+
]
84+
85+
# Only run hardware tests if --hardware flag is provided
86+
if "--hardware" in sys.argv:
87+
tests.extend([
88+
("Connection Test", test_connection),
89+
("Client Creation Test", test_client_creation),
90+
("Auto-detection Test", test_auto_detect)
91+
])
92+
93+
results = []
94+
for name, test_func in tests:
95+
logger.info(f"Running {name}...")
96+
try:
97+
success = test_func()
98+
results.append((name, success))
99+
except Exception as e:
100+
logger.error(f"Test {name} raised exception: {e}")
101+
results.append((name, False))
102+
103+
# Print summary
104+
logger.info("\n--- Test Results ---")
105+
all_passed = True
106+
for name, success in results:
107+
status = "✅ PASS" if success else "❌ FAIL"
108+
logger.info(f"{status} - {name}")
109+
all_passed = all_passed and success
110+
111+
if all_passed:
112+
logger.info("\n✅ All tests passed!")
113+
return 0
114+
else:
115+
logger.warning("\n❌ Some tests failed")
116+
return 1
117+
118+
if __name__ == "__main__":
119+
sys.exit(main())

0 commit comments

Comments
 (0)