V2: Add XD/YD support and Serial support #94
Conversation
Expanded ClickPLC driver to support XD, YD, and CTD register read/write operations. Updated register type mapping and improved float decoding using BinaryPayloadDecoder. Added serial interface support to AsyncioModbusClient for both pymodbus 2.x and 3.x compatibility. Updated README to document new register types and access modes.
Enhanced parsing logic for XD and YD register addresses to support special cases like '0u', updated range checks, and improved data packing/unpacking for Modbus operations. Added the u_index() helper to normalize address indices and refactored related methods for clarity and correctness when reading and writing XD/YD registers.
Introduces a _convert_from_registers helper to AsyncioModbusClient for converting register data using the underlying client. Minor formatting change in driver.py for address validation.
Replaced manual struct packing/unpacking and BinaryPayloadDecoder usage with unified _convert_from_registers calls for register value decoding. Updated methods to handle single and multiple register reads more consistently, improved handling of special cases for XD and YD registers, and adjusted string decoding logic for TXT registers. Updated type hints and imports to reflect changes.
Added support for configuring baudrate, parity, stopbits, and bytesize in both ClickPLC and AsyncioModbusClient constructors. These parameters are now passed to the underlying serial client for improved serial connection flexibility.
andygarcha-adc
changed the title
|
As a general note, This means code you write on one version may fail on another, and you'll have to test to be sure. Hence all the ugly checks of Here I'd suggest avoiding the new If you want to test the CI, you can perhaps enable Github Actions in your fork. |
| from typing import Any, ClassVar, overload | ||
| from typing import Any, ClassVar, Literal, overload | ||
|
|
||
| from pymodbus.constants import Endian |
There was a problem hiding this comment.
avoid, if possible (see main comment)
| address, | ||
| tag_filepath='', | ||
| timeout=1, | ||
| interfacetype: Literal["TCP", "Serial"] = 'TCP', |
There was a problem hiding this comment.
Your approach would likely work fine (I haven't tested it).
However, this isn't the only driver I maintain and I prefer them to have similar patterns. In this case, it's pretty straightforward to determine if we are using serial or TCP and just select the correct client automatically:
See https://github.com/alexrudd2/alicat/blob/master/alicat/driver.py#L33-L48
| interfacetype: Literal["TCP", "Serial"] = 'TCP', | ||
| *, | ||
| baudrate=38400, | ||
| parity='O', |
There was a problem hiding this comment.
pyserial has nice constants available for this purpose rather than using "raw" integers:
https://pyserial.readthedocs.io/en/latest/pyserial_api.html#serial.Serial
| data_type = self.data_types[category].rstrip(digits) | ||
|
|
||
| # go through all the data |
There was a problem hiding this comment.
your other comments are generally helpful, but this is just noise. Remove obvious comments
| @@ -189,6 +239,7 @@ async def _get_x(self, start: int, end: int | None) -> dict: | |||
| coils = await self.read_coils(start_coil, count) | |||
| output = {} | |||
| current = start | |||
| # NOTE: could use client.convert_to_registers | |||
| if count == 1: | ||
| return values[0] | ||
| return {f'ds{start + i}': v for i, v in enumerate(values)} | ||
| register_values: int | list[int] = self.client.convert_from_registers( |
There was a problem hiding this comment.
what was wrong with the existing code?
| return values[0] | ||
| return {f'dd{start + i}': v for i, v in enumerate(values)} | ||
|
|
||
| registers_real = registers.copy() |
There was a problem hiding this comment.
what's going on here - does the existing code not work? (it's certainly possible I introduced a bug...)
|
|
||
| category = start[:i].lower() | ||
| start_index = 0.5 if start[i:].lower() == "0u" else int(start[i:]) | ||
| end_index = 0.5 if end is not None and end[i:].lower() == "0u" else None if end is None else int(end[i:]) |
There was a problem hiding this comment.
What is going on here??
Is there some attempt to read only half a register? Use a bitmask in that case.
| """Here's the deal with this method. | ||
|
|
||
| I had to denote for XD and YD if somebody was trying to get/set | ||
| XD0u or YD0u. So if that happens, then I pass through 0.5 as the |
There was a problem hiding this comment.
Hmm, yeah whoever came up with XD0u or YD0u ... fucked up. And you can tell them I said so.
Rather than making this whole convoluted mess of all the other XD/YD registers, let's instead special-case XD0u and YD0u higher up so that _set_xd etc. never has to deal with it
| @@ -715,6 +941,25 @@ async def _set_td(self, start: int, data: list[int]): | |||
|
|
|||
| await self.write_registers(address, values) | |||
|
|
|||
| async def _set_ctd(self, start: int, data: list[int]): | |||
There was a problem hiding this comment.
Interesting, I had thought it was a bad idea to set CTD registers but apparently it's OK.
This should have a new test for the roundtrip.
Hey Alex, Let me know if this was more along the lines of what you were thinking.
I'm using
git cherry-pickthis time instead of just one rebase. I've pulled out the stuff related to XD and YD, as well as the serial support (just because they were kind of intertwined). Hopefully this is a little bit easier to read - but if this isn't what you had in mind, please let me know. I have not had much experience working in group settings before - all of my previous git commits are on private, personal repositories so formatting and procedure aren't my forte.