Implement KnxTpUartClient and transport classes with connection handling and message encoding

Signed-off-by: Tony <tonylu@tony-cloud.com>

lib/knx_stack_tpuart.dart

@@ -1,2 +1,3 @@
+export 'src/knx_tpuart_client.dart';
 export 'src/knx_tpuart_support.dart';
-export 'src/knx_tpuart_transport.dart';
+export 'src/knx_tpuart_transport.dart';

lib/src/knx_tpuart_client.dart

@@ -0,0 +1,558 @@
+import 'dart:async';
+import 'dart:typed_data';
+
+import 'package:knx_stack/knx_stack.dart';
+
+const int _resetRequest = 0x01;
+const int _resetIndication = 0x03;
+const int _stateRequest = 0x02;
+const int _stateIndicationMask = 0x07;
+const int _setAddressRequest = 0x28;
+const int _lDataConfirmPositive = 0x8b;
+const int _lDataConfirmNegative = 0x0b;
+const int _busyIndication = 0xc0;
+
+enum KnxTpUartConnectionState { disconnected, resetWait, setAddressWait, stateWait, online, error }
+
+final class KnxTpUartClient implements KnxTransport {
+  KnxTpUartClient({
+    required KnxTransport transport,
+    this.individualAddress,
+    this.initTimeout = const Duration(milliseconds: 500),
+    this.confirmationTimeout = const Duration(seconds: 2),
+    this.retryDelay = const Duration(milliseconds: 50),
+    this.maxInitRetries = 5,
+    this.maxSendAttempts = 3,
+  }) : _transport = transport,
+       _receiver = _KnxTpUartReceiver();
+
+  final KnxTransport _transport;
+  final KnxIndividualAddress? individualAddress;
+  final Duration initTimeout;
+  final Duration confirmationTimeout;
+  final Duration retryDelay;
+  final int maxInitRetries;
+  final int maxSendAttempts;
+  final _KnxTpUartReceiver _receiver;
+
+  final StreamController<Uint8List> _frames = StreamController<Uint8List>.broadcast();
+  final StreamController<CemiLDataInd> _messages = StreamController<CemiLDataInd>.broadcast();
+
+  StreamSubscription<Uint8List>? _subscription;
+  Completer<void>? _connectCompleter;
+  Timer? _initTimer;
+  Timer? _confirmationTimer;
+  _PendingSend? _pendingSend;
+  Future<void> _sendQueue = Future<void>.value();
+  Uint8List? _lastSentTelegram;
+  bool _isDisconnecting = false;
+  int _initRetries = 0;
+  KnxTpUartConnectionState _state = KnxTpUartConnectionState.disconnected;
+
+  @override
+  String get transportId => 'knx_tpuart_client';
+
+  @override
+  KnxFeature get feature => KnxFeature.tpuartTransport;
+
+  @override
+  KnxFeatureSupport get support => _transport.support;
+
+  @override
+  bool get isConnected => _state == KnxTpUartConnectionState.online;
+
+  KnxTpUartConnectionState get connectionState => _state;
+
+  @override
+  Stream<Uint8List> get inboundFrames => _frames.stream;
+
+  Stream<CemiLDataInd> get inboundMessages => _messages.stream;
+
+  Future<void> sendCemi(CemiLDataMessage frame) => send(frame.toBytes());
+
+  @override
+  Future<void> connect() async {
+    support.requireSupported('connect');
+    if (isConnected) {
+      return;
+    }
+    if (_connectCompleter != null) {
+      return _connectCompleter!.future;
+    }
+
+    _isDisconnecting = false;
+    _subscription ??= _transport.inboundFrames.listen(
+      _handleChunk,
+      onError: _handleTransportError,
+      onDone: _handleTransportDone,
+    );
+
+    final completer = Completer<void>();
+    _connectCompleter = completer;
+
+    try {
+      await _transport.connect();
+      _initRetries = 0;
+      _sendResetRequest();
+      await completer.future;
+    } catch (error) {
+      await _cleanupAfterFailure(error);
+      rethrow;
+    } finally {
+      _connectCompleter = null;
+    }
+  }
+
+  @override
+  Future<void> disconnect() async {
+    _isDisconnecting = true;
+    _cancelTimers();
+    _receiver.reset();
+    _state = KnxTpUartConnectionState.disconnected;
+    _rejectPending(StateError('TP-UART disconnected.'));
+    _lastSentTelegram = null;
+
+    await _subscription?.cancel();
+    _subscription = null;
+    await _transport.disconnect();
+    _isDisconnecting = false;
+  }
+
+  @override
+  Future<void> send(Uint8List frame) {
+    support.requireSupported('send');
+    _sendQueue = _sendQueue.then((_) => _sendFrame(frame), onError: (_) => _sendFrame(frame));
+    return _sendQueue;
+  }
+
+  Future<void> _sendFrame(Uint8List frame) async {
+    if (!isConnected) {
+      throw StateError('TP-UART client is not connected.');
+    }
+    if (_pendingSend != null) {
+      throw StateError('Another TP-UART frame is already waiting for confirmation.');
+    }
+
+    final telegram = _cemiToTpTelegram(frame);
+    if (telegram == null) {
+      throw ArgumentError('Only cEMI L_Data.req frames with zero additional info are supported.');
+    }
+
+    final completer = Completer<void>();
+    _pendingSend = _PendingSend(telegram: telegram, completer: completer);
+    await _transmitPendingSend();
+    await completer.future;
+  }
+
+  void _handleChunk(Uint8List chunk) {
+    _receiver.addChunk(chunk, onControlByte: _handleControlByte, onTelegram: _handleTpTelegram);
+  }
+
+  void _handleControlByte(int byte) {
+    if (byte == _resetIndication) {
+      if (_state == KnxTpUartConnectionState.resetWait) {
+        _initRetries = 0;
+        _initTimer?.cancel();
+        if (individualAddress != null) {
+          _state = KnxTpUartConnectionState.setAddressWait;
+          unawaited(
+            _transport.send(
+              Uint8List.fromList([_setAddressRequest, ...individualAddress!.toBytes()]),
+            ),
+          );
+        }
+        _requestState();
+      } else if (isConnected) {
+        _initRetries = 0;
+        _sendResetRequest();
+      }
+      return;
+    }
+
+    if (byte == _busyIndication) {
+      _retryPendingSend(StateError('TP-UART bus busy.'));
+      return;
+    }
+
+    if (byte == _lDataConfirmPositive || byte == _lDataConfirmNegative) {
+      final pending = _pendingSend;
+      _confirmationTimer?.cancel();
+      _confirmationTimer = null;
+      _pendingSend = null;
+      _lastSentTelegram = null;
+      if (pending == null) {
+        return;
+      }
+      if (byte == _lDataConfirmPositive) {
+        if (!pending.completer.isCompleted) {
+          pending.completer.complete();
+        }
+      } else if (!pending.completer.isCompleted) {
+        pending.completer.completeError(StateError('TP-UART negative confirmation.'));
+      }
+      return;
+    }
+
+    if ((byte & _stateIndicationMask) == _stateIndicationMask) {
+      _initRetries = 0;
+      _initTimer?.cancel();
+      _initTimer = null;
+      _state = KnxTpUartConnectionState.online;
+      final completer = _connectCompleter;
+      if (completer != null && !completer.isCompleted) {
+        completer.complete();
+      }
+    }
+  }
+
+  void _handleTpTelegram(Uint8List telegram) {
+    if (_lastSentTelegram != null &&
+        _telegramEqualsIgnoringRepeatBit(telegram, _lastSentTelegram!)) {
+      _lastSentTelegram = null;
+      return;
+    }
+
+    final cemi = _tpTelegramToCemi(telegram);
+    if (cemi == null) {
+      return;
+    }
+
+    _frames.add(cemi);
+    final message = CEMI.fromBytes(cemi);
+    if (message is CemiLDataInd) {
+      _messages.add(message);
+    }
+  }
+
+  void _handleTransportError(Object error) {
+    if (_isDisconnecting) {
+      return;
+    }
+    _state = KnxTpUartConnectionState.error;
+    final completer = _connectCompleter;
+    if (completer != null && !completer.isCompleted) {
+      completer.completeError(error);
+    }
+    _rejectPending(error);
+  }
+
+  void _handleTransportDone() {
+    if (_isDisconnecting) {
+      return;
+    }
+    _handleTransportError(StateError('Underlying TP-UART transport closed.'));
+  }
+
+  Future<void> _cleanupAfterFailure(Object error) async {
+    _cancelTimers();
+    _receiver.reset();
+    _state = KnxTpUartConnectionState.error;
+    _rejectPending(error);
+    await _subscription?.cancel();
+    _subscription = null;
+    await _transport.disconnect();
+  }
+
+  void _sendResetRequest() {
+    _state = KnxTpUartConnectionState.resetWait;
+    unawaited(_transport.send(Uint8List.fromList([_resetRequest])));
+    _armInitTimer(
+      expectedState: KnxTpUartConnectionState.resetWait,
+      action: _sendResetRequest,
+      errorMessage: 'TP-UART reset timeout.',
+    );
+  }
+
+  void _requestState() {
+    _state = KnxTpUartConnectionState.stateWait;
+    unawaited(_transport.send(Uint8List.fromList([_stateRequest])));
+    _armInitTimer(
+      expectedState: KnxTpUartConnectionState.stateWait,
+      action: _requestState,
+      errorMessage: 'TP-UART state request timeout.',
+    );
+  }
+
+  void _armInitTimer({
+    required KnxTpUartConnectionState expectedState,
+    required void Function() action,
+    required String errorMessage,
+  }) {
+    _initTimer?.cancel();
+    _initTimer = Timer(initTimeout, () {
+      if (_state != expectedState) {
+        return;
+      }
+      _initRetries += 1;
+      if (_initRetries < maxInitRetries) {
+        action();
+        return;
+      }
+      _state = KnxTpUartConnectionState.error;
+      final error = TimeoutException(errorMessage, initTimeout);
+      final completer = _connectCompleter;
+      if (completer != null && !completer.isCompleted) {
+        completer.completeError(error);
+      }
+    });
+  }
+
+  Future<void> _transmitPendingSend() async {
+    final pending = _pendingSend;
+    if (pending == null) {
+      return;
+    }
+
+    final telegram = _telegramForAttempt(pending.telegram, pending.attempts);
+    _lastSentTelegram = telegram;
+    _confirmationTimer?.cancel();
+    _confirmationTimer = Timer(confirmationTimeout, () {
+      _retryPendingSend(
+        TimeoutException('Timed out waiting for TP-UART confirmation.', confirmationTimeout),
+      );
+    });
+
+    try {
+      await _transport.send(_wrapTpUartServices(telegram));
+    } catch (error) {
+      _confirmationTimer?.cancel();
+      _confirmationTimer = null;
+      _retryPendingSend(error);
+    }
+  }
+
+  void _retryPendingSend(Object error) {
+    final pending = _pendingSend;
+    if (pending == null) {
+      return;
+    }
+    _confirmationTimer?.cancel();
+    _confirmationTimer = null;
+    _lastSentTelegram = null;
+
+    if (pending.attempts + 1 < maxSendAttempts) {
+      pending.attempts += 1;
+      unawaited(Future<void>.delayed(retryDelay, _transmitPendingSend));
+      return;
+    }
+
+    _pendingSend = null;
+    if (!pending.completer.isCompleted) {
+      pending.completer.completeError(error);
+    }
+  }
+
+  Uint8List _telegramForAttempt(Uint8List telegram, int attempts) {
+    final bytes = Uint8List.fromList(telegram);
+    if (attempts > 0 && (bytes[0] & 0x20) != 0) {
+      bytes[0] &= 0xdf;
+      bytes[bytes.length - 1] = _tpChecksum(bytes, excludeChecksum: true);
+    }
+    return bytes;
+  }
+
+  void _rejectPending(Object error) {
+    _confirmationTimer?.cancel();
+    _confirmationTimer = null;
+    final pending = _pendingSend;
+    _pendingSend = null;
+    if (pending != null && !pending.completer.isCompleted) {
+      pending.completer.completeError(error);
+    }
+  }
+
+  void _cancelTimers() {
+    _initTimer?.cancel();
+    _initTimer = null;
+    _confirmationTimer?.cancel();
+    _confirmationTimer = null;
+  }
+}
+
+final class _PendingSend {
+  _PendingSend({required this.telegram, required this.completer});
+
+  final Uint8List telegram;
+  final Completer<void> completer;
+  int attempts = 0;
+}
+
+final class _KnxTpUartReceiver {
+  final Duration interByteTimeout = const Duration(seconds: 1);
+  final List<int> _buffer = <int>[];
+  bool _extendedFrame = false;
+  DateTime? _lastRead;
+
+  void addChunk(
+    Uint8List chunk, {
+    required void Function(int byte) onControlByte,
+    required void Function(Uint8List telegram) onTelegram,
+  }) {
+    for (final byte in chunk) {
+      _processByte(byte, onControlByte: onControlByte, onTelegram: onTelegram);
+    }
+  }
+
+  void reset() {
+    _buffer.clear();
+    _extendedFrame = false;
+    _lastRead = null;
+  }
+
+  void _processByte(
+    int byte, {
+    required void Function(int byte) onControlByte,
+    required void Function(Uint8List telegram) onTelegram,
+  }) {
+    if (_buffer.isEmpty) {
+      if (_isControlByte(byte)) {
+        onControlByte(byte);
+        return;
+      }
+      if (byte == 0xcb || (byte & 0x17) == 0x13) {
+        return;
+      }
+    }
+
+    final now = DateTime.now();
+    if (_buffer.isNotEmpty && _lastRead != null && now.difference(_lastRead!) > interByteTimeout) {
+      _buffer.clear();
+    }
+
+    if (_buffer.isEmpty) {
+      if (_isFrameStart(byte)) {
+        _buffer.add(byte);
+        _lastRead = now;
+      }
+      return;
+    }
+
+    _buffer.add(byte);
+    _lastRead = now;
+    _checkCompleteFrame(onTelegram);
+  }
+
+  bool _isControlByte(int byte) {
+    return byte == _resetIndication ||
+        byte == _lDataConfirmPositive ||
+        byte == _lDataConfirmNegative ||
+        byte == _busyIndication ||
+        (byte & _stateIndicationMask) == _stateIndicationMask;
+  }
+
+  bool _isFrameStart(int byte) {
+    _extendedFrame = (byte & 0x80) == 0;
+    return (byte & 0x50) == 0x10;
+  }
+
+  void _checkCompleteFrame(void Function(Uint8List telegram) onTelegram) {
+    final minLength = _extendedFrame ? 7 : 6;
+    if (_buffer.length < minLength) {
+      return;
+    }
+    final payloadLength = _extendedFrame ? _buffer[6] : (_buffer[5] & 0x0f);
+    final totalLength = payloadLength + (_extendedFrame ? 9 : 8);
+    if (_buffer.length < totalLength) {
+      return;
+    }
+    final frame = Uint8List.fromList(_buffer.sublist(0, totalLength));
+    if (_validateTpChecksum(frame)) {
+      onTelegram(frame);
+    }
+    reset();
+  }
+}
+
+Uint8List? _cemiToTpTelegram(Uint8List data) {
+  if (data.length < 10 || data[1] != 0x00) {
+    return null;
+  }
+  final ctrl = data.sublist(2);
+  final standard = (ctrl[0] & 0x80) != 0;
+  final tpLength = standard ? data.length - 2 : data.length - 1;
+  if (tpLength < 8) {
+    return null;
+  }
+
+  final telegram = Uint8List(tpLength);
+  if (standard) {
+    telegram[0] = ctrl[0];
+    telegram.setRange(1, 5, ctrl.sublist(2, 6));
+    telegram[5] = (ctrl[1] & 0xf0) | (ctrl[6] & 0x0f);
+    if (tpLength > 7) {
+      telegram.setRange(6, 6 + tpLength - 7, ctrl.sublist(7, 7 + tpLength - 7));
+    }
+  } else {
+    telegram.setRange(0, tpLength - 1, ctrl.sublist(0, tpLength - 1));
+  }
+  telegram[telegram.length - 1] = _tpChecksum(telegram, excludeChecksum: true);
+  return telegram;
+}
+
+Uint8List? _tpTelegramToCemi(Uint8List data) {
+  if (data.length < 8 || !_validateTpChecksum(data)) {
+    return null;
+  }
+  final extended = _isExtendedTpFrame(data);
+  final cemiLength = data.length + (extended ? 2 : 3) - 1;
+  final cemi = Uint8List(cemiLength);
+  cemi[0] = cemiLDataIndMessageCode;
+  cemi[1] = 0x00;
+  cemi[2] = data[0];
+  if (extended) {
+    cemi.setRange(2, 2 + data.length - 1, data.sublist(0, data.length - 1));
+  } else {
+    cemi[3] = data[5] & 0xf0;
+    cemi.setRange(4, 8, data.sublist(1, 5));
+    cemi[8] = data[5] & 0x0f;
+    final copyLength = cemi[8] + 1;
+    if (9 + copyLength > cemi.length || 6 + copyLength > data.length) {
+      return null;
+    }
+    cemi.setRange(9, 9 + copyLength, data.sublist(6, 6 + copyLength));
+  }
+  return cemi;
+}
+
+Uint8List _wrapTpUartServices(Uint8List telegram) {
+  final result = Uint8List(telegram.length * 2);
+  for (var index = 0; index < telegram.length; index += 1) {
+    final control = (index == telegram.length - 1 ? 0x40 : 0x80) | (index & 0x3f);
+    result[index * 2] = control;
+    result[index * 2 + 1] = telegram[index];
+  }
+  return result;
+}
+
+bool _isExtendedTpFrame(Uint8List data) => data.isNotEmpty && (data[0] & 0xd3) == 0x10;
+
+bool _validateTpChecksum(Uint8List data) {
+  if (data.length < 2) {
+    return false;
+  }
+  return data[data.length - 1] == _tpChecksum(data, excludeChecksum: true);
+}
+
+int _tpChecksum(Uint8List data, {required bool excludeChecksum}) {
+  var checksum = 0xff;
+  final limit = excludeChecksum ? data.length - 1 : data.length;
+  for (var index = 0; index < limit; index += 1) {
+    checksum ^= data[index];
+  }
+  return checksum & 0xff;
+}
+
+bool _telegramEqualsIgnoringRepeatBit(Uint8List left, Uint8List right) {
+  if (left.length != right.length) {
+    return false;
+  }
+  if ((left[0] & ~0x20) != (right[0] & ~0x20)) {
+    return false;
+  }
+  for (var index = 1; index < left.length; index += 1) {
+    if (left[index] != right[index]) {
+      return false;
+    }
+  }
+  return true;
+}

