blob: 942168b73014cd3d76500d89df89e3a5d66381f0 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
"""
HDLC framing for AX.25 — flags, bit-stuffing, CRC-16-CCITT.
"""
from __future__ import annotations
import sys
from pathlib import Path
_ROOT = Path(__file__).resolve().parents[3]
_DAEMON = _ROOT / "stacks" / "daemon"
if str(_DAEMON) not in sys.path:
sys.path.insert(0, str(_DAEMON))
from ax25_codec import ax25_crc # noqa: E402
HDLC_FLAG = 0x7E
HDLC_ESCAPE = 0x7D
HDLC_XOR = 0x20
def bit_stuff(data: bytes) -> bytes:
out = bytearray()
ones = 0
for b in data:
for bit_i in range(7, -1, -1):
bit = (b >> bit_i) & 1
if bit:
ones += 1
out.append(1)
if ones == 5:
out.append(0)
ones = 0
else:
ones = 0
out.append(0)
# pack bits to bytes MSB-first
buf = bytearray()
acc = 0
n = 0
for bit in out:
acc = (acc << 1) | bit
n += 1
if n == 8:
buf.append(acc)
acc = 0
n = 0
if n:
buf.append(acc << (8 - n))
return bytes(buf)
def bit_unstuff(data: bytes) -> bytes:
bits: list[int] = []
for b in data:
for bit_i in range(7, -1, -1):
bits.append((b >> bit_i) & 1)
out: list[int] = []
ones = 0
i = 0
while i < len(bits):
bit = bits[i]
if bit:
ones += 1
out.append(1)
if ones == 5 and i + 1 < len(bits) and bits[i + 1] == 0:
i += 1
ones = 0
elif ones == 6:
break
else:
ones = 0
out.append(0)
i += 1
buf = bytearray()
for j in range(0, len(out), 8):
chunk = out[j : j + 8]
if len(chunk) < 8:
break
val = 0
for bit in chunk:
val = (val << 1) | bit
buf.append(val)
return bytes(buf)
def build_hdlc_frame(ax25_body: bytes) -> bytes:
"""AX.25 body (no FCS) → on-air HDLC bit stream as bytes (pre tone encoding)."""
crc = ax25_crc(ax25_body)
payload = ax25_body + bytes((crc & 0xFF, crc >> 8))
stuffed = bit_stuff(payload)
return bytes([HDLC_FLAG]) + stuffed + bytes([HDLC_FLAG])
def parse_hdlc_stream(raw_bits: bytes) -> list[bytes]:
"""Extract AX.25 bodies (with FCS) from demodulated byte stream between flags."""
frames: list[bytes] = []
in_frame = False
cur = bytearray()
for b in raw_bits:
if b == HDLC_FLAG:
if in_frame and cur:
unstuffed = bit_unstuff(bytes(cur))
if len(unstuffed) >= 2:
frames.append(unstuffed)
cur.clear()
in_frame = True
continue
if in_frame:
cur.append(b)
return frames
|
