Skip to content

Commit f486109

Browse files
author
Tom Softreck
committed
update
1 parent b749394 commit f486109

7 files changed

Lines changed: 1288 additions & 6 deletions

File tree

‎modapi/__init__.py‎

Lines changed: 20 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,9 @@
1919
)
2020
logger = logging.getLogger(__name__)
2121

22-
# Load environment variables from .env files
22+
2323
def load_env_files():
24-
"""Load environment variables from .env files in project directories"""
24+
"""Load environment variables from .env files in project directories."""
2525
# Try to load from current directory
2626
if load_dotenv(dotenv_path='.env'):
2727
logger.debug('Loaded .env from current directory')
@@ -36,12 +36,26 @@ def load_env_files():
3636
if hyper_env.exists() and load_dotenv(dotenv_path=hyper_env):
3737
logger.debug(f'Loaded .env from {hyper_env}')
3838

39+
3940
# Load environment variables
4041
load_env_files()
4142

4243
# Import components after environment is configured
43-
from .client import ModbusClient
44-
from .api import create_rest_app, start_mqtt_broker
45-
from .shell import main as shell_main
44+
from modapi.client import ModbusClient # noqa: E402
45+
from modapi.api import create_rest_app # noqa: E402
46+
from modapi.shell import main as shell_main # noqa: E402
47+
48+
49+
def start_mqtt_broker(*args, **kwargs):
50+
"""Stub for MQTT broker (not implemented in this version)."""
51+
logger.warning("MQTT broker is not implemented in this version")
52+
return None
53+
4654

47-
__all__ = ['ModbusClient', 'create_rest_app', 'start_mqtt_broker', 'shell_main', 'load_env_files']
55+
__all__ = [
56+
'ModbusClient',
57+
'create_rest_app',
58+
'start_mqtt_broker',
59+
'shell_main',
60+
'load_env_files'
61+
]

‎modapi/api/cmd.py‎

Lines changed: 259 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,259 @@
1+
"""
2+
modapi.api.cmd - Direct command execution for Modbus communication
3+
"""
4+
5+
import json
6+
import logging
7+
from typing import Dict, Any, List, Optional, Union, Tuple
8+
9+
from ..client import ModbusClient, auto_detect_modbus_port
10+
11+
# Configure logging
12+
logger = logging.getLogger(__name__)
13+
14+
def create_response(command: str) -> Dict[str, Any]:
15+
"""
16+
Create a base response dictionary
17+
18+
Args:
19+
command: Command string
20+
21+
Returns:
22+
Response dictionary with basic fields
23+
"""
24+
return {
25+
'command': command,
26+
'success': False,
27+
'timestamp': None,
28+
'operation': None,
29+
'error': None
30+
}
31+
32+
def output_json(data: Dict[str, Any]):
33+
"""
34+
Output data as formatted JSON
35+
36+
Args:
37+
data: Data to output
38+
"""
39+
print(json.dumps(data, indent=2))
40+
41+
def execute_command(command: str, args: List[str], port: Optional[str] = None,
42+
baudrate: Optional[int] = None, timeout: Optional[float] = None,
43+
verbose: bool = False) -> Tuple[bool, Dict[str, Any]]:
44+
"""
45+
Execute a Modbus command
46+
47+
Args:
48+
command: Command to execute (rc, wc, ri, rh, wh)
49+
args: Command arguments
50+
port: Modbus serial port (default: auto-detect)
51+
baudrate: Baud rate (default: from .env or 9600)
52+
timeout: Timeout in seconds (default: from .env or 1.0)
53+
verbose: Enable verbose logging
54+
55+
Returns:
56+
Tuple of (success, response_data)
57+
"""
58+
# Create response for JSON output
59+
response = create_response(f"{command} {' '.join(args)}")
60+
response['verbose'] = verbose
61+
62+
try:
63+
# Use the configured port or auto-detect
64+
if not port:
65+
port = auto_detect_modbus_port()
66+
if not port:
67+
response['error'] = "Could not auto-detect Modbus port"
68+
return False, response
69+
response['port_source'] = 'auto_detected'
70+
else:
71+
response['port_source'] = 'command_line'
72+
73+
response['port'] = port
74+
75+
# Initialize modbus client
76+
modbus = ModbusClient(
77+
port=port,
78+
baudrate=baudrate,
79+
timeout=timeout,
80+
verbose=verbose
81+
)
82+
83+
if not modbus.connect():
84+
response['error'] = f"Failed to connect to port {port}"
85+
return False, response
86+
87+
# Process commands
88+
cmd = command.lower()
89+
response['operation'] = cmd
90+
91+
try:
92+
if cmd == 'rc': # Read coils
93+
if len(args) < 2:
94+
response['error'] = "Usage: rc <address> <count> [unit]"
95+
return False, response
96+
97+
address = int(args[0])
98+
count = int(args[1])
99+
unit = int(args[2]) if len(args) > 2 else 1
100+
101+
response.update({
102+
'address': address,
103+
'count': count,
104+
'unit': unit,
105+
'register_type': 'coil'
106+
})
107+
108+
result = modbus.read_coils(address, count, unit)
109+
if result is not None:
110+
response.update({
111+
'success': True,
112+
'data': {
113+
'start_address': address,
114+
'end_address': address + count - 1,
115+
'values': result,
116+
'values_dict': {str(i): val for i, val in enumerate(result, address)}
117+
}
118+
})
119+
else:
120+
response['error'] = "Failed to read coils"
121+
122+
elif cmd == 'wc': # Write coil
123+
if len(args) < 2:
124+
response['error'] = "Usage: wc <address> <value> [unit]"
125+
return False, response
126+
127+
address = int(args[0])
128+
value = args[1].lower() in ('1', 'true', 'on')
129+
unit = int(args[2]) if len(args) > 2 else 1
130+
131+
response.update({
132+
'address': address,
133+
'value': value,
134+
'value_display': 'ON' if value else 'OFF',
135+
'unit': unit,
136+
'register_type': 'coil'
137+
})
138+
139+
if modbus.write_coil(address, value, unit):
140+
response.update({
141+
'success': True,
142+
'message': f"Coil {address} set to {'ON' if value else 'OFF'}",
143+
'data': {
144+
'address': address,
145+
'value': value,
146+
'value_display': 'ON' if value else 'OFF'
147+
}
148+
})
149+
else:
150+
response['error'] = f"Failed to write coil {address}"
151+
152+
elif cmd == 'ri': # Read discrete inputs
153+
if len(args) < 2:
154+
response['error'] = "Usage: ri <address> <count> [unit]"
155+
return False, response
156+
157+
address = int(args[0])
158+
count = int(args[1])
159+
unit = int(args[2]) if len(args) > 2 else 1
160+
161+
response.update({
162+
'address': address,
163+
'count': count,
164+
'unit': unit,
165+
'register_type': 'discrete_input'
166+
})
167+
168+
result = modbus.read_discrete_inputs(address, count, unit)
169+
if result is not None:
170+
response.update({
171+
'success': True,
172+
'data': {
173+
'address': address,
174+
'count': count,
175+
'values': [bool(v) for v in result],
176+
'values_display': ['ON' if v else 'OFF' for v in result]
177+
},
178+
'message': f"Read {count} discrete inputs starting at address {address}"
179+
})
180+
else:
181+
response['error'] = "Failed to read discrete inputs"
182+
183+
elif cmd == 'rh': # Read holding registers
184+
if len(args) < 2:
185+
response['error'] = "Usage: rh <address> <count> [unit]"
186+
return False, response
187+
188+
address = int(args[0])
189+
count = int(args[1])
190+
unit = int(args[2]) if len(args) > 2 else 1
191+
192+
response.update({
193+
'address': address,
194+
'count': count,
195+
'unit': unit,
196+
'register_type': 'holding_register'
197+
})
198+
199+
result = modbus.read_holding_registers(address, count, unit)
200+
if result is not None:
201+
response.update({
202+
'success': True,
203+
'data': {
204+
'address': address,
205+
'count': count,
206+
'values': result,
207+
'values_dict': {str(i): val for i, val in enumerate(result, address)},
208+
'hex_values': [f"0x{val:04X}" for val in result]
209+
},
210+
'message': f"Read {count} holding registers starting at address {address}"
211+
})
212+
else:
213+
response['error'] = "Failed to read holding registers"
214+
215+
elif cmd == 'wh': # Write holding register
216+
if len(args) < 2:
217+
response['error'] = "Usage: wh <address> <value> [unit]"
218+
return False, response
219+
220+
address = int(args[0])
221+
value = int(args[1])
222+
unit = int(args[2]) if len(args) > 2 else 1
223+
224+
response.update({
225+
'address': address,
226+
'value': value,
227+
'value_hex': f"0x{value:04X}",
228+
'unit': unit,
229+
'register_type': 'holding_register'
230+
})
231+
232+
if modbus.write_register(address, value, unit):
233+
response.update({
234+
'success': True,
235+
'message': f"Register {address} set to {value} (0x{value:04X})",
236+
'data': {
237+
'address': address,
238+
'value': value,
239+
'value_hex': f"0x{value:04X}"
240+
}
241+
})
242+
else:
243+
response['error'] = f"Failed to write register {address}"
244+
245+
else:
246+
response['error'] = f"Unknown command: {cmd}"
247+
return False, response
248+
249+
finally:
250+
if 'modbus' in locals():
251+
modbus.disconnect()
252+
253+
# Return success status
254+
return response.get('success', False), response
255+
256+
except Exception as e:
257+
logger.error(f"Error executing command: {e}")
258+
response['error'] = str(e)
259+
return False, response

0 commit comments

Comments
 (0)