lib/src/knx_tpuart_transport_io.dart

@@ -7,7 +7,11 @@ import 'package:knx_stack/knx_stack.dart';
 import 'knx_tpuart_support.dart';
 
 final class KnxTpUartTransport implements KnxTransport {
-  KnxTpUartTransport({required this.portName, this.baudRate = 19200, this.readTimeout = const Duration(milliseconds: 50)});
+  KnxTpUartTransport({
+    required this.portName,
+    this.baudRate = 19200,
+    this.readTimeout = const Duration(milliseconds: 50),
+  });
 
   final String portName;
   final int baudRate;
@@ -51,7 +55,7 @@ final class KnxTpUartTransport implements KnxTransport {
     final config = SerialPortConfig()
       ..baudRate = baudRate
       ..bits = 8
-      ..parity = SerialPortParity.none
+      ..parity = SerialPortParity.even
       ..stopBits = 1
       ..setFlowControl(SerialPortFlowControl.none);
     serialPort.config = config;
@@ -90,4 +94,4 @@ final class KnxTpUartTransport implements KnxTransport {
       throw StateError('Short serial write: expected ${frame.length}, wrote $bytesWritten.');
     }
   }
-}
+}

lib/src/knx_tpuart_transport_stub.dart

@@ -5,10 +5,15 @@ import 'package:knx_stack/knx_stack.dart';
 import 'knx_tpuart_support.dart';
 
 final class KnxTpUartTransport implements KnxTransport {
-  KnxTpUartTransport({required this.portName, this.baudRate = 19200});
+  KnxTpUartTransport({
+    required this.portName,
+    this.baudRate = 19200,
+    this.readTimeout = const Duration(milliseconds: 50),
+  });
 
   final String portName;
   final int baudRate;
+  final Duration readTimeout;
 
   @override
   String get transportId => 'knx_tpuart';
@@ -33,4 +38,4 @@ final class KnxTpUartTransport implements KnxTransport {
 
   @override
   Future<void> send(Uint8List frame) async => support.requireSupported('send');
-}
+}