diff --git a/pymodbus/repl/client/mclient.py b/pymodbus/repl/client/mclient.py index 765b1ad066..ac8a62f659 100644 --- a/pymodbus/repl/client/mclient.py +++ b/pymodbus/repl/client/mclient.py @@ -1,13 +1,16 @@ """Modbus Clients to be used with REPL.""" # pylint: disable=missing-type-doc import functools +from typing import TYPE_CHECKING from pymodbus.client import ModbusSerialClient as _ModbusSerialClient from pymodbus.client import ModbusTcpClient as _ModbusTcpClient +from pymodbus.client.base import ModbusBaseSyncClient as _ModbusBaseSyncClient from pymodbus.diag_message import ( ChangeAsciiInputDelimiterRequest, ClearCountersRequest, ClearOverrunCountRequest, + DiagnosticStatusResponse, ForceListenOnlyModeRequest, GetClearModbusPlusRequest, RestartCommunicationsOptionRequest, @@ -24,14 +27,22 @@ ReturnSlaveNoResponseCountRequest, ) from pymodbus.exceptions import ModbusIOException -from pymodbus.mei_message import ReadDeviceInformationRequest +from pymodbus.mei_message import ( + ReadDeviceInformationRequest, + ReadDeviceInformationResponse, +) from pymodbus.other_message import ( GetCommEventCounterRequest, + GetCommEventCounterResponse, GetCommEventLogRequest, + GetCommEventLogResponse, ReadExceptionStatusRequest, + ReadExceptionStatusResponse, ReportSlaveIdRequest, + ReportSlaveIdResponse, ) from pymodbus.pdu import ExceptionResponse, ModbusExceptions +from pymodbus.register_write_message import MaskWriteRegisterResponse def make_response_dict(resp): @@ -63,8 +74,11 @@ def _wrapper(*args, **kwargs): return _wrapper - -class ExtendedRequestSupport: # pylint: disable=(too-many-public-methods +if TYPE_CHECKING: + _Base = _ModbusBaseSyncClient +else: + _Base = object +class ExtendedRequestSupport(_Base): # pylint: disable=(too-many-public-methods """Extended request support.""" @staticmethod @@ -98,7 +112,7 @@ def read_coils(self, address, count=1, slave=0, **kwargs): :param kwargs: :returns: List of register values """ - resp = super().read_coils( # pylint: disable=no-member + resp = super().read_coils( address, count, slave, **kwargs ) if not resp.isError(): @@ -114,7 +128,7 @@ def read_discrete_inputs(self, address, count=1, slave=0, **kwargs): :param kwargs: :return: List of bits """ - resp = super().read_discrete_inputs( # pylint: disable=no-member + resp = super().read_discrete_inputs( address, count, slave, **kwargs ) if not resp.isError(): @@ -131,7 +145,7 @@ def write_coil(self, address, value, slave=0, **kwargs): :param kwargs: :return: """ - resp = super().write_coil( # pylint: disable=no-member + resp = super().write_coil( address, value, slave, **kwargs ) return resp @@ -146,7 +160,7 @@ def write_coils(self, address, values, slave=0, **kwargs): :param kwargs: :return: """ - resp = super().write_coils( # pylint: disable=no-member + resp = super().write_coils( address, values, slave, **kwargs ) return resp @@ -161,7 +175,7 @@ def write_register(self, address, value, slave=0, **kwargs): :param kwargs: :return: """ - resp = super().write_register( # pylint: disable=no-member + resp = super().write_register( address, value, slave, **kwargs ) return resp @@ -176,7 +190,7 @@ def write_registers(self, address, values, slave=0, **kwargs): :param kwargs: :return: """ - resp = super().write_registers( # pylint: disable=no-member + resp = super().write_registers( address, values, slave, **kwargs ) return resp @@ -190,7 +204,7 @@ def read_holding_registers(self, address, count=1, slave=0, **kwargs): :param kwargs: :return: """ - resp = super().read_holding_registers( # pylint: disable=no-member + resp = super().read_holding_registers( address, count, slave, **kwargs ) if not resp.isError(): @@ -206,7 +220,7 @@ def read_input_registers(self, address, count=1, slave=0, **kwargs): :param kwargs: :return: """ - resp = super().read_input_registers( # pylint: disable=no-member + resp = super().read_input_registers( address, count, slave, **kwargs ) if not resp.isError(): @@ -235,7 +249,7 @@ def readwrite_registers( :param kwargs: :return: """ - resp = super().readwrite_registers( # pylint: disable=no-member + resp = super().readwrite_registers( read_address=read_address, read_count=read_count, write_address=write_address, @@ -264,7 +278,7 @@ def mask_write_register( :param kwargs: :return: """ - resp = super().mask_write_register( # pylint: disable=no-member + resp: MaskWriteRegisterResponse = super().mask_write_register( address=address, and_mask=and_mask, or_mask=or_mask, slave=slave, **kwargs ) if not resp.isError(): @@ -285,7 +299,7 @@ def read_device_information(self, read_code=None, object_id=0x00, **kwargs): :return: """ request = ReadDeviceInformationRequest(read_code, object_id, **kwargs) - resp = self.execute(request) # pylint: disable=no-member + resp: ReadDeviceInformationResponse = self.execute(request) if not resp.isError(): return { "function_code": resp.function_code, @@ -306,7 +320,7 @@ def report_slave_id(self, slave=0, **kwargs): :return: """ request = ReportSlaveIdRequest(slave, **kwargs) - resp = self.execute(request) # pylint: disable=no-member + resp: ReportSlaveIdResponse = self.execute(request) if not resp.isError(): return { "function_code": resp.function_code, @@ -324,7 +338,7 @@ def read_exception_status(self, slave=0, **kwargs): :return: """ request = ReadExceptionStatusRequest(slave, **kwargs) - resp = self.execute(request) # pylint: disable=no-member + resp: ReadExceptionStatusResponse = self.execute(request) if not resp.isError(): return {"function_code": resp.function_code, "status": resp.status} return ExtendedRequestSupport._process_exception(resp, slave=request.slave_id) @@ -338,7 +352,7 @@ def get_com_event_counter(self, **kwargs): :return: """ request = GetCommEventCounterRequest(**kwargs) - resp = self.execute(request) # pylint: disable=no-member + resp: GetCommEventCounterResponse = self.execute(request) if not resp.isError(): return { "function_code": resp.function_code, @@ -357,7 +371,7 @@ def get_com_event_log(self, **kwargs): :return: """ request = GetCommEventLogRequest(**kwargs) - resp = self.execute(request) # pylint: disable=no-member + resp: GetCommEventLogResponse = self.execute(request) if not resp.isError(): return { "function_code": resp.function_code, @@ -370,7 +384,7 @@ def get_com_event_log(self, **kwargs): def _execute_diagnostic_request(self, request): """Execute diagnostic request.""" - resp = self.execute(request) # pylint: disable=no-member + resp: DiagnosticStatusResponse = self.execute(request) if not resp.isError(): return { "function code": resp.function_code,