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
|
#!/usr/bin/env python3
"""One-shot AX.25 UI beacon matching HyBBX [broadcast] INI (HyBBX must stay running).
Connects to the internal circuit hub and sends an HBX AX25_UI frame with TX flag
the same way hybbx_broadcast_ax25() does. Requires packet_radio1 linked with
link_password on the Main.
"""
import argparse
import socket
import struct
import sys
import time
MAGIC = b"HBX"
VERSION = 0x01
HEADER_SIZE = 12
PROTO_LINK_AUTH = 0x03
PROTO_LINK_AUTH_ACK = 0x04
PROTO_AX25_UI = 0x02
FLAG_TX = 0x0002
FLAG_PATH = 0x0004
ADDR_LEN = 7
def be16(v):
return struct.pack(">H", v)
def be32(v):
return struct.pack(">I", v)
def encode_frame(proto, flags, payload):
return bytes([ord("H"), ord("B"), ord("X"), VERSION, proto]) + be16(flags) + be32(len(payload)) + payload
def encode_addr(call, ssid=0, last=False):
call = call.upper().ljust(6)[:6]
raw = bytes((ord(c) << 1) for c in call)
ssid_b = ((ssid & 0x0F) << 1) | (0x60 if not last else 0x61)
return raw + bytes([ssid_b])
def pack_ax25_ui(dest, src, message):
body = bytes([0]) # digi count
body += encode_addr(dest, last=False)
body += encode_addr(src, last=True)
body += message.encode("ascii", errors="replace")
return encode_frame(PROTO_AX25_UI, FLAG_TX | FLAG_PATH, body)
def link_auth_payload(password, link_id, role, baud=2400, duplex="half", bandwidth="low", mhz="27.235"):
lines = [
f"password={password}",
f"role={role}",
f"id={link_id}",
f"baud={baud}",
f"duplex={duplex}",
f"bandwidth={bandwidth}",
f"frequency_mhz={mhz}",
]
return "\n".join(lines).encode("ascii")
def recv_frames(sock, timeout=5.0):
sock.settimeout(timeout)
buf = b""
try:
while True:
chunk = sock.recv(4096)
if not chunk:
break
buf += chunk
except socket.timeout:
pass
return buf
def main():
ap = argparse.ArgumentParser(description="HyBBX one-shot AX.25 RF beacon via circuit hub")
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--port", type=int, default=7323)
ap.add_argument("--password", default="changeme")
ap.add_argument("--link-id", default="beacon-once")
ap.add_argument("--mycall", default="CB-0")
ap.add_argument("--dest", default="*")
ap.add_argument("--message", default="Broadcast: CB-0 online")
args = ap.parse_args()
sock = socket.create_connection((args.host, args.port), timeout=5)
auth = link_auth_payload(args.password, args.link_id, "gateway")
sock.sendall(encode_frame(PROTO_LINK_AUTH, 0, auth))
ack = recv_frames(sock, 3.0)
if b"ok=yes" not in ack:
print("LINK_AUTH failed:", ack[:200], file=sys.stderr)
sys.exit(1)
print(f"LINK_AUTH ok — TX {args.mycall} -> {args.dest}: {args.message!r}")
sock.sendall(pack_ax25_ui(args.dest, args.mycall, args.message))
time.sleep(2)
sock.close()
print("Beacon frame sent to circuit hub.")
if __name__ == "__main__":
main()
|
