python-energymeter/energymeter/modbus_tcp_async.py

class AsyncModbusTCPMeter(ModbusTCPMeter):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.reader = self.writer = None

    async def _connect(self):
        self.reader, self.writer = await asyncio.open_connection(host=self.port, port=self.tcp_port)

    async def read(self, regnames=None):
        if regnames is None:
            registers = self.REGS
            return await self._read_multiple(registers)

        if isinstance(regnames, str):
            registers = [register for register in self.REGS if register['name'] == regnames]
            if len(registers) == 0:
                return "Register not found on device."
            return await self._read_single(registers[0])

        if isinstance(regnames, Iterable):
            registers = [register for register in self.REGS if register['name'] in regnames]
            return await self._read_multiple(registers)

    async def _read_single(self, register):
        num_regs = register['length']
        message = self._modbus_message(register['start'], num_regs)
        if self.writer is None:
            await self._connect()
        self.writer.write(message)
        await self.writer.drain()

        data = await self.reader.readexactly(9 + 2 * num_regs)
        return self._convert_value(data[9:], signed=register['signed'], decimals=register['decimals'], isFloat=register['isFloat']|False)

    async def _read_multiple(self, registers):
        registers.sort(key=lambda reg: reg['start'])
        results = {}
        for reg_range in self._split_ranges(registers):
            # Prepare the request
            first_reg = min([register['start'] for register in reg_range])
            num_regs = max([register['start'] + register['length'] for register in reg_range]) - first_reg

            if self.writer is None:
                await self._connect()
            self.writer.write(self._modbus_message(first_reg, num_regs))
            await self.writer.drain()

            # Receive the response

            data = await self.reader.readexactly(9 + 2 * num_regs)
            results.update(self._interpret_result(data[9:], reg_range))
        return results