Coverage for kv4p/protocol/kiss.py: 100%

1"""KISS framing.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from collections.abc import Callable 

7 

8from kv4p.constants.kiss import KISS_FEND, KISS_FESC, KISS_TFEND, KISS_TFESC 

9 

10logger = logging.getLogger(__name__) 

11 

12 

13class KissParser: 

14 """Incremental KISS frame parser.""" 

15 

16 def __init__(self, on_frame: Callable[[int, bytes], None]) -> None: 

17 self._on_frame = on_frame 

18 self._in_frame = False 

19 self._escaped = False 

20 self._buf = bytearray() 

21 

22 def feed(self, data: bytes) -> None: 

23 """Feed serial bytes.""" 

24 for byte in data: 

25 self._feed_byte(byte) 

26 

27 def _feed_byte(self, byte: int) -> None: 

28 if byte == KISS_FEND: 

29 if self._in_frame and self._buf: 

30 command = self._buf[0] 

31 payload = bytes(self._buf[1:]) 

32 logger.debug("serial rx KISS command=0x%02x payload=%d", command, len(payload)) 

33 self._on_frame(command, payload) 

34 self._buf.clear() 

35 self._in_frame = True 

36 self._escaped = False 

37 return 

38 

39 if not self._in_frame: 

40 return 

41 

42 if self._escaped: 

43 if byte == KISS_TFEND: 

44 self._buf.append(KISS_FEND) 

45 elif byte == KISS_TFESC: 

46 self._buf.append(KISS_FESC) 

47 else: 

48 logger.warning("invalid KISS escape byte 0x%02x", byte) 

49 self._escaped = False 

50 return 

51 

52 if byte == KISS_FESC: 

53 self._escaped = True 

54 return 

55 

56 self._buf.append(byte) 

57 

58 

59def encode_kiss_frame(command: int, payload: bytes) -> bytes: 

60 """Encode a KISS frame.""" 

61 out = bytearray([KISS_FEND, command]) 

62 for byte in payload: 

63 if byte == KISS_FEND: 

64 out.extend((KISS_FESC, KISS_TFEND)) 

65 elif byte == KISS_FESC: 

66 out.extend((KISS_FESC, KISS_TFESC)) 

67 else: 

68 out.append(byte) 

69 out.append(KISS_FEND) 

70 return bytes(out)