diff --git a/lib/knx_stack_tpuart.dart b/lib/knx_stack_tpuart.dart index 8013fd0..bee40dd 100644 --- a/lib/knx_stack_tpuart.dart +++ b/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'; \ No newline at end of file +export 'src/knx_tpuart_transport.dart'; diff --git a/lib/src/knx_tpuart_client.dart b/lib/src/knx_tpuart_client.dart new file mode 100644 index 0000000..991fe9e --- /dev/null +++ b/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 _frames = StreamController.broadcast(); + final StreamController _messages = StreamController.broadcast(); + + StreamSubscription? _subscription; + Completer? _connectCompleter; + Timer? _initTimer; + Timer? _confirmationTimer; + _PendingSend? _pendingSend; + Future _sendQueue = Future.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 get inboundFrames => _frames.stream; + + Stream get inboundMessages => _messages.stream; + + Future sendCemi(CemiLDataMessage frame) => send(frame.toBytes()); + + @override + Future 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(); + _connectCompleter = completer; + + try { + await _transport.connect(); + _initRetries = 0; + _sendResetRequest(); + await completer.future; + } catch (error) { + await _cleanupAfterFailure(error); + rethrow; + } finally { + _connectCompleter = null; + } + } + + @override + Future 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 send(Uint8List frame) { + support.requireSupported('send'); + _sendQueue = _sendQueue.then((_) => _sendFrame(frame), onError: (_) => _sendFrame(frame)); + return _sendQueue; + } + + Future _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(); + _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 _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 _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.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 completer; + int attempts = 0; +} + +final class _KnxTpUartReceiver { + final Duration interByteTimeout = const Duration(seconds: 1); + final List _buffer = []; + 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; +} diff --git a/lib/src/knx_tpuart_transport_io.dart b/lib/src/knx_tpuart_transport_io.dart index 3f0da7c..f24c3da 100644 --- a/lib/src/knx_tpuart_transport_io.dart +++ b/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.'); } } -} \ No newline at end of file +} diff --git a/lib/src/knx_tpuart_transport_stub.dart b/lib/src/knx_tpuart_transport_stub.dart index deb3197..6c0934e 100644 --- a/lib/src/knx_tpuart_transport_stub.dart +++ b/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 send(Uint8List frame) async => support.requireSupported('send'); -} \ No newline at end of file +} diff --git a/test/knx_tpuart_client_test.dart b/test/knx_tpuart_client_test.dart new file mode 100644 index 0000000..8226086 --- /dev/null +++ b/test/knx_tpuart_client_test.dart @@ -0,0 +1,166 @@ +import 'dart:async'; +import 'dart:typed_data'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:knx_stack/knx_stack.dart'; +import 'package:knx_stack_tpuart/knx_stack_tpuart.dart'; + +void main() { + group('KnxTpUartClient', () { + test('completes the reset and state handshake', () async { + final transport = _FakeKnxTransport(); + final client = KnxTpUartClient(transport: transport); + + final connectFuture = client.connect(); + await _flushMicrotasks(); + + expect(transport.sentFrames, hasLength(1)); + expect(transport.sentFrames.single, [0x01]); + + transport.emit(const [0x03]); + await _flushMicrotasks(); + + expect(transport.sentFrames, hasLength(2)); + expect(transport.sentFrames.last, [0x02]); + + transport.emit(const [0x07]); + await connectFuture; + + expect(client.isConnected, isTrue); + }); + + test('encodes cEMI writes into TP-UART service bytes', () async { + final transport = _FakeKnxTransport(); + final client = KnxTpUartClient(transport: transport); + await _connectClient(client, transport); + + final sendFuture = client.send(_sampleCemiWrite().toBytes()); + await _flushMicrotasks(); + + expect(transport.sentFrames.last, [ + 0x80, + 0xbc, + 0x81, + 0x11, + 0x82, + 0x01, + 0x83, + 0x00, + 0x84, + 0x01, + 0x85, + 0xe2, + 0x86, + 0x00, + 0x87, + 0x80, + 0x88, + 0x64, + 0x49, + 0x54, + ]); + + transport.emit(const [0x8b]); + await sendFuture; + }); + + test('decodes inbound TP telegrams into cEMI indications', () async { + final transport = _FakeKnxTransport(); + final client = KnxTpUartClient(transport: transport); + await _connectClient(client, transport); + + final frameFuture = client.inboundFrames.first; + final messageFuture = client.inboundMessages.first; + + transport.emit(const [0xbc, 0x11, 0x01, 0x00, 0x01, 0xe2, 0x00, 0x80, 0x64, 0x54]); + + expect(await frameFuture, [ + 0x29, + 0x00, + 0xbc, + 0xe0, + 0x11, + 0x01, + 0x00, + 0x01, + 0x02, + 0x00, + 0x80, + 0x64, + ]); + expect((await messageFuture).destinationAddress.canonical, '0/0/1'); + }); + }); +} + +Future _connectClient(KnxTpUartClient client, _FakeKnxTransport transport) async { + final connectFuture = client.connect(); + await _flushMicrotasks(); + transport.emit(const [0x03]); + await _flushMicrotasks(); + transport.emit(const [0x07]); + await connectFuture; +} + +Future _flushMicrotasks() async { + await Future.delayed(Duration.zero); +} + +CemiLDataReq _sampleCemiWrite() { + final controlField2 = ExtendedControlField() + ..addressType = AddressType.group + ..hopCount = 6 + ..eff = ExtendedFrameFormat.pointToPointOrStandardGroupAddressedLDataExtendedFrame; + + return CemiLDataReq( + controlField1: ControlField(0xbc), + controlField2: controlField2, + sourceAddress: KnxIndividualAddress.parse('1.1.1'), + destinationAddress: KnxGroupAddress.parse('0/0/1'), + tpdu: TPDU(data: Uint8List.fromList([0x64])), + ); +} + +final class _FakeKnxTransport implements KnxTransport { + final StreamController _frames = StreamController.broadcast(); + final List sentFrames = []; + + bool _isConnected = false; + + @override + String get transportId => 'fake_tpuart'; + + @override + KnxFeature get feature => KnxFeature.tpuartTransport; + + @override + KnxFeatureSupport get support => const KnxFeatureSupport.supported( + feature: KnxFeature.tpuartTransport, + platform: KnxPlatform.macos, + ); + + @override + bool get isConnected => _isConnected; + + @override + Stream get inboundFrames => _frames.stream; + + @override + Future connect() async { + _isConnected = true; + } + + @override + Future disconnect() async { + _isConnected = false; + } + + @override + Future send(Uint8List frame) async { + sentFrames.add(Uint8List.fromList(frame)); + } + + void emit(List frame) { + _frames.add(Uint8List.fromList(frame)); + } +}