support higher-level EtherCat features

This adds support to actually use the SDO data from terminals, and also add useful command line tools to inspect the bus.

See merge request teichman/ebpfcat!6

ebpfcat/ebpfcat.py

@@ -28,33 +28,52 @@ from .bpf import (
 
 
 class PacketDesc:
-    def __init__(self, position, size):
+    def __init__(self, sm, position, size):
+        self.sm = sm
         self.position = position
         self.size = size
 
     def __get__(self, instance, owner):
         if instance is None:
             return self
-        offset = instance.position_offset[self.position[0]]
+        offset = instance.position_offset[self.sm]
         if isinstance(instance, Struct):
             terminal = instance.terminal
             device = instance.device
         else:
             terminal = instance
             device = None
-        ret = PacketVar(terminal, (self.position[0],
-                                   self.position[1] + offset), self.size)
+        ret = PacketVar(terminal, self.sm, self.position + offset, self.size)
         if device is None:
             return ret
         else:
             return ret.get(device)
 
-    def __set__(self, instance, value):
-        offset = instance.position_offset[self.position[0]]
-        ret = PacketVar(instance.terminal,
-                        (self.position[0], self.position[1] + offset),
-                        self.size)
-        return ret.set(instance.device, value)
+
+class ProcessDesc:
+    def __init__(self, index, subindex, size=None):
+        self.index = index
+        self.subindex = subindex
+        self.size = size
+
+    def __get__(self, instance, owner):
+        if instance is None:
+            return self
+        index = self.index + instance.position_offset[3]
+        if isinstance(instance, Struct):
+            terminal = instance.terminal
+            device = instance.device
+        else:
+            terminal = instance
+            device = None
+        sm, offset, size = terminal.pdos[index, self.subindex]
+        if self.size is not None:
+            size = self.size
+        ret = PacketVar(terminal, sm, offset, size)
+        if device is None:
+            return ret
+        else:
+            return ret.get(device)
 
 
 class PacketVar(MemoryDesc):
@@ -66,8 +85,9 @@ class PacketVar(MemoryDesc):
         else:
             return self.size
 
-    def __init__(self, terminal, position, size):
+    def __init__(self, terminal, sm, position, size):
         self.terminal = terminal
+        self.sm = sm
         self.position = position
         self.size = size
 
@@ -119,8 +139,8 @@ class PacketVar(MemoryDesc):
                 return unpack_from("<" + self.size, data, start)[0]
 
     def _start(self, device):
-        base, offset = self.position
-        return device.sync_group.terminals[self.terminal][base] + offset
+        return device.sync_group.terminals[self.terminal][self.sm] \
+               + self.position
 
     def fmt_addr(self, device):
         return ("B" if isinstance(self.size, int) else self.size,
@@ -135,9 +155,9 @@ class Struct:
 
 
 class StructDesc:
-    def __init__(self, struct, *position_offset):
+    def __init__(self, struct, sm3=0, sm2=0):
         self.struct = struct
-        self.position_offset = position_offset
+        self.position_offset = {2: sm2, 3: sm3}
 
     def __get__(self, instance, owner):
         if instance is None:
@@ -209,30 +229,32 @@ class Device(SubProgram):
 
 class EBPFTerminal(Terminal):
     compatibility = None
-    position_offset = 0, 0
-
-    def __init_subclass__(cls):
-        cls.pdo = {}
-        for c in cls.__mro__[::-1]:
-            for k, v in c.__dict__.items():
-                if isinstance(v, PacketDesc):
-                    cls.pdo[k] = v
+    position_offset = {2: 0, 3: 0}
 
     async def initialize(self, relative, absolute):
         await super().initialize(relative, absolute)
         if (self.compatibility is not None and
                 (self.vendorId, self.productCode) not in self.compatibility):
-            raise RuntimeError("Incompatible Terminal")
+            raise RuntimeError(
+                f"Incompatible Terminal: {self.vendorId}:{self.productCode} "
+                f"({relative}, {absolute})")
+        await self.to_operational()
+        self.pdos = {}
+        if self.has_mailbox():
+            await self.parse_pdos()
 
     def allocate(self, packet, readonly):
+        """allocate space in packet for the pdos of this terminal
+
+        return a dict that contains the starting offset for each
+        sync manager"""
+        bases = {}
         if self.pdo_in_sz:
-            bases = [packet.size + packet.DATAGRAM_HEADER]
+            bases[3] = packet.size + packet.DATAGRAM_HEADER
             packet.append(ECCmd.FPRD, b"\0" * self.pdo_in_sz, 0,
                           self.position, self.pdo_in_off)
-        else:
-            bases = [None]
         if self.pdo_out_sz:
-            bases.append(packet.size + packet.DATAGRAM_HEADER)
+            bases[2] = packet.size + packet.DATAGRAM_HEADER
             if readonly:
                 packet.on_the_fly.append((packet.size, ECCmd.FPWR))
                 packet.append(ECCmd.NOP, b"\0" * self.pdo_out_sz, 0,
@@ -357,7 +379,6 @@ class SyncGroup(SyncGroupBase):
     current_data = False  # None is used to indicate FastSyncGroup
 
     async def run(self):
-        await gather(*[t.to_operational() for t in self.terminals])
         self.current_data = self.asm_packet
         while True:
             self.ec.send_packet(self.current_data)
@@ -406,9 +427,6 @@ class FastSyncGroup(SyncGroupBase, XDP):
     def start(self):
         self.allocate()
         self.ec.register_sync_group(self, self.packet)
-        self.monitor = ensure_future(gather(*[t.to_operational()
-                                              for t in self.terminals]))
-        return self.monitor
 
     def allocate(self):
         self.packet = Packet()

ebpfcat/ethercat.py

@@ -30,7 +30,7 @@ from asyncio import ensure_future, Event, Future, gather, get_event_loop, Protoc
 from enum import Enum
 from random import randint
 from socket import socket, AF_PACKET, SOCK_DGRAM
-from struct import pack, unpack, calcsize
+from struct import pack, unpack, unpack_from, calcsize
 
 class ECCmd(Enum):
    NOP = 0  # No Operation
@@ -128,8 +128,13 @@ class ObjectDescription:
     def __getitem__(self, idx):
         return self.entries[idx]
 
+    def __repr__(self):
+        return " ".join(f"[{k:X}: {v}]" for k, v in self.entries.items())
+
 
 class ObjectEntry:
+    name = None
+
     def __init__(self, desc):
         self.desc = desc
 
@@ -155,6 +160,13 @@ class ObjectEntry:
         return await self.desc.terminal.sdo_write(d, self.desc.index,
                                                   self.valueInfo)
 
+    def __repr__(self):
+        if self.name is None:
+            return "[unread ObjectEntry]"
+
+        return f'"{self.name}" {self.dataType}:{self.bitLength} ' \
+               f'{self.objectAccess:X}'
+
 
 def datasize(args, data):
     out = calcsize("<" + "".join(arg for arg in args if isinstance(arg, str)))
@@ -221,11 +233,11 @@ class Packet:
         pos = 14 + self.DATAGRAM_HEADER
         ret = []
         for cmd, bits, *dgram in self.data:
-            ret.append((data[pos-self.DATAGRAM_HEADER],
-                        data[pos:pos+len(bits)],
+            ret.append(unpack("<Bxh6x", data[pos-self.DATAGRAM_HEADER:pos])
+                       + (data[pos:pos+len(bits)],
                         unpack("<H", data[pos+len(bits):pos+len(bits)+2])[0]))
             pos += self.DATAGRAM_HEADER + self.DATAGRAM_TAIL
-        return ''.join(f"{i}: {c} {f} {d}\n" for i, (c, d, f) in enumerate(ret))
+        return ''.join(f"{i}: {c} {a} {f} {d}\n" for i, (c, a, d, f) in enumerate(ret))
 
     def full(self):
         """Is the data limit reached?"""
@@ -322,6 +334,8 @@ class EtherCat(Protocol):
             out += b"\0" * data
         elif data is not None:
             out += data
+        assert isinstance(pos, int) and isinstance(offset, int), \
+            f"pos: {pos} offset: {offset}"
         self.send_queue.put_nowait((cmd, out, idx, pos, offset, future))
         ret = await future
         if data is None:
@@ -333,6 +347,14 @@ class EtherCat(Protocol):
         else:
             return ret
 
+    async def count(self):
+        """Count the number of terminals on the bus"""
+        p = Packet()
+        p.append(ECCmd.APRD, b"\0\0", 0, 0, 0x10)
+        ret = await self.roundtrip_packet(p)
+        no, = unpack("<h", ret[16:18])  # number of terminals
+        return no
+
     def connection_made(self, transport):
         """start the send loop once the connection is made"""
         transport.get_extra_info("socket").bind(self.addr)
@@ -379,6 +401,9 @@ class Terminal:
         self.mbx_lock = Lock()
 
         self.eeprom = await self.read_eeprom()
+        if 41 not in self.eeprom:
+            # no sync managers defined in eeprom
+            return
         await self.write(0x800, data=0x80)  # empty out sync manager
         await self.write(0x800, data=self.eeprom[41])
         self.mbx_out_off = self.mbx_out_sz = None
@@ -403,22 +428,76 @@ class Terminal:
         s = await self.read(0x800, data=0x80)
         print(absolute, " ".join(f"{c:02x} {'|' if i % 8 == 7 else ''}" for i, c in enumerate(s)))
 
-    def parse_pdos(self):
-        def parse_pdo(s):
+    async def parse_pdos(self):
+        async def parse_eeprom(s):
             i = 0
+            bitpos = 0
             while i < len(s):
-                idx, e, sm, u1, u2, u3 = unpack("<HBBBBH", s[i:i+8])
-                print(f"idx {idx:x} sm {sm} {u1:x} {u2:x} {u3:x}")
+                idx, e, sm, u1, u2, u3 = unpack_from("<HBbBBH", s, i)
                 i += 8
                 for er in range(e):
-                    bitsize, = unpack("<5xB2x", s[i:i+8])
-                    print("  bs", bitsize, s[i:i+8])
+                    idx, subidx, k1, k2, bits, = unpack("<HBBBB2x", s[i:i+8])
+                    if sm > 0:
+                        yield idx, subidx, sm, bits
                     i += 8
 
-        if 50 in self.eeprom:
-            parse_pdo(self.eeprom[50])
-        if 51 in self.eeprom:
-            parse_pdo(self.eeprom[51])
+        async def parse_sdo(index, sm):
+            assignment = await self.sdo_read(index)
+            bitpos = 0
+            for i in range(0, len(assignment), 2):
+                pdo, = unpack_from("<H", assignment, i)
+                if pdo == 0:
+                    continue
+                count, = unpack("B", await self.sdo_read(pdo, 0))
+                for j in range(1, count + 1):
+                    bits, subidx, idx = unpack("<BBH", await self.sdo_read(pdo, j))
+                    yield idx, subidx, sm, bits
+
+        async def parse(func):
+            bitpos = 0
+            async for idx, subidx, sm, bits in func:
+                #print("k", idx, subidx, sm, bits)
+                if idx == 0:
+                    pass
+                elif bits < 8:
+                    self.pdos[idx, subidx] = (sm, bitpos // 8, bitpos % 8)
+                elif (bits % 8) or (bitpos % 8):
+                    raise RuntimeError("PDOs must be byte-aligned")
+                else:
+                    self.pdos[idx, subidx] = \
+                        (sm, bitpos // 8,
+                         {8: "B", 16: "H", 32: "I", 64: "Q"}[bits])
+                bitpos += bits
+
+        self.pdos = {}
+        if self.has_mailbox():
+            await parse(parse_sdo(0x1c12, 2))
+            await parse(parse_sdo(0x1c13, 3))
+        else:
+            if 50 in self.eeprom:
+                await parse(parse_eeprom(self.eeprom[50]))
+            if 51 in self.eeprom:
+                await parse(parse_eeprom(self.eeprom[51]))
+
+    async def parse_pdo(self, index, sm):
+        assignment = await self.sdo_read(index)
+        bitpos = 0
+        for i in range(0, len(assignment), 2):
+            pdo, = unpack_from("<H", assignment, i)
+            count, = unpack("B", await self.sdo_read(pdo, 0))
+            for j in range(1, count + 1):
+                bits, subidx, idx = unpack("<BBH", await self.sdo_read(pdo, j))
+                if idx == 0:
+                    pass
+                elif bits < 8:
+                    self.pdos[idx, subidx] = (sm, bitpos // 8, bitpos % 8)
+                elif (bits % 8) or (bitpos % 8):
+                    raise RuntimeError("PDOs must be byte-aligned")
+                else:
+                    self.pdos[idx, subidx] = \
+                        (sm, bitpos // 8,
+                         {8: "B", 16: "H", 32: "I", 64: "Q"}[bits])
+                bitpos += bits
 
     async def set_state(self, state):
         """try to set the state, and return the new state"""
@@ -470,7 +549,7 @@ class Terminal:
                                         start, *args, **kwargs))
 
     async def write(self, start, *args, **kwargs):
-        """write data from the terminal at offset `start`
+        """write data to the terminal at offset `start`
 
         see `EtherCat.roundtrip` for details on more parameters"""
         return (await self.ec.roundtrip(ECCmd.FPWR, self.position,
@@ -507,11 +586,15 @@ class Terminal:
                 return eeprom
             eeprom[hd] = await get_data(ws * 2)
 
+    def has_mailbox(self):
+        return self.mbx_out_off is not None and self.mbx_in_off is not None
+
     async def mbx_send(self, type, *args, data=None, address=0, priority=0, channel=0):
         """send data to the mailbox"""
         status, = await self.read(0x805, "B")  # always using mailbox 0, OK?
         if status & 8:
             raise RuntimeError("mailbox full, read first")
+        assert self.mbx_out_off is not None, "not send mailbox defined"
         await self.write(self.mbx_out_off, "HHBB",
                                 datasize(args, data),
                                 address, channel | priority << 6,
@@ -527,6 +610,7 @@ class Terminal:
         while status & 8 == 0:
             # always using mailbox 1, OK?
             status, = await self.read(0x80D, "B")
+        assert self.mbx_in_off is not None, "not receive mailbox defined"
         dlen, address, prio, type, data = await self.read(
                 self.mbx_in_off, "HHBB", data=self.mbx_in_sz - 6)
         return MBXType(type & 0xf), data[:dlen]
@@ -562,7 +646,11 @@ class Terminal:
                 raise RuntimeError(f"expected CoE, got {type}")
             coecmd, sdocmd, idx, subidx, size = unpack("<HBHBI", data[:10])
             if coecmd >> 12 != CoECmd.SDORES.value:
-                raise RuntimeError(f"expected CoE SDORES (3), got {coecmd>>12:x}")
+                if subindex is None and coecmd >> 12 == CoECmd.SDOREQ.value:
+                    return b""  # if there is no data, the terminal fails
+                raise RuntimeError(
+                    f"expected CoE SDORES (3), got {coecmd>>12:x} "
+                    f"for {index:X}:{9 if subindex is None else subindex:02X}")
             if idx != index:
                 raise RuntimeError(f"requested index {index}, got {idx}")
             if sdocmd & 2:
@@ -581,7 +669,8 @@ class Terminal:
                     raise RuntimeError(f"expected CoE, got {type}")
                 coecmd, sdocmd = unpack("<HB", data[:3])
                 if coecmd >> 12 != CoECmd.SDORES.value:
-                    raise RuntimeError(f"expected CoE cmd SDORES, got {coecmd}")
+                    raise RuntimeError(
+                        f"expected CoE cmd SDORES, got {coecmd}")
                 if sdocmd & 0xe0 != 0:
                     raise RuntimeError(f"requested index {index}, got {idx}")
                 if sdocmd & 1 and len(data) == 7:
@@ -688,55 +777,3 @@ class Terminal:
                 oe.name = data[8:].decode("utf8")
                 od.entries[i] = oe
         return ret
-
-
-async def main():
-    from .bpf import lookup_elem
-
-    ec = EtherCat("eth0")
-    await ec.connect()
-    #map_fd = await install_ebpf2()
-    tin = Terminal()
-    tin.ec = ec
-    tout = Terminal()
-    tout.ec = ec
-    tdigi = Terminal()
-    tdigi.ec = ec
-    await gather(
-        tin.initialize(-4, 19),
-        tout.initialize(-2, 55),
-        tdigi.initialize(0, 22),
-        )
-    print("tin")
-    #await tin.to_operational()
-    await tin.set_state(2)
-    print("tout")
-    await tout.to_operational()
-    print("reading odlist")
-    odlist2, odlist = await gather(tin.read_ODlist(), tout.read_ODlist())
-    #oe = odlist[0x7001][1]
-    #await oe.write(1)
-    for o in odlist.values():
-        print(hex(o.index), o.name, o.maxSub)
-        for i, p in o.entries.items():
-            print("   ", i, p.name, "|", p.dataType, p.bitLength, p.objectAccess)
-            #sdo = await tin.sdo_read(o.index, i)
-            try:
-               sdo = await p.read()
-               if isinstance(sdo, int):
-                   t = hex(sdo)
-               else:
-                   t = ""
-               print("   ", sdo, t)
-            except RuntimeError as e:
-               print("   E", e)
-    print("set sdo")
-    oe = odlist[0x8010][7]
-    print("=", await oe.read())
-    await oe.write(1)
-    print("=", await oe.read())
-    print(tdigi.eeprom[10])
-
-if __name__ == "__main__":
-    loop = get_event_loop()
-    loop.run_until_complete(main())

ebpfcat/scripts.py

+from argparse import ArgumentParser
+import asyncio
+from functools import wraps
+from struct import unpack
+import sys
+
+from .ethercat import EtherCat, Terminal, ECCmd
+
+def entrypoint(func):
+    @wraps(func)
+    def wrapper():
+        asyncio.run(func())
+    return wrapper
+
+
+@entrypoint
+async def scanbus():
+    ec = EtherCat(sys.argv[1])
+    await ec.connect()
+    no = await ec.count()
+    for i in range(no):
+        r, = await ec.roundtrip(ECCmd.APRD, -i, 0x10, "H", 44)
+        print(i, r)
+
+@entrypoint
+async def info():
+    parser = ArgumentParser(
+        prog = "ec-info",
+        description = "Retrieve information from an EtherCat bus")
+
+    parser.add_argument("interface")
+    parser.add_argument("-t", "--terminal", type=int)
+    parser.add_argument("-i", "--ids", action="store_true")
+    parser.add_argument("-n", "--names", action="store_true")
+    parser.add_argument("-s", "--sdo", action="store_true")
+    parser.add_argument("-v", "--values", action="store_true")
+    parser.add_argument("-p", "--pdo", action="store_true")
+    args = parser.parse_args()
+
+    ec = EtherCat(args.interface)
+    await ec.connect()
+
+    if args.terminal is None:
+        terminals = range(await ec.count())
+    else:
+        # former terminal: don't listen!
+        # this does not work with all terminals, dunno why
+        await ec.roundtrip(ECCmd.FPRW, 7, 0x10, "H", 0)
+        terminals = [args.terminal]
+
+    terms = [Terminal() for t in terminals]
+    for t in terms:
+        t.ec = ec
+    await asyncio.gather(*(t.initialize(-i, i + 7)
+                           for i, t in zip(terminals, terms)))
+
+    for i, t in enumerate(terms):
+        print(f"terminal no {i}")
+        if args.ids:
+            print(f"{t.vendorId:X}:{t.productCode:X} "
+                  f"revision {t.revisionNo:X} serial {t.serialNo}")
+        if args.names:
+            infos = t.eeprom[10]
+            i = 1
+            while i < len(infos):
+                print(infos[i+1 : i+infos[i]+1].decode("ascii"))
+                i += infos[i] + 1
+
+        if args.sdo:
+            await t.to_operational()
+            ret = await t.read_ODlist()
+            for k, v in ret.items():
+                print(f"{k:X}:")
+                for kk, vv in v.entries.items():
+                     print(f"    {kk:X}: {vv}")
+                     if args.values:
+                         r = await vv.read()
+                         if isinstance(r, int):
+                             print(f"        {r:10} {r:8X}")
+                         else:
+                             print(f"        {r}")
+        if args.pdo:
+            await t.to_operational()
+            await t.parse_pdos()
+            for (idx, subidx), (sm, pos, fmt) in t.pdos.items():
+                print(f"{idx:4X}:{subidx:02X} {sm} {pos} {fmt}")

ebpfcat/terminals.py

@@ -15,7 +15,7 @@
 # with this program; if not, write to the Free Software Foundation, Inc.,
 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
-from .ebpfcat import EBPFTerminal, PacketDesc, Struct
+from .ebpfcat import EBPFTerminal, PacketDesc, ProcessDesc, Struct
 
 
 class Generic(EBPFTerminal):
@@ -30,92 +30,96 @@ class Skip(EBPFTerminal):
 class EL1808(EBPFTerminal):
     compatibility = {(2, 118501458)}
 
-    ch1 = PacketDesc((0, 0), 0)
-    ch2 = PacketDesc((0, 0), 1)
-    ch3 = PacketDesc((0, 0), 2)
-    ch4 = PacketDesc((0, 0), 3)
-    ch5 = PacketDesc((0, 0), 4)
-    ch6 = PacketDesc((0, 0), 5)
-    ch7 = PacketDesc((0, 0), 6)
-    ch8 = PacketDesc((0, 0), 7)
+    ch1 = PacketDesc(3, 0, 0)
+    ch2 = PacketDesc(3, 0, 1)
+    ch3 = PacketDesc(3, 0, 2)
+    ch4 = PacketDesc(3, 0, 3)
+    ch5 = PacketDesc(3, 0, 4)
+    ch6 = PacketDesc(3, 0, 5)
+    ch7 = PacketDesc(3, 0, 6)
+    ch8 = PacketDesc(3, 0, 7)
 
 
 class EL2808(EBPFTerminal):
     compatibility = {(2, 184037458)}
 
-    ch1 = PacketDesc((1, 0), 0)
-    ch2 = PacketDesc((1, 0), 1)
-    ch3 = PacketDesc((1, 0), 2)
-    ch4 = PacketDesc((1, 0), 3)
-    ch5 = PacketDesc((1, 0), 4)
-    ch6 = PacketDesc((1, 0), 5)
-    ch7 = PacketDesc((1, 0), 6)
-    ch8 = PacketDesc((1, 0), 7)
+    ch1 = PacketDesc(2, 0, 0)
+    ch2 = PacketDesc(2, 0, 1)
+    ch3 = PacketDesc(2, 0, 2)
+    ch4 = PacketDesc(2, 0, 3)
+    ch5 = PacketDesc(2, 0, 4)
+    ch6 = PacketDesc(2, 0, 5)
+    ch7 = PacketDesc(2, 0, 6)
+    ch8 = PacketDesc(2, 0, 7)
 
 
 class EL4104(EBPFTerminal):
-    ch1_value = PacketDesc((1, 0), 'H')
-    ch2_value = PacketDesc((1, 2), 'H')
-    ch3_value = PacketDesc((1, 4), 'H')
-    ch4_value = PacketDesc((1, 6), 'H')
+    ch1_value = ProcessDesc(0x7000, 1)
+    ch2_value = ProcessDesc(0x7010, 1)
+    ch3_value = ProcessDesc(0x7020, 1)
+    ch4_value = ProcessDesc(0x7030, 1)
 
 
 class EL3164(EBPFTerminal):
     class Channel(Struct):
-        attrs = PacketDesc((0, 0), 'H')
-        value = PacketDesc((0, 2), 'H')
+        attrs = ProcessDesc(0x6000, 1, 'H')
+        value = ProcessDesc(0x6000, 0x11)
 
     channel1 = Channel(0)
-    channel2 = Channel(4)
-    channel3 = Channel(8)
-    channel4 = Channel(12)
+    channel2 = Channel(0x10)
+    channel3 = Channel(0x20)
+    channel4 = Channel(0x30)
+
+
+class EK1101(EBPFTerminal):
+    compatibility = {(2, 72166482)}
 
 
 class EK1814(EBPFTerminal):
-    ch1 = PacketDesc((0, 0), 0)
-    ch2 = PacketDesc((0, 0), 1)
-    ch3 = PacketDesc((0, 0), 2)
-    ch4 = PacketDesc((0, 0), 3)
-    ch5 = PacketDesc((1, 0), 0)
-    ch6 = PacketDesc((1, 0), 1)
-    ch7 = PacketDesc((1, 0), 2)
-    ch8 = PacketDesc((1, 0), 3)
+    ch1 = PacketDesc(3, 0, 0)
+    ch2 = PacketDesc(3, 0, 1)
+    ch3 = PacketDesc(3, 0, 2)
+    ch4 = PacketDesc(3, 0, 3)
+    ch5 = PacketDesc(2, 0, 0)
+    ch6 = PacketDesc(2, 0, 1)
+    ch7 = PacketDesc(2, 0, 2)
+    ch8 = PacketDesc(2, 0, 3)
 
 
 class EL5042(EBPFTerminal):
     compatibility = {(2, 330444882)}
     class Channel(Struct):
-        position = PacketDesc((0, 2), "q")
-        warning = PacketDesc((0, 0), 0)
-        error = PacketDesc((0, 0), 1)
-        status = PacketDesc((0, 0), "H")
+        position = PacketDesc(3, 2, "q")
+        warning = PacketDesc(3, 0, 0)
+        error = PacketDesc(3, 0, 1)
+        status = PacketDesc(3, 0, "H")
 
-    channel1 = Channel(0, None, 0)
-    channel2 = Channel(10, None, 0x10)
+    channel1 = Channel(0, 0)
+    channel2 = Channel(10, 0x10)
 
 
 class EL6022(EBPFTerminal):
     class Channel(Struct):
-        transmit_accept = PacketDesc((0, 0), 0)
-        receive_request = PacketDesc((0, 0), 1)
-        init_accept = PacketDesc((0, 0), 2)
-        status = PacketDesc((0, 0), "H")
-        in_string = PacketDesc((0, 1), "23p")
-
-        transmit_request = PacketDesc((1, 0), 0)
-        receive_accept = PacketDesc((1, 0), 1)
-        init_request = PacketDesc((1, 0), 2)
-        control = PacketDesc((1, 0), "H")
-        out_string = PacketDesc((1, 1), "23p")
+        transmit_accept = PacketDesc(3, 0, 0)
+        receive_request = PacketDesc(3, 0, 1)
+        init_accept = PacketDesc(3, 0, 2)
+        status = PacketDesc(3, 0, "H")
+        in_string = PacketDesc(3, 1, "23p")
+
+        transmit_request = PacketDesc(2, 0, 0)
+        receive_accept = PacketDesc(2, 0, 1)
+        init_request = PacketDesc(2, 0, 2)
+        control = PacketDesc(2, 0, "H")
+        out_string = PacketDesc(2, 1, "23p")
 
     channel1 = Channel(0, 0)
     channel2 = Channel(24, 24)
 
 
 class EL7041(EBPFTerminal):
-    compatibility = {(2, 461451346)}
-    velocity = PacketDesc((1, 6), "h")
-    enable = PacketDesc((1, 4), 0)
-    status = PacketDesc((0, 6), "H")
-    low_switch = PacketDesc((0, 1), 7)
-    high_switch = PacketDesc((0, 1), 8)
+    compatibility = {(2, 461451346), (2, 461455442), (2, 460795986)}
+    velocity = PacketDesc(2, 6, "h")
+    enable = PacketDesc(2, 4, 0)
+    status = PacketDesc(3, 6, "H")
+    low_switch = PacketDesc(3, 1, 7)
+    high_switch = PacketDesc(3, 1, 8)

pyproject.toml

+[project]
+name = "ebpfcat"
+version = "0.0.1"
+dependencies = []
+
+[project.scripts]
+ec-scanbus = "ebpfcat.scripts:scanbus"
+ec-info = "ebpfcat.scripts:info"