diff --git a/app/core/modbus_client.py b/app/core/modbus_client.py index 2212943..04d4d6a 100644 --- a/app/core/modbus_client.py +++ b/app/core/modbus_client.py @@ -223,10 +223,11 @@ def _read_registers( for attempt in range(num_attempts): try: response = read_method( - address=address, count=count, slave=slave_id + address=address, count=count, device_id=slave_id ) - last_response = response + # Only assign last_response after validation succeeds if self._is_valid_response(response, op_name, slave_id): + last_response = response if attempt > 0: logger.info( "modbus_read_success_after_retry", @@ -314,7 +315,8 @@ def _read_registers( # This prevents response mix-up when multiple slave IDs share a connection self.close() - return last_response + # Return None - do not return potentially invalid last_response + return None finally: self._restore_timeout(orig_timeout) @@ -340,11 +342,15 @@ def write_holding_register(self, slave_id: int, address: int, value: int): for attempt in range(self.max_retries): try: response = self._client.write_register( - address=address, value=value, slave=slave_id + address=address, value=value, device_id=slave_id ) - last_response = response + # Only assign last_response after validation succeeds if self._is_valid_response(response, "write_holding_register", slave_id): + last_response = response return response + else: + # Invalid response - clear last_response + last_response = None except (ModbusException, OSError) as exc: logger.warning( "modbus_write_exception", @@ -370,7 +376,9 @@ def write_holding_register(self, slave_id: int, address: int, value: int): max_attempts=self.max_retries, ) time.sleep(self.retry_delay) - return last_response + + # Return None - do not return potentially invalid last_response + return None def is_connected(self) -> bool: return bool(self._client and self._client.is_socket_open()) @@ -399,8 +407,10 @@ def __init__(self, device_configs: Iterable[DeviceConfig]) -> None: cfg.device_id: cfg for cfg in device_configs } # Map (host, port) -> ModbusGateway + # IMPORTANT: One gateway per host:port, NOT per slave_id + # All devices sharing the same host:port share ONE connection self._gateways: Dict[Tuple[str, int], ModbusGateway] = {} - # Map (host, port) -> Lock + # Map (host, port) -> Lock - ensures serialized access per gateway self._locks: Dict[Tuple[str, int], asyncio.Lock] = {} self._manager_lock = asyncio.Lock() @@ -411,8 +421,16 @@ def __init__(self, device_configs: Iterable[DeviceConfig]) -> None: recovery_timeout=float(settings.CIRCUIT_BREAKER_RECOVERY_TIMEOUT), ) ) + + # Validate for problematic configurations + self._validate_gateway_config() def _create_gateway(self, config: DeviceConfig) -> ModbusGateway: + """Create a gateway for a device. + + Note: Gateway is shared across all devices on the same host:port, + regardless of slave_id. Access is serialized via locks. + """ return ModbusGateway( host=config.host, port=config.port, @@ -422,18 +440,77 @@ def _create_gateway(self, config: DeviceConfig) -> ModbusGateway: retry_delay=config.retry_delay, ) + def _validate_gateway_config(self) -> None: + """Check for problematic configurations across all devices. + + Validates: + - Multiple devices on same host:port with different slave_ids (allowed but warned) + - Duplicate (host, port, slave_id) combinations (error) + """ + # Track (host, port, slave_id) -> list of device_ids + endpoint_map: Dict[Tuple[str, int, int], List[str]] = {} + # Track (host, port) -> list of device_ids for warnings + gateway_map: Dict[Tuple[str, int], List[str]] = {} + + for device_id, config in self._configs.items(): + endpoint_key = (config.host, config.port, config.slave_id) + gateway_key = (config.host, config.port) + + if endpoint_key not in endpoint_map: + endpoint_map[endpoint_key] = [] + endpoint_map[endpoint_key].append(device_id) + + if gateway_key not in gateway_map: + gateway_map[gateway_key] = [] + gateway_map[gateway_key].append(device_id) + + # Check for duplicate endpoints (exact same host:port:slave_id) + duplicates = {key: devs for key, devs in endpoint_map.items() if len(devs) > 1} + if duplicates: + for (host, port, slave_id), device_ids in duplicates.items(): + logger.error( + "modbus_duplicate_endpoint_error", + host=host, + port=port, + slave_id=slave_id, + device_ids=device_ids, + message=( + f"CRITICAL: Multiple devices with identical endpoint: " + f"{host}:{port} slave_id={slave_id} -> devices: {device_ids}" + ), + ) + + # Warn about shared gateways (multiple devices on same host:port) + shared_gateways = {key: devs for key, devs in gateway_map.items() if len(devs) > 1} + if shared_gateways: + for (host, port), device_ids in shared_gateways.items(): + slave_ids = [self._configs[did].slave_id for did in device_ids] + logger.warning( + "modbus_shared_gateway", + host=host, + port=port, + device_ids=device_ids, + slave_ids=slave_ids, + message=( + f"Multiple devices share gateway {host}:{port} - " + f"polling will be serialized to prevent response mix-up. " + f"Devices: {device_ids} (slave_ids: {slave_ids})" + ), + ) + async def _get_gateway_and_lock(self, device_id: str) -> Tuple[ModbusGateway, asyncio.Lock]: config = self._configs.get(device_id) if not config: raise DeviceNotFoundError(f"Unknown device_id '{device_id}'") - key = (config.host, config.port) + # Key is (host, port) only - all devices on same gateway share connection + gateway_key = (config.host, config.port) async with self._manager_lock: - if key not in self._gateways: - self._gateways[key] = self._create_gateway(config) - self._locks[key] = asyncio.Lock() - return self._gateways[key], self._locks[key] + if gateway_key not in self._gateways: + self._gateways[gateway_key] = self._create_gateway(config) + self._locks[gateway_key] = asyncio.Lock() + return self._gateways[gateway_key], self._locks[gateway_key] async def _run_with_gateway(self, device_id: str, func_name: str, *args, **kwargs): """Execute a gateway method directly without circuit breaker. @@ -551,19 +628,22 @@ async def reset_gateway(self, device_id: str) -> None: This is useful when a connection becomes unresponsive. The next request will create a fresh connection. + + Note: Resets the entire gateway (host:port), not just one device. """ config = self._configs.get(device_id) if not config: raise DeviceNotFoundError(f"Unknown device_id '{device_id}'") - key = (config.host, config.port) + # Key is (host, port) - resets the shared gateway + gateway_key = (config.host, config.port) async with self._manager_lock: - if key in self._gateways: - gateway = self._gateways[key] + if gateway_key in self._gateways: + gateway = self._gateways[gateway_key] await asyncio.to_thread(gateway.close) - del self._gateways[key] - del self._locks[key] + del self._gateways[gateway_key] + del self._locks[gateway_key] logger.info( "modbus_gateway_reset", device_id=device_id, @@ -576,6 +656,7 @@ async def reload_configs(self, new_configs: Iterable[DeviceConfig]) -> None: """Reload device configurations dynamically. This closes connections for removed devices and updates the config map. + Also handles devices that moved to different host/port (stale gateway cleanup). """ old_device_ids = set(self._configs.keys()) new_device_ids = {cfg.device_id for cfg in new_configs} @@ -588,8 +669,29 @@ async def reload_configs(self, new_configs: Iterable[DeviceConfig]) -> None: except DeviceNotFoundError: pass # Already removed + # Check for devices that changed host/port (moved endpoints) + moved_devices: List[str] = [] + for new_cfg in new_configs: + if new_cfg.device_id in self._configs: + old_cfg = self._configs[new_cfg.device_id] + old_endpoint = (old_cfg.host, old_cfg.port) + new_endpoint = (new_cfg.host, new_cfg.port) + if old_endpoint != new_endpoint: + # Device moved to different endpoint - close old gateway + old_gateway_key = old_endpoint + if old_gateway_key in self._gateways: + try: + await self.reset_gateway(new_cfg.device_id) + moved_devices.append(new_cfg.device_id) + except DeviceNotFoundError: + pass + # Update configs self._configs = {cfg.device_id: cfg for cfg in new_configs} + + # Re-validate gateway config after reload + self._validate_gateway_config() + logger.info( "modbus_configs_reloaded", device_count=len(new_configs), @@ -613,9 +715,19 @@ def get_gateways_status(self) -> List[dict]: """Return status of all active gateways.""" status_list = [] for (host, port), gateway in self._gateways.items(): + # Get all devices using this gateway + device_ids = [ + dev_id for dev_id, cfg in self._configs.items() + if cfg.host == host and cfg.port == port + ] + slave_ids = [ + self._configs[dev_id].slave_id for dev_id in device_ids + ] status_list.append({ "host": host, "port": port, + "device_ids": device_ids, + "slave_ids": slave_ids, "connected": gateway.is_connected(), }) return status_list diff --git a/app/services/poller.py b/app/services/poller.py index aa97efa..89ceb0e 100644 --- a/app/services/poller.py +++ b/app/services/poller.py @@ -5,7 +5,7 @@ import asyncio import time from copy import deepcopy -from typing import List, Dict, Any, Set +from typing import Dict, List, Any, Set, Tuple from app.core.cache import RegisterCache @@ -257,7 +257,8 @@ async def poll_registers( This function implements: - Race condition prevention: Takes snapshot of targets at start of each cycle - - Parallel polling: Polls all targets concurrently using asyncio.gather + - Gateway-serialized polling: Targets on same gateway (host:port) are polled sequentially + - Parallel gateway polling: Different gateways can be polled in parallel - Hot-reload support: Reloads targets from database each cycle Args: @@ -279,7 +280,7 @@ async def poll_registers( "polling_service_started", database_mode=use_database, interval_seconds=interval_seconds, - parallel_polling=True, + gateway_serial_polling=True, message="Polling service started", ) @@ -317,41 +318,102 @@ async def poll_registers( ) cycle_start_time = time.time() - # PARALLEL POLLING: Poll all targets concurrently - # This significantly improves performance when polling multiple devices - polling_tasks = [ - _poll_single_target(target, manager, cache, mqtt_manager) - for target in targets + # GROUP TARGETS BY GATEWAY (host:port) + # This ensures targets on the same gateway are polled sequentially + # to prevent response mix-up from shared connections + gateway_groups: Dict[Tuple[str, int], List[dict]] = {} + + for target in targets: + device_id = target.get("device_id") + if not device_id: + continue + + # Get device config to find gateway key + config = manager.get_config(device_id) + if not config: + logger.warning( + "polling_target_unknown_device", + device_id=device_id, + message="Skipping target with unknown device", + ) + continue + + gateway_key = (config.host, config.port) + if gateway_key not in gateway_groups: + gateway_groups[gateway_key] = [] + gateway_groups[gateway_key].append(target) + + # LOG gateway grouping for debugging + for gw_key, gw_targets in gateway_groups.items(): + logger.debug( + "polling_gateway_group", + host=gw_key[0], + port=gw_key[1], + target_count=len(gw_targets), + device_ids=[t.get("device_id") for t in gw_targets], + ) + + # POLL EACH GATEWAY GROUP SEQUENTIALLY, BUT DIFFERENT GATEWAYS IN PARALLEL + # This prevents response mix-up while maintaining some parallelism + + async def poll_gateway_group( + gateway_key: Tuple[str, int], + group_targets: List[dict] + ) -> Tuple[int, int]: + """Poll all targets for one gateway sequentially. + + Returns (success_count, failure_count) for this gateway. + """ + success_count = 0 + failure_count = 0 + + for target in group_targets: + result = await _poll_single_target(target, manager, cache, mqtt_manager) + if isinstance(result, tuple): + success, _ = result + if success: + success_count += 1 + else: + failure_count += 1 + else: + failure_count += 1 + + return success_count, failure_count + + # Create tasks for each gateway group (parallel across gateways) + gateway_tasks = [ + poll_gateway_group(gw_key, gw_targets) + for gw_key, gw_targets in gateway_groups.items() ] - # Wait for all polling tasks to complete (with return_exceptions=True - # to prevent one failure from stopping others) - results = await asyncio.gather(*polling_tasks, return_exceptions=True) + # Wait for all gateway groups to complete + gateway_results = await asyncio.gather(*gateway_tasks, return_exceptions=True) # Process results success_count = 0 failure_count = 0 - for i, result in enumerate(results): + for i, result in enumerate(gateway_results): if isinstance(result, Exception): + gw_key = list(gateway_groups.keys())[i] logger.error( - "polling_task_exception", - task_index=i, - target=targets[i] if i < len(targets) else None, + "polling_gateway_exception", + gateway_host=gw_key[0], + gateway_port=gw_key[1], exception=str(result), exception_type=type(result).__name__, - message="Polling task raised exception", + message="Gateway group raised exception", exc_info=True, ) - failure_count += 1 + # Count all targets in this group as failures + failure_count += len(gateway_groups[gw_key]) elif isinstance(result, tuple): - success, error_msg = result - if success: - success_count += 1 - else: - failure_count += 1 + s, f = result + success_count += s + failure_count += f else: - failure_count += 1 + gw_key = list(gateway_groups.keys())[i] + failure_count += len(gateway_groups[gw_key]) cycle_duration = time.time() - cycle_start_time cycle_duration_ms = cycle_duration * 1000 @@ -367,6 +429,7 @@ async def poll_registers( success_count=success_count, failure_count=failure_count, total_targets=len(targets), + gateway_count=len(gateway_groups), duration_seconds=round(cycle_duration, 2), duration_ms=round(cycle_duration_ms, 2), message="Polling cycle completed", diff --git a/requirements.txt b/requirements.txt index d5918ce..9302b56 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,7 +10,7 @@ sqlmodel==0.0.27 asyncpg==0.31.0 # Modbus Communication -pymodbus==3.9.2 +pymodbus==3.13.0 # MQTT gmqtt==0.7.0 diff --git a/tests/unit/test_modbus_client.py b/tests/unit/test_modbus_client.py index 3760329..61b9924 100644 --- a/tests/unit/test_modbus_client.py +++ b/tests/unit/test_modbus_client.py @@ -53,6 +53,7 @@ async def test_read_registers_success(modbus_manager): mock_response = MagicMock() mock_response.isError.return_value = False mock_response.registers = [10, 20, 30] + mock_response.slave_id = 1 # Must match requested slave_id mock_instance.read_holding_registers.return_value = mock_response mock_instance.connect.return_value = True mock_instance.is_socket_open.return_value = True @@ -70,7 +71,7 @@ async def test_read_registers_success(modbus_manager): # Verify assert result == [10, 20, 30] mock_instance.read_holding_registers.assert_called_with( - address=0, count=3, slave=1 + address=0, count=3, device_id=1 ) @@ -87,6 +88,7 @@ async def test_read_registers_retry_success(modbus_manager): mock_response = MagicMock() mock_response.isError.return_value = False mock_response.registers = [99] + mock_response.slave_id = 1 # Must match requested slave_id mock_instance.read_holding_registers.side_effect = [ ModbusIOException("Connection lost"), @@ -151,6 +153,7 @@ async def test_write_register_success(modbus_manager): mock_response = MagicMock() mock_response.isError.return_value = False + mock_response.slave_id = 1 # Must match requested slave_id mock_instance.write_register.return_value = mock_response with patch_gateway_client(MockClient): @@ -163,7 +166,7 @@ async def test_write_register_success(modbus_manager): ) # Verify - mock_instance.write_register.assert_called_with(address=5, value=123, slave=1) + mock_instance.write_register.assert_called_with(address=5, value=123, device_id=1) @pytest.mark.asyncio