Merge PR #67: fix(meter): speak RTU over TCP to the house meter's converter

The RS485 converter at 192.168.0.10 forwards raw RTU frames (unit, PDU,
CRC-16), as sdm_modbus does with ModbusRtuFramer. The gateway sent Modbus
TCP with an MBAP header, which the converter never answers, so every poll
timed out. TcpTransport now takes a framing: MBAP for MK01, RTU for the
meter.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

gateway/src/main/java/home/gateway/Gateway.java

@@ -165,7 +165,9 @@ public final class Gateway implements AutoCloseable {
             result.add(new HeatPumpDevice(heatPumpModbus, heatPump.settings()));
         }
         if (Switchable.isOn(devices.houseMeter())) {
-            result.add(new HouseMeterDevice(modbusOverTcp(devices.houseMeter().connection())));
+            // Its RS485 converter forwards RTU frames as they are, not Modbus TCP.
+            var meterLink = devices.houseMeter().connection();
+            result.add(new HouseMeterDevice(new PduModbusClient(own(TcpTransport.rtuOverTcp(meterLink.settings())))));
         }
         if (Switchable.isOn(devices.deye())) {
             result.add(new DeyeDevice(deyeModbus(devices.deye())));

gateway/src/main/java/home/gateway/adapters/modbus/Framing.java

@@ -0,0 +1,13 @@
+package home.gateway.adapters.modbus;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+
+/** How one request PDU and its response travel over the socket of a {@link TcpTransport}. */
+interface Framing {
+
+    byte[] frame(int unitId, byte[] pdu);
+
+    /** The response PDU, function code first, for the request framed last. */
+    byte[] readResponse(DataInputStream in, int unitId) throws IOException;
+}

gateway/src/main/java/home/gateway/adapters/modbus/MbapFraming.java

@@ -0,0 +1,46 @@
+package home.gateway.adapters.modbus;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/** Modbus TCP: an MBAP header with a transaction id before the PDU. */
+final class MbapFraming implements Framing {
+
+    private static final int HEADER_LENGTH = 7;
+
+    private int transactionId;
+
+    @Override
+    public byte[] frame(int unitId, byte[] pdu) {
+        transactionId = (transactionId + 1) & 0xFFFF;
+        return ByteBuffer.allocate(HEADER_LENGTH + pdu.length)
+                .putShort((short) transactionId)
+                .putShort((short) 0)
+                .putShort((short) (pdu.length + 1))
+                .put((byte) unitId)
+                .put(pdu)
+                .array();
+    }
+
+    @Override
+    public byte[] readResponse(DataInputStream in, int expectedUnitId) throws IOException {
+        ByteBuffer header = ByteBuffer.wrap(in.readNBytes(HEADER_LENGTH));
+        if (header.remaining() < HEADER_LENGTH) {
+            throw new IOException("connection closed by peer");
+        }
+        int id = header.getShort() & 0xFFFF;
+        int protocol = header.getShort() & 0xFFFF;
+        int length = header.getShort() & 0xFFFF;
+        int unitId = header.get() & 0xFF;
+        if (protocol != 0 || id != transactionId || unitId != expectedUnitId || length < 2) {
+            throw new IOException("unexpected MBAP header: transaction %d, protocol %d, length %d, unit %d"
+                    .formatted(id, protocol, length, unitId));
+        }
+        byte[] pdu = in.readNBytes(length - 1);
+        if (pdu.length < length - 1) {
+            throw new IOException("connection closed by peer");
+        }
+        return pdu;
+    }
+}

gateway/src/main/java/home/gateway/adapters/modbus/RtuFraming.java

@@ -0,0 +1,74 @@
+package home.gateway.adapters.modbus;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Modbus RTU frames (unit, PDU, CRC-16) sent as they are over TCP, as a plain RS485 converter
+ * forwards them. There is no length header, so the response length follows from its function code;
+ * only reads are framed, as the house meter is never written.
+ */
+final class RtuFraming implements Framing {
+
+    private static final int CRC_LENGTH = 2;
+
+    @Override
+    public byte[] frame(int unitId, byte[] pdu) {
+        ByteBuffer frame = ByteBuffer.allocate(1 + pdu.length + CRC_LENGTH).put((byte) unitId).put(pdu);
+        int crc = crc(frame.array(), 1 + pdu.length);
+        return frame.put((byte) crc).put((byte) (crc >>> 8)).array();
+    }
+
+    @Override
+    public byte[] readResponse(DataInputStream in, int expectedUnitId) throws IOException {
+        byte[] head = readExactly(in, 2);
+        int unitId = head[0] & 0xFF;
+        int function = head[1] & 0xFF;
+        if (unitId != expectedUnitId) {
+            throw new IOException("RTU response from unit " + unitId + ", expected " + expectedUnitId);
+        }
+        byte[] body = switch (function) {
+            case 3, 4 -> withCount(in);
+            default -> {
+                if ((function & 0x80) == 0) {
+                    throw new IOException("RTU response with unsupported function " + function);
+                }
+                yield readExactly(in, 1);
+            }
+        };
+        byte[] crcBytes = readExactly(in, CRC_LENGTH);
+        byte[] frame = ByteBuffer.allocate(2 + body.length).put(head).put(body).array();
+        int expected = crc(frame, frame.length);
+        int received = (crcBytes[0] & 0xFF) | (crcBytes[1] & 0xFF) << 8;
+        if (received != expected) {
+            throw new IOException("RTU response CRC %04x, expected %04x".formatted(received, expected));
+        }
+        return ByteBuffer.allocate(1 + body.length).put((byte) function).put(body).array();
+    }
+
+    private static byte[] withCount(DataInputStream in) throws IOException {
+        int count = readExactly(in, 1)[0] & 0xFF;
+        return ByteBuffer.allocate(1 + count).put((byte) count).put(readExactly(in, count)).array();
+    }
+
+    private static byte[] readExactly(DataInputStream in, int length) throws IOException {
+        byte[] bytes = in.readNBytes(length);
+        if (bytes.length < length) {
+            throw new IOException("connection closed by peer");
+        }
+        return bytes;
+    }
+
+    /** Modbus CRC-16: polynomial 0xA001, start 0xFFFF; sent low byte first. */
+    static int crc(byte[] bytes, int length) {
+        int crc = 0xFFFF;
+        for (int i = 0; i < length; i++) {
+            crc ^= bytes[i] & 0xFF;
+            for (int bit = 0; bit < 8; bit++) {
+                crc = (crc & 1) != 0 ? (crc >>> 1) ^ 0xA001 : crc >>> 1;
+            }
+        }
+        return crc;
+    }
+}

gateway/src/main/java/home/gateway/adapters/modbus/TcpTransport.java

@@ -5,7 +5,6 @@ import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.net.InetSocketAddress;
 import java.net.Socket;
-import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -13,10 +12,11 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * Modbus TCP (MBAP). The MK01 gateway serves a single client and a second connection drops the
- * first, so requests never overlap and the socket is reused for the requests of one poll. It is
- * closed after a few idle seconds: MK01 resets a connection left idle for minutes, which cost every
- * other poll, and while one is open the vendor's app cannot connect.
+ * Modbus over a TCP socket: Modbus TCP (MBAP) or RTU frames as an RS485 converter forwards them.
+ * The MK01 gateway serves a single client and a second connection drops the first, so requests
+ * never overlap and the socket is reused for the requests of one poll. It is closed after a few
+ * idle seconds: MK01 resets a connection left idle for minutes, which cost every other poll, and
+ * while one is open the vendor's app cannot connect.
  */
 public final class TcpTransport implements ModbusTransport {
 
@@ -24,26 +24,32 @@ public final class TcpTransport implements ModbusTransport {
                            Duration minimumPause) {
     }
 
-    private static final int HEADER_LENGTH = 7;
     static final Duration IDLE_CLOSE = Duration.ofSeconds(5);
 
     // Not synchronized: a virtual thread blocked in a monitor pins its carrier thread.
     private final ReentrantLock lock = new ReentrantLock();
     private final Settings settings;
+    private final Framing framing;
     private final Duration idleClose;
     private final ScheduledExecutorService idleCloser = Executors.newSingleThreadScheduledExecutor(
             Thread.ofVirtual().name("modbus-idle-close").factory());
     private Socket socket;
     private DataInputStream in;
-    private int transactionId;
     private long lastResponseNanos;
 
+    /** Modbus TCP, as MK01 speaks it. */
     public TcpTransport(Settings settings) {
-        this(settings, IDLE_CLOSE);
+        this(settings, new MbapFraming(), IDLE_CLOSE);
     }
 
-    TcpTransport(Settings settings, Duration idleClose) {
+    /** RTU frames with their CRC and no MBAP header, as a plain RS485 converter forwards them. */
+    public static TcpTransport rtuOverTcp(Settings settings) {
+        return new TcpTransport(settings, new RtuFraming(), IDLE_CLOSE);
+    }
+
+    TcpTransport(Settings settings, Framing framing, Duration idleClose) {
         this.settings = settings;
+        this.framing = framing;
         this.idleClose = idleClose;
         this.lastResponseNanos = System.nanoTime() - settings.minimumPause().toNanos();
     }
@@ -62,9 +68,8 @@ public final class TcpTransport implements ModbusTransport {
         try {
             awaitPause();
             connectIfNeeded();
-            int id = nextTransactionId();
-            socket.getOutputStream().write(frame(id, unitId, requestPdu));
-            byte[] response = readResponse(id, unitId);
+            socket.getOutputStream().write(framing.frame(unitId, requestPdu));
+            byte[] response = framing.readResponse(in, unitId);
             lastResponseNanos = System.nanoTime();
             if (!idleCloser.isShutdown()) {
                 idleCloser.schedule(this::closeIfIdle, idleClose.toNanos(), TimeUnit.NANOSECONDS);
@@ -130,41 +135,6 @@ public final class TcpTransport implements ModbusTransport {
         in = new DataInputStream(candidate.getInputStream());
     }
 
-    private int nextTransactionId() {
-        transactionId = (transactionId + 1) & 0xFFFF;
-        return transactionId;
-    }
-
-    private static byte[] frame(int transactionId, int unitId, byte[] pdu) {
-        return ByteBuffer.allocate(HEADER_LENGTH + pdu.length)
-                .putShort((short) transactionId)
-                .putShort((short) 0)
-                .putShort((short) (pdu.length + 1))
-                .put((byte) unitId)
-                .put(pdu)
-                .array();
-    }
-
-    private byte[] readResponse(int expectedTransactionId, int expectedUnitId) throws IOException {
-        ByteBuffer header = ByteBuffer.wrap(in.readNBytes(HEADER_LENGTH));
-        if (header.remaining() < HEADER_LENGTH) {
-            throw new IOException("connection closed by peer");
-        }
-        int id = header.getShort() & 0xFFFF;
-        int protocol = header.getShort() & 0xFFFF;
-        int length = header.getShort() & 0xFFFF;
-        int unitId = header.get() & 0xFF;
-        if (protocol != 0 || id != expectedTransactionId || unitId != expectedUnitId || length < 2) {
-            throw new IOException("unexpected MBAP header: transaction %d, protocol %d, length %d, unit %d"
-                    .formatted(id, protocol, length, unitId));
-        }
-        byte[] pdu = in.readNBytes(length - 1);
-        if (pdu.length < length - 1) {
-            throw new IOException("connection closed by peer");
-        }
-        return pdu;
-    }
-
     private void closeSocket() {
         if (socket == null) {
             return;

gateway/src/test/java/home/gateway/adapters/modbus/MockModbusServer.java

@@ -8,16 +8,24 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
-/** Local Modbus TCP fake answering FC3, FC4 and FC6; unknown addresses get exception code 2. */
+/**
+ * Local Modbus fake answering FC3, FC4 and FC6; unknown addresses get exception code 2. It speaks
+ * Modbus TCP, or RTU frames over TCP when made with {@link #rtuOverTcp()}.
+ */
 public final class MockModbusServer implements Closeable {
 
+    private record Request(int transactionId, int unitId, byte[] pdu) {
+    }
+
     private final ServerSocket server;
+    private final boolean rtu;
     private final Map<Integer, Integer> holding = new ConcurrentHashMap<>();
     private final Map<Integer, Integer> input = new ConcurrentHashMap<>();
     private final AtomicInteger connections = new AtomicInteger();
@@ -26,15 +34,26 @@ public final class MockModbusServer implements Closeable {
     private volatile Duration responseDelay = Duration.ZERO;
     private volatile boolean dropNextRequest;
     private volatile boolean wrongTransactionId;
+    private volatile boolean wrongCrc;
+    private volatile boolean wrongUnit;
     private volatile boolean overlapped;
     private volatile boolean wrongWriteEcho;
     private volatile int writeException;
 
     public MockModbusServer() throws IOException {
+        this(false);
+    }
+
+    private MockModbusServer(boolean rtu) throws IOException {
+        this.rtu = rtu;
         server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
         Thread.ofVirtual().start(this::acceptLoop);
     }
 
+    static MockModbusServer rtuOverTcp() throws IOException {
+        return new MockModbusServer(true);
+    }
+
     public int port() {
         return server.getLocalPort();
     }
@@ -90,6 +109,16 @@ public final class MockModbusServer implements Closeable {
         wrongTransactionId = true;
     }
 
+    /** RTU only: answers the next request with a CRC that does not match. */
+    void answerWithWrongCrc() {
+        wrongCrc = true;
+    }
+
+    /** Answers the next request as if another unit had sent it. */
+    void answerFromAnotherUnit() {
+        wrongUnit = true;
+    }
+
     @Override
     public void close() throws IOException {
         server.close();
@@ -112,15 +141,13 @@ public final class MockModbusServer implements Closeable {
         try (client) {
             DataInputStream in = new DataInputStream(client.getInputStream());
             while (true) {
-                ByteBuffer header = ByteBuffer.wrap(in.readNBytes(7));
-                if (header.remaining() < 7) {
+                Request request = rtu ? readRtu(in) : readMbap(in);
+                if (request == null) {
                     return;
                 }
-                int transactionId = header.getShort() & 0xFFFF;
-                header.getShort();
-                int length = header.getShort() & 0xFFFF;
-                int unitId = header.get() & 0xFF;
-                byte[] pdu = in.readNBytes(length - 1);
+                int transactionId = request.transactionId();
+                int unitId = request.unitId();
+                byte[] pdu = request.pdu();
                 if (dropNextRequest) {
                     dropNextRequest = false;
                     return;
@@ -131,7 +158,12 @@ public final class MockModbusServer implements Closeable {
                     wrongTransactionId = false;
                     transactionId++;
                 }
-                client.getOutputStream().write(frame(transactionId, unitId, answer(unitId, pdu)));
+                byte[] answer = answer(unitId, pdu);
+                if (wrongUnit) {
+                    wrongUnit = false;
+                    unitId++;
+                }
+                client.getOutputStream().write(rtu ? rtuFrame(unitId, answer) : frame(transactionId, unitId, answer));
             }
         } catch (IOException ignored) {
             // the client went away, which is a normal end of a test
@@ -140,6 +172,43 @@ public final class MockModbusServer implements Closeable {
         }
     }
 
+    private static Request readMbap(DataInputStream in) throws IOException {
+        ByteBuffer header = ByteBuffer.wrap(in.readNBytes(7));
+        if (header.remaining() < 7) {
+            return null;
+        }
+        int transactionId = header.getShort() & 0xFFFF;
+        header.getShort();
+        int length = header.getShort() & 0xFFFF;
+        int unitId = header.get() & 0xFF;
+        return new Request(transactionId, unitId, in.readNBytes(length - 1));
+    }
+
+    /** Every read request (FC3, FC4) is unit, function, four bytes and the CRC. */
+    private static Request readRtu(DataInputStream in) throws IOException {
+        byte[] frame = in.readNBytes(8);
+        if (frame.length < 8) {
+            return null;
+        }
+        int crc = RtuFraming.crc(frame, 6);
+        if ((frame[6] & 0xFF) != (crc & 0xFF) || (frame[7] & 0xFF) != crc >>> 8) {
+            throw new IOException("request with a wrong CRC");
+        }
+        return new Request(0, frame[0] & 0xFF, Arrays.copyOfRange(frame, 1, 6));
+    }
+
+    private byte[] rtuFrame(int unitId, byte[] pdu) {
+        byte[] frame = ByteBuffer.allocate(1 + pdu.length + 2).put((byte) unitId).put(pdu).array();
+        int crc = RtuFraming.crc(frame, 1 + pdu.length);
+        if (wrongCrc) {
+            wrongCrc = false;
+            crc ^= 1;
+        }
+        frame[frame.length - 2] = (byte) crc;
+        frame[frame.length - 1] = (byte) (crc >>> 8);
+        return frame;
+    }
+
     private void pause() {
         try {
             Thread.sleep(responseDelay);

gateway/src/test/java/home/gateway/adapters/modbus/RtuOverTcpTransportTest.java

@@ -0,0 +1,70 @@
+package home.gateway.adapters.modbus;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.IOException;
+import java.time.Duration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** The house meter's RS485 converter: RTU frames with their CRC over TCP, no MBAP header. */
+class RtuOverTcpTransportTest {
+
+    private static final byte[] READ_INPUT_0 = {4, 0, 0, 0, 2};
+
+    private MockModbusServer server;
+    private TcpTransport transport;
+
+    @BeforeEach
+    void start() throws IOException {
+        server = MockModbusServer.rtuOverTcp();
+        server.input(0, 0x436B);
+        server.input(1, 0xA8F6);
+        transport = TcpTransport.rtuOverTcp(new TcpTransport.Settings(
+                "localhost", server.port(), Duration.ofSeconds(2), Duration.ofSeconds(2), Duration.ZERO));
+    }
+
+    @AfterEach
+    void stop() throws IOException {
+        transport.close();
+        server.close();
+    }
+
+    @Test
+    void framesTheRequestWithTheModbusCrc() {
+        // The specification's example: read 10 holding registers from unit 1 ends in C5 CD.
+        assertArrayEquals(new byte[] {1, 3, 0, 0, 0, 10, (byte) 0xC5, (byte) 0xCD},
+                new RtuFraming().frame(1, new byte[] {3, 0, 0, 0, 10}));
+    }
+
+    @Test
+    void readsInputRegisters() throws IOException {
+        int[] words = new PduModbusClient(transport).readInputRegisters(1, 0, 2);
+
+        assertEquals(235.66f, Float.intBitsToFloat(words[0] << 16 | words[1]), 0.005f);
+    }
+
+    @Test
+    void returnsExceptionPduAsIs() throws IOException {
+        assertArrayEquals(new byte[] {(byte) 0x84, 2}, transport.exchange(1, new byte[] {4, 0, 99, 0, 1}));
+    }
+
+    @Test
+    void rejectsAWrongCrcAndReconnectsAfterwards() throws IOException {
+        server.answerWithWrongCrc();
+        assertThrows(IOException.class, () -> transport.exchange(1, READ_INPUT_0));
+
+        assertEquals(6, transport.exchange(1, READ_INPUT_0).length);
+        assertEquals(2, server.connections());
+    }
+
+    @Test
+    void rejectsAnAnswerFromAnotherUnit() {
+        server.answerFromAnotherUnit();
+
+        assertThrows(IOException.class, () -> transport.exchange(1, READ_INPUT_0));
+    }
+}

gateway/src/test/java/home/gateway/adapters/modbus/TcpTransportTest.java

@@ -66,7 +66,7 @@ class TcpTransportTest {
         transport.close();
         transport = new TcpTransport(new TcpTransport.Settings(
                 "localhost", server.port(), Duration.ofSeconds(2), Duration.ofSeconds(2), Duration.ZERO),
-                Duration.ofMillis(100));
+                new MbapFraming(), Duration.ofMillis(100));
 
         transport.exchange(1, READ_HOLDING_10);
         transport.exchange(1, READ_HOLDING_10);