LoRa port for TARPN node: Difference between revisions
No edit summary |
No edit summary |
||
Line 186: | Line 186: | ||
# Replace KissHelper.py with this contents (this extends it to handle all KISS packets not just APRS, and makes it a little more tolerant of unexpected failures rather than just stopping the process at the first weird packet): | # Replace KissHelper.py with this contents (this extends it to handle all KISS packets not just APRS, and makes it a little more tolerant of unexpected failures rather than just stopping the process at the first weird packet): | ||
<nowiki> | <nowiki> | ||
#!/usr/bin/python3 | |||
# This program is free software: you can redistribute it and/or modify | |||
# it under the terms of the GNU General Public License as published by | |||
# the Free Software Foundation, either version 3 of the License, or | |||
# (at your option) any later version. | |||
# | |||
# This program is distributed in the hope that it will be useful, | |||
# but WITHOUT ANY WARRANTY; without even the implied warranty of | |||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |||
# GNU General Public License for more details. | |||
# | |||
# You should have received a copy of the GNU General Public License | |||
# along with this program. If not, see <http://www.gnu.org/licenses/>. | |||
# This program provides basic KISS AX.25 APRS frame encoding and decoding. | |||
# Note that only APRS relevant structures are tested. It might not work | |||
# for generic AX.25 frames. | |||
# | |||
# Inspired by: | |||
# * Python script to decode AX.25 from KISS frames over a serial TNC | |||
# https://gist.github.com/mumrah/8fe7597edde50855211e27192cce9f88 | |||
# | |||
# * Sending a raw AX.25 frame with Python | |||
# https://thomask.sdf.org/blog/2018/12/15/sending-raw-ax25-python.html | |||
# | |||
# KISS-TNC for LoRa radio modem | |||
# https://github.com/IZ7BOJ/RPi-LoRa-KISS-TNC | |||
import struct | |||
import datetime | |||
import config | |||
KISS_FEND = 0xC0 # Frame start/end marker | |||
KISS_FESC = 0xDB # Escape character | |||
KISS_TFEND = 0xDC # If after an escape, means there was an 0xC0 in the source message | |||
KISS_TFESC = 0xDD # If after an escape, means there was an 0xDB in the source message | |||
# APRS data types | |||
DATA_TYPES_POSITION = b"!'/@`" | |||
DATA_TYPE_MESSAGE = b":" | |||
DATA_TYPE_THIRD_PARTY = b"}" | |||
def logf(message): | |||
timestamp = datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S - ') | |||
if config.log_enable: | |||
fileLog = open(config.logpath,"a") | |||
fileLog.write(timestamp + message+"\n") | |||
fileLog.close() | |||
print(timestamp + message) | |||
# from LoRa OE_Style to KISS | |||
# Addresses must be 6 bytes plus the SSID byte, each character shifted left by 1 | |||
# If it's the final address in the header, set the low bit to 1 | |||
# If it has been digipeated, set the H bit to 1 | |||
# Ignoring command/response for simple example | |||
def encode_address(s, final): | |||
H=0b00000000 #default H bit low | |||
if chr(s[-1])=='*': #example: WIDE1-1* or WIDE1* | |||
s=s[:-1] | |||
H=0b10000000 | |||
if b"-" not in s: | |||
s = s + b"-0" # default to SSID 0 | |||
call, ssid = s.split(b'-') | |||
if len(call) < 6: | |||
call = call + b" "*(6 - len(call)) # pad with spaces | |||
encoded_call = [x << 1 for x in call[0:6]] | |||
encoded_ssid = (int(ssid) << 1) | H | 0b01100000 | (0b00000001 if final else 0) | |||
return encoded_call + [encoded_ssid] | |||
def decode_address(data, cursor): | |||
(a1, a2, a3, a4, a5, a6, a7) = struct.unpack("<BBBBBBB", data[cursor:cursor + 7]) | |||
hrr = a7 >> 5 | |||
ssid = (a7 >> 1) & 0xf | |||
ext = a7 & 0x1 | |||
addr = struct.pack("<BBBBBB", a1 >> 1, a2 >> 1, a3 >> 1, a4 >> 1, a5 >> 1, a6 >> 1) | |||
if ssid != 0: | |||
call = addr.strip() + "-{}".format(ssid).encode() | |||
else: | |||
call = addr | |||
return (call, hrr, ext) | |||
def ax25parser(frame): #extracts fields from ax25 frames and add signal report do payload, if specified | |||
try: | |||
pos = 0 | |||
# DST | |||
(dest_addr, dest_hrr, dest_ext) = decode_address(frame, pos) | |||
pos += 7 | |||
print("DST: ", dest_addr) | |||
# SRC | |||
(src_addr, src_hrr, src_ext) = decode_address(frame, pos) | |||
pos += 7 | |||
print("SRC: ", src_addr) | |||
# REPEATERS | |||
ext = src_ext | |||
rpt_list = b"" | |||
while ext == 0: | |||
rpt_addr, rpt_hrr, ext = decode_address(frame, pos) | |||
if rpt_hrr==b'111': # H bit high-->packet has been digipeated | |||
rpt_addr+=b'*' | |||
rpt_list += b","+rpt_addr | |||
print("RPT: ", rpt_addr) | |||
pos += 7 | |||
# CTRL | |||
ctrl = frame[pos] | |||
pos += 1 | |||
if (ctrl & 0x3) == 0x3: | |||
#(pid,) = struct.unpack("<B", frame[pos]) | |||
try: | |||
pid = frame[pos] | |||
except Exception: | |||
pid = 240 | |||
print("PID: "+str(pid)) | |||
pos += 1 | |||
elif (ctrl & 0x3) == 0x1: | |||
# decode_sframe(ctrl, frame, pos) | |||
logf("SFRAME") | |||
return None,None,None,None,None | |||
elif (ctrl & 0x1) == 0x0: | |||
# decode_iframe(ctrl, frame, pos) | |||
logf("IFRAME") | |||
return None,None,None,None,None | |||
payload=frame[pos:] | |||
print("payload: ", payload) | |||
try: | |||
dti=payload[0] | |||
except Exception: | |||
dti=":" | |||
logf("Extracted AX25 parameters from AX25 Frame") | |||
logf("From: "+repr(src_addr)[2:-1]+" To: "+repr(dest_addr)[2:-1]+" Via: "+repr(rpt_list)[3:-1]+" PID: "+str(hex(pid))+" Payload: "+str(payload)[str(payl | |||
oad).find("'"):-1]) | |||
return src_addr,dest_addr,rpt_list,payload,dti | |||
except Exception: | |||
return None,None,None,None,None | |||
def encode_kiss_AX25(frame,signalreport): #from Lora to Kiss, Standard AX25 | |||
src_addr,dest_addr,rpt_list,payload,dti=ax25parser(frame) #only for logging | |||
# Escape the packet in case either KISS_FEND or KISS_FESC ended up in our stream | |||
if config.appendSignalReport and str(dti) != DATA_TYPE_MESSAGE: | |||
frame += b" "+str.encode(signalreport,'utf-8') | |||
packet_escaped = [] | |||
for x in frame: | |||
if x == KISS_FEND: | |||
packet_escaped += [KISS_FESC, KISS_TFEND] | |||
elif x == KISS_FESC: | |||
packet_escaped += [KISS_FESC, KISS_TFESC] | |||
else: | |||
packet_escaped += [x] | |||
# Build the frame that we will send to aprx and turn it into a string | |||
kiss_cmd = 0x00 # Two nybbles combined - TNC 0, command 0 (send data) | |||
kiss_frame = [KISS_FEND, kiss_cmd] + packet_escaped + [KISS_FEND] | |||
try: | |||
output = bytearray(kiss_frame) | |||
except ValueError: | |||
logf("Invalid value in frame.") | |||
return None | |||
return output | |||
def encode_kiss_OE(frame,signalreport): #from Lora to Kiss, OE_Style | |||
# Ugly frame disassembling | |||
if not b":" in frame: | |||
logf("Can't decode OE LoRa Frame") | |||
return None | |||
path = frame.split(b":")[0] | |||
payload = frame[frame.find(b":")+1:] | |||
dti = payload[0] | |||
src_addr = path.split(b">")[0] | |||
digis = path[path.find(b">") + 1:].split(b",") | |||
dest_addr = digis.pop(0) | |||
# destination address | |||
packet = encode_address(dest_addr.upper(), False) | |||
# source address | |||
packet += encode_address(src_addr.upper(), len(digis) == 0) | |||
# digipeaters | |||
for digi in digis: | |||
final_addr = digis.index(digi) == len(digis) - 1 | |||
packet += encode_address(digi.upper(), final_addr) | |||
# control field | |||
packet += [0x03] # This is an UI frame | |||
# protocol ID | |||
packet += [0xF0] # No protocol | |||
# information field | |||
logf("Extracted AX25 parameters from OE LoRa Frame") | |||
logf("From: "+repr(src_addr)[2:-1]+" To: "+repr(dest_addr)[2:-1]+" Via: "+str(digis)[1:-1].replace("b","").replace("'","")+" Payload: "+repr(payload)[2:-1]) | |||
packet += payload | |||
if config.appendSignalReport and str(dti) != DATA_TYPE_MESSAGE: | |||
#some SW (es OE5BPA) append newline character at the end of packet. Must be cut for appending signal report | |||
if chr(packet[-1])=="\n": | |||
packet=packet[:-1] | |||
packet += b" "+str.encode(signalreport,'utf-8') | |||
# Escape the packet in case either KISS_FEND or KISS_FESC ended up in our stream | |||
packet_escaped = [] | |||
for x in packet: | |||
if x == KISS_FEND: | |||
packet_escaped += [KISS_FESC, KISS_TFEND] | |||
elif x == KISS_FESC: | |||
packet_escaped += [KISS_FESC, KISS_TFESC] | |||
else: | |||
packet_escaped += [x] | |||
# Build the frame that we will send to Dire Wolf and turn it into a string | |||
kiss_cmd = 0x00 # Two nybbles combined - TNC 0, command 0 (send data) | |||
kiss_frame = [KISS_FEND, kiss_cmd] + packet_escaped + [KISS_FEND] | |||
try: | |||
#print(bytearray(kiss_frame).hex()) | |||
output = bytearray(kiss_frame) | |||
except ValueError: | |||
logf("Invalid value in frame.") | |||
return None | |||
return output | |||
def decode_kiss_OE(frame): #From Kiss to LoRa, OE_Style | |||
if frame[0] != 0xC0 or frame[len(frame) - 1] != 0xC0: | |||
logf("Kiss Header not found, abort decoding of Frame: "+repr(frame)) | |||
return None | |||
frame=frame[2:len(frame) - 1] #cut kiss delimitator 0xc0 and command 0x00 | |||
src_addr,dest_addr,rpt_list,payload,dti=ax25parser(frame) #only for logging | |||
#build OE_style frame piece by piece | |||
result = src_addr.strip()+b">"+dest_addr.strip()+rpt_list+b":"+payload | |||
return result | |||
def decode_kiss_AX25(frame): #from kiss to LoRA, Standard AX25 | |||
result = b"" | |||
if frame[0] != 0xC0 or frame[len(frame) - 1] != 0xC0: | |||
logf("Kiss Header not found, abort decoding of Frame: "+repr(frame)) | |||
return None | |||
frame=frame[2:len(frame) - 1] #cut kiss delimitator 0xc0 and command 0x00 | |||
src_addr,dest_addr,rpt_list,payload,dti=ax25parser(frame) #only for logging | |||
return frame | |||
class SerialParser(): | |||
'''Simple parser for KISS frames. It handles multiple frames in one packet | |||
and calls the callback function on each frame''' | |||
STATE_IDLE = 0 | |||
STATE_FEND = 1 | |||
STATE_DATA = 2 | |||
KISS_FEND = KISS_FEND | |||
def __init__(self, frame_cb=None): | |||
self.frame_cb = frame_cb | |||
self.reset() | |||
def reset(self): | |||
self.state = self.STATE_IDLE | |||
self.cur_frame = bytearray() | |||
def parse(self, data): | |||
#Call parse with a string of one or more characters | |||
for c in data: | |||
if self.state == self.STATE_IDLE: | |||
if c == self.KISS_FEND: | |||
self.cur_frame.append(c) | |||
self.state = self.STATE_FEND | |||
elif self.state == self.STATE_FEND: | |||
if c == self.KISS_FEND: | |||
self.reset() | |||
else: | |||
self.cur_frame.append(c) | |||
self.state = self.STATE_DATA | |||
elif self.state == self.STATE_DATA: | |||
self.cur_frame.append(c) | |||
if c == self.KISS_FEND: | |||
# frame complete | |||
if self.frame_cb: | |||
self.frame_cb(self.cur_frame) | |||
self.reset() | |||
if __name__ == "__main__": | |||
# Playground for testing | |||
kissframe = b"\xc0\x00\x82\xa0\xa4\xa6@@`\x9e\x8ar\xa8\x96\x90p\x88\x92\x8e\x92@@f\x88\x92\x8e\x92@@e\x03\xf0!4725.51N/00939.86E[322/002/A=001306 Batt=3.99V | |||
\xc0" | |||
#test decode KISS->OE | |||
print(decode_kiss_OE(kissframe)) | |||
#test decode KISS->AX25 | |||
print(decode_kiss_AX25(kissframe)) | |||
#test encode OE->KISS | |||
OE_frame = b"OE9TKH-8>APRS,digi-3,digi-2:!4725.51N/00939.86E[322/002/A=001306 Batt=3.99V\n" | |||
signalreport="Level:-115dBm, SNR:0dB" | |||
print(encode_kiss_OE(OE_frame,signalreport)) | |||
#test encode AX25->KISS | |||
ax25_frame = b"\x82\xa0\xa4\xa6@@`\x9e\x8ar\xa8\x96\x90p\x88\x92\x8e\x92@@f\x88\x92\x8e\x92@@e\x03\xf0!4725.51N/00939.86E[322/002/A=001306 Batt=3.99V\n" | |||
print(encode_kiss_AX25(ax25_frame,signalreport)) | |||
</nowiki> | </nowiki> | ||
NOTE: The above KissHelper.py has not been optimized at all, it was just made to work with all KISS packets and made fault-tolerant so it doens't crash. The OE-based encoding can be removed, and many simplifications of the code can be made! This is the quick and dirty version. | |||
Start/test your TCP-based KISS TNC with this command (you should see it print out the config values and say it's listening for connections, with no errors listed): | Start/test your TCP-based KISS TNC with this command (you should see it print out the config values and say it's listening for connections, with no errors listed): |
Revision as of 23:26, 14 August 2023
KV4P has been experimenting with adding LoRa ports to his TARPN node. LoRa is very desirable for TARPN because LoRa transceivers are very inexpensive yet have good range, are very small and low-power, and reduce the overall setup cost of a new link.
Materials:
- Raspberry Pi Zero 2 W (~$15 when in stock), WITH gpio header. You can use any Rapsberry Pi model for this, but this is what I used.
- Adafruit LoRa Radio Bonnet with OLED - RFM95W @ 915MHz - RadioFruit ($32.50)
- 78mm of 16AWG magnet wire (33cm band 1/4 wave antenna), soldered to the radio bonnet antenna center conductor via (right next to the connector we won't use) (~$12 for a spool)
This assumes you already have a TARPN node. With this experimental guide, you'll be buiding a LoRa radio that shows up on your network as a TCP-based KISS TNC, which port 11 or port 12 of your TARPN node will connect to via the network.
I'm going to assume you already have the Raspberry Pi configured with Raspberry Pi OS, and connected to your local network (the same network as your TARPN node). You'll want to assign a static IP to it (rather than DHCP), so you can ensure your TARPN node will be able to find it on the network after restarts.
Get LoRa bonnet working
You can read more about these steps on this Adafruit page.
- sudo apt install python3-pip
- sudo pip3 install --upgrade setuptools
- sudo pip3 install --upgrade adafruit-python-shell
- wget https://github.com/adafruit/Adafruit_CircuitPython_framebuf/raw/main/examples/font5x8.bin
- wget https://raw.githubusercontent.com/adafruit/Raspberry-Pi-Installer-Scripts/master/raspi-blinka.py
- sudo python3 raspi-blinka.py (this will require reboot at the end)
- sudo pip3 install adafruit-circuitpython-ssd1306
- sudo pip3 install adafruit-circuitpython-framebuf
- sudo pip3 install adafruit-circuitpython-rfm9x
- create this test script rfm9x_check.py, which you should run ("python3 rfm9x_check.py") to prove your bonnet is properly connected and working:
# SPDX-FileCopyrightText: 2018 Brent Rubell for Adafruit Industries # # SPDX-License-Identifier: MIT """ Wiring Check, Pi Radio w/RFM9x Learn Guide: https://learn.adafruit.com/lora-and-lorawan-for-raspberry-pi Author: Brent Rubell for Adafruit Industries """ import time import busio from digitalio import DigitalInOut, Direction, Pull import board # Import the SSD1306 module. import adafruit_ssd1306 # Import the RFM9x radio module. import adafruit_rfm9x # Button A btnA = DigitalInOut(board.D5) btnA.direction = Direction.INPUT btnA.pull = Pull.UP # Button B btnB = DigitalInOut(board.D6) btnB.direction = Direction.INPUT btnB.pull = Pull.UP # Button C btnC = DigitalInOut(board.D12) btnC.direction = Direction.INPUT btnC.pull = Pull.UP # Create the I2C interface. i2c = busio.I2C(board.SCL, board.SDA) # 128x32 OLED Display reset_pin = DigitalInOut(board.D4) display = adafruit_ssd1306.SSD1306_I2C(128, 32, i2c, reset=reset_pin) # Clear the display. display.fill(0) display.show() width = display.width height = display.height # Configure RFM9x LoRa Radio CS = DigitalInOut(board.CE1) RESET = DigitalInOut(board.D25) spi = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO) while True: # Clear the image display.fill(0) # Attempt to set up the RFM9x Module try: rfm9x = adafruit_rfm9x.RFM9x(spi, CS, RESET, 915.0) display.text('RFM9x: Detected', 0, 0, 1) except RuntimeError as error: # Thrown on version mismatch display.text('RFM9x: ERROR', 0, 0, 1) print('RFM9x Error: ', error) # Check buttons if not btnA.value: # Button A Pressed display.text('Ada', width-85, height-7, 1) display.show() time.sleep(0.1) if not btnB.value: # Button B Pressed display.text('Fruit', width-75, height-7, 1) display.show() time.sleep(0.1) if not btnC.value: # Button C Pressed display.text('Radio', width-65, height-7, 1) display.show() time.sleep(0.1) display.show() time.sleep(0.1)
Turn it into a TCP KISS TNC
See the github project for more info.
- sudo apt install git aprx python3-rpi.gpio python3-spidev python3-pil python3-smbus
- git clone https://github.com/IZ7BOJ/RPi-LoRa-KISS-TNC-2ndgen.git
- cd RPi-LoRa-KISS-TNC-2ndgen
- git clone https://github.com/mayeranalytics/pySX127x.git
- sudo mv board_config.py pySX127x/SX127x/board_config.py
- sudo mv LoRa.py pySX127x/SX127x//LoRa.py
- Replace config.py with these contents. You can/should customize the frequency to anything in the 33cm band (but at least 500kHz from the band edge) and if you have poor link conditions you can increase "spreadingfactor" above 7, which is the fastest this bonnet can handle:
## Config file for RPi-LoRa-KISS-TNC 2nd generation ## Lora Module selection sx127x = True #if True, enables sx127x family, else sx126x ## Display enable and font disp_en = False font_size = 8 ## Log enable and path log_enable = True logpath='/var/log/lora/lora.log' #log filename. Give r/w permission! ## KISS Settings # Where to listen? # TCP_HOST can be "localhost", "0.0.0.0" or a specific interface address # TCP_PORT as configured in aprx.conf <interface> section TCP_HOST = "0.0.0.0" TCP_PORT = 10001 ## Hardware Settings # See datasheets for detailed pinout. # The default pin assignment refers to PCB designed by I8FUC. # The user can wire the module by his own and change pin assignment. # assign "-1" if the pin is not used # Settings valid for both SX126x and SX127x busId = 0 #SPI Bus ID. Must be enabled on raspberry (sudo raspi-config). Default is 0 csId = 1 #SPI Chip Select pin. Valid values are 0 or 1 irqPin = 22 #DIO0 of sx127x and DIO1 of sx126x, used for IRQ in rx # Settings valid only for SX126x. Default pin assignment refers to the PCB schematic /doc/LoRa_RPi_Companion_2022.pdf resetPin = 6 busyPin = 4 # If txen and rxen are disabled (=-1), then DIO2 will be set as RF Switch control txenPin = 0 #In Ebyte modules, it's used for switching on the tx pa. rxenPin = 1 #In Ebyte modules, it's used for switching on the rx lna # If Lora module has a TCXO, the following parameter must be True # If True, the DIO3 line will be set as control voltage of the TCXO tcxo=False ## LoRa Settings valid for both SX127x and SX126x modules frequency = 910300000 #frequency in Hz preamble = 32 #valid preable lenght is 8/16/24/32 spreadingFactor = 7 #valid spreading factor is between 5 and 12 bandwidth = 500000 #possible BW values: 7800, 10400,15600, 20800, 31250, 41700, 62500, 125000, 250000, 500000 codingrate = 5 #valid code rate denominator is between 5 and 8 appendSignalReport = False #append signal report when packets are forwarded to aprs server outputPower = 17 #maximum TX power is 22(22dBm) for SX126x, and 17 (17dBm) for SX127x . Higher values will be forced to max allowed! TX_OE_Style = False #if True, tx RF packets are in OE Style, otherwise in standard AX25 #sync_word = 0x1424 #sync word is x4y4. Es: 0x12 of 1st gen LoRa chip --> 0x1424 of 2nd gen LoRa chip sync_word = 0x12 crc = True #defines if CRC is calculated and transmitted in the header. Note that modem works in explicit mode #LoRa Settings valid only for SX126x RX_GAIN_POWER_SAVING = False #If false, receiver is set in boosted gain mode (needs more power)
This github project needs more modification to work with TARPN, since it was originally written only for APRS packets and will break with any other kind. It also assumed ~400MHz, whereas the Adafruit bonnet is 33cm (902-928MHz). Here is how to update it (TODO these should be in a forked github project to remove these steps):
- nano pySX127x/SX127x/board_config.py (Change low_band = True, to False for 33cm support)
- nano pySX127x/SX127x/LoRa.py (Search for “43” and change multiple occurences to 915MHz)
- nano LoraAprsKissTnc_sx127x.py (Ditto, there's only 1 occurence in this one)
- sudo mkdir /var/log/lora
- sudo touch /var/log/lora/lora.log
- Replace KissHelper.py with this contents (this extends it to handle all KISS packets not just APRS, and makes it a little more tolerant of unexpected failures rather than just stopping the process at the first weird packet):
#!/usr/bin/python3 # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. # This program provides basic KISS AX.25 APRS frame encoding and decoding. # Note that only APRS relevant structures are tested. It might not work # for generic AX.25 frames. # # Inspired by: # * Python script to decode AX.25 from KISS frames over a serial TNC # https://gist.github.com/mumrah/8fe7597edde50855211e27192cce9f88 # # * Sending a raw AX.25 frame with Python # https://thomask.sdf.org/blog/2018/12/15/sending-raw-ax25-python.html # # KISS-TNC for LoRa radio modem # https://github.com/IZ7BOJ/RPi-LoRa-KISS-TNC import struct import datetime import config KISS_FEND = 0xC0 # Frame start/end marker KISS_FESC = 0xDB # Escape character KISS_TFEND = 0xDC # If after an escape, means there was an 0xC0 in the source message KISS_TFESC = 0xDD # If after an escape, means there was an 0xDB in the source message # APRS data types DATA_TYPES_POSITION = b"!'/@`" DATA_TYPE_MESSAGE = b":" DATA_TYPE_THIRD_PARTY = b"}" def logf(message): timestamp = datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S - ') if config.log_enable: fileLog = open(config.logpath,"a") fileLog.write(timestamp + message+"\n") fileLog.close() print(timestamp + message) # from LoRa OE_Style to KISS # Addresses must be 6 bytes plus the SSID byte, each character shifted left by 1 # If it's the final address in the header, set the low bit to 1 # If it has been digipeated, set the H bit to 1 # Ignoring command/response for simple example def encode_address(s, final): H=0b00000000 #default H bit low if chr(s[-1])=='*': #example: WIDE1-1* or WIDE1* s=s[:-1] H=0b10000000 if b"-" not in s: s = s + b"-0" # default to SSID 0 call, ssid = s.split(b'-') if len(call) < 6: call = call + b" "*(6 - len(call)) # pad with spaces encoded_call = [x << 1 for x in call[0:6]] encoded_ssid = (int(ssid) << 1) | H | 0b01100000 | (0b00000001 if final else 0) return encoded_call + [encoded_ssid] def decode_address(data, cursor): (a1, a2, a3, a4, a5, a6, a7) = struct.unpack("<BBBBBBB", data[cursor:cursor + 7]) hrr = a7 >> 5 ssid = (a7 >> 1) & 0xf ext = a7 & 0x1 addr = struct.pack("<BBBBBB", a1 >> 1, a2 >> 1, a3 >> 1, a4 >> 1, a5 >> 1, a6 >> 1) if ssid != 0: call = addr.strip() + "-{}".format(ssid).encode() else: call = addr return (call, hrr, ext) def ax25parser(frame): #extracts fields from ax25 frames and add signal report do payload, if specified try: pos = 0 # DST (dest_addr, dest_hrr, dest_ext) = decode_address(frame, pos) pos += 7 print("DST: ", dest_addr) # SRC (src_addr, src_hrr, src_ext) = decode_address(frame, pos) pos += 7 print("SRC: ", src_addr) # REPEATERS ext = src_ext rpt_list = b"" while ext == 0: rpt_addr, rpt_hrr, ext = decode_address(frame, pos) if rpt_hrr==b'111': # H bit high-->packet has been digipeated rpt_addr+=b'*' rpt_list += b","+rpt_addr print("RPT: ", rpt_addr) pos += 7 # CTRL ctrl = frame[pos] pos += 1 if (ctrl & 0x3) == 0x3: #(pid,) = struct.unpack("<B", frame[pos]) try: pid = frame[pos] except Exception: pid = 240 print("PID: "+str(pid)) pos += 1 elif (ctrl & 0x3) == 0x1: # decode_sframe(ctrl, frame, pos) logf("SFRAME") return None,None,None,None,None elif (ctrl & 0x1) == 0x0: # decode_iframe(ctrl, frame, pos) logf("IFRAME") return None,None,None,None,None payload=frame[pos:] print("payload: ", payload) try: dti=payload[0] except Exception: dti=":" logf("Extracted AX25 parameters from AX25 Frame") logf("From: "+repr(src_addr)[2:-1]+" To: "+repr(dest_addr)[2:-1]+" Via: "+repr(rpt_list)[3:-1]+" PID: "+str(hex(pid))+" Payload: "+str(payload)[str(payl oad).find("'"):-1]) return src_addr,dest_addr,rpt_list,payload,dti except Exception: return None,None,None,None,None def encode_kiss_AX25(frame,signalreport): #from Lora to Kiss, Standard AX25 src_addr,dest_addr,rpt_list,payload,dti=ax25parser(frame) #only for logging # Escape the packet in case either KISS_FEND or KISS_FESC ended up in our stream if config.appendSignalReport and str(dti) != DATA_TYPE_MESSAGE: frame += b" "+str.encode(signalreport,'utf-8') packet_escaped = [] for x in frame: if x == KISS_FEND: packet_escaped += [KISS_FESC, KISS_TFEND] elif x == KISS_FESC: packet_escaped += [KISS_FESC, KISS_TFESC] else: packet_escaped += [x] # Build the frame that we will send to aprx and turn it into a string kiss_cmd = 0x00 # Two nybbles combined - TNC 0, command 0 (send data) kiss_frame = [KISS_FEND, kiss_cmd] + packet_escaped + [KISS_FEND] try: output = bytearray(kiss_frame) except ValueError: logf("Invalid value in frame.") return None return output def encode_kiss_OE(frame,signalreport): #from Lora to Kiss, OE_Style # Ugly frame disassembling if not b":" in frame: logf("Can't decode OE LoRa Frame") return None path = frame.split(b":")[0] payload = frame[frame.find(b":")+1:] dti = payload[0] src_addr = path.split(b">")[0] digis = path[path.find(b">") + 1:].split(b",") dest_addr = digis.pop(0) # destination address packet = encode_address(dest_addr.upper(), False) # source address packet += encode_address(src_addr.upper(), len(digis) == 0) # digipeaters for digi in digis: final_addr = digis.index(digi) == len(digis) - 1 packet += encode_address(digi.upper(), final_addr) # control field packet += [0x03] # This is an UI frame # protocol ID packet += [0xF0] # No protocol # information field logf("Extracted AX25 parameters from OE LoRa Frame") logf("From: "+repr(src_addr)[2:-1]+" To: "+repr(dest_addr)[2:-1]+" Via: "+str(digis)[1:-1].replace("b","").replace("'","")+" Payload: "+repr(payload)[2:-1]) packet += payload if config.appendSignalReport and str(dti) != DATA_TYPE_MESSAGE: #some SW (es OE5BPA) append newline character at the end of packet. Must be cut for appending signal report if chr(packet[-1])=="\n": packet=packet[:-1] packet += b" "+str.encode(signalreport,'utf-8') # Escape the packet in case either KISS_FEND or KISS_FESC ended up in our stream packet_escaped = [] for x in packet: if x == KISS_FEND: packet_escaped += [KISS_FESC, KISS_TFEND] elif x == KISS_FESC: packet_escaped += [KISS_FESC, KISS_TFESC] else: packet_escaped += [x] # Build the frame that we will send to Dire Wolf and turn it into a string kiss_cmd = 0x00 # Two nybbles combined - TNC 0, command 0 (send data) kiss_frame = [KISS_FEND, kiss_cmd] + packet_escaped + [KISS_FEND] try: #print(bytearray(kiss_frame).hex()) output = bytearray(kiss_frame) except ValueError: logf("Invalid value in frame.") return None return output def decode_kiss_OE(frame): #From Kiss to LoRa, OE_Style if frame[0] != 0xC0 or frame[len(frame) - 1] != 0xC0: logf("Kiss Header not found, abort decoding of Frame: "+repr(frame)) return None frame=frame[2:len(frame) - 1] #cut kiss delimitator 0xc0 and command 0x00 src_addr,dest_addr,rpt_list,payload,dti=ax25parser(frame) #only for logging #build OE_style frame piece by piece result = src_addr.strip()+b">"+dest_addr.strip()+rpt_list+b":"+payload return result def decode_kiss_AX25(frame): #from kiss to LoRA, Standard AX25 result = b"" if frame[0] != 0xC0 or frame[len(frame) - 1] != 0xC0: logf("Kiss Header not found, abort decoding of Frame: "+repr(frame)) return None frame=frame[2:len(frame) - 1] #cut kiss delimitator 0xc0 and command 0x00 src_addr,dest_addr,rpt_list,payload,dti=ax25parser(frame) #only for logging return frame class SerialParser(): '''Simple parser for KISS frames. It handles multiple frames in one packet and calls the callback function on each frame''' STATE_IDLE = 0 STATE_FEND = 1 STATE_DATA = 2 KISS_FEND = KISS_FEND def __init__(self, frame_cb=None): self.frame_cb = frame_cb self.reset() def reset(self): self.state = self.STATE_IDLE self.cur_frame = bytearray() def parse(self, data): #Call parse with a string of one or more characters for c in data: if self.state == self.STATE_IDLE: if c == self.KISS_FEND: self.cur_frame.append(c) self.state = self.STATE_FEND elif self.state == self.STATE_FEND: if c == self.KISS_FEND: self.reset() else: self.cur_frame.append(c) self.state = self.STATE_DATA elif self.state == self.STATE_DATA: self.cur_frame.append(c) if c == self.KISS_FEND: # frame complete if self.frame_cb: self.frame_cb(self.cur_frame) self.reset() if __name__ == "__main__": # Playground for testing kissframe = b"\xc0\x00\x82\xa0\xa4\xa6@@`\x9e\x8ar\xa8\x96\x90p\x88\x92\x8e\x92@@f\x88\x92\x8e\x92@@e\x03\xf0!4725.51N/00939.86E[322/002/A=001306 Batt=3.99V \xc0" #test decode KISS->OE print(decode_kiss_OE(kissframe)) #test decode KISS->AX25 print(decode_kiss_AX25(kissframe)) #test encode OE->KISS OE_frame = b"OE9TKH-8>APRS,digi-3,digi-2:!4725.51N/00939.86E[322/002/A=001306 Batt=3.99V\n" signalreport="Level:-115dBm, SNR:0dB" print(encode_kiss_OE(OE_frame,signalreport)) #test encode AX25->KISS ax25_frame = b"\x82\xa0\xa4\xa6@@`\x9e\x8ar\xa8\x96\x90p\x88\x92\x8e\x92@@f\x88\x92\x8e\x92@@e\x03\xf0!4725.51N/00939.86E[322/002/A=001306 Batt=3.99V\n" print(encode_kiss_AX25(ax25_frame,signalreport))
NOTE: The above KissHelper.py has not been optimized at all, it was just made to work with all KISS packets and made fault-tolerant so it doens't crash. The OE-based encoding can be removed, and many simplifications of the code can be made! This is the quick and dirty version.
Start/test your TCP-based KISS TNC with this command (you should see it print out the config values and say it's listening for connections, with no errors listed):
- sudo python3 Start_lora-tnc.py
When you run the TNC for long-term use, you'll want to run it with "nohup sudo python3 Start_lora-tnc.py &" which will disassociate it with your linux account and run it in the background until manually terminated. In the future, we should wrap this in a systemd service.
TARPN configuration
On your TARPN node, you won't be able to run it with "tarpn service start" until TARPN is updated to allow TCP-based KISS TNCs on port 11 and 12. For now, you can force this by "tarpn service stop", then editing bpq/bpq32.cfg.
- nano bpq/bpq32.cfg
- Search for "PORTNUM=11" to find the right section
- Paste this in:
PORT PORTNUM=11 ; Optional but sets port number if stated ID=LoRa link to KV4P-3 ; Displayed by PORTS command TYPE=ASYNC ; Port is RS232 Com PROTOCOL=KISS ; TNC is used in KISS (or JKISS) mode ;; See ..\RelatedFiles\KissRoms\KissRoms.zip. FULLDUP=0 ; Only meaningful for KISS (or JKISS) devices ;COMPORT=/dev/ttyUSB0 ;SPEED=57600 ; RS232 COM PORT SPEED CHANNEL=A ; A for single channel TNC, A or B for multichannel PERSIST=180 ; PERSIST=256/(# of transmitters-1) SLOTTIME=50 ; CMSA interval timer in milliseconds TXDELAY=50 ; Transmit keyup delay in milliseconds TXTAIL=1 ; TX key down, in milliseconds, at packet end QUALITY=1 ; Quality factor applied to node broadcasts heard on ; ; this port, unless overridden by a locked route ; ; entry. Setting to 0 stops node broadcasts MINQUAL=81 ; Entries in the nodes table with qualities greater or ; ; equal to MINQUAL will be sent on this port. A value ; ; of 0 sends everything. MAXFRAME=1 ; Max outstanding frames (1 thru 7) FRACK=2000 ; Level 2 timout in milliseconds RESPTIME=40 ; Level 2 delayed ack timer in milliseconds RETRIES=20 ; Level 2 maximum retry value PACLEN=136 ; Default max packet length for this port UNPROTO=ID ; BTEXT broadcast addrs format: DEST[,digi1[,digi2]] L3ONLY=0 ; 1=No user downlink connects on this port DIGIFLAG=0 ; Digipeat: 0=OFF, 1=ALL, 255=UI Only DIGIPORT=0 ; Port on which to send digi'd frames (0 = same port) USERS=0 ; Maximum number of L2 sessions, 0 = no limit IGNOREUNLOCKEDROUTES=0 ; ignore node broadcasts from other than locked routes IPADDR=127.0.0.1 ; Address of PC runnning the TNC. Usually this is the same machine as BPQ, so use 127.0.0.1 but can be on anoth> TCPPORT=10001 ; Port TNC listens on. Default for UZ7HO is 8100 and Direwolf 8001 KISSOPTIONS=NOPARAMS ENDPORT
You also need to add a fixed route for port 11 or 12 in bpq32.cfg. The appropriate section looks like this, and has commented-out port 11 and 12 routes already listed:
ROUTES: ; Locked routes (31 maximum) KV4P-3,200,11
Of course, replace "KV4P-3" with whatever neighbor you plan to link to with that LoRa link.
Start TARPN with ./linbpq (instead of "tarpn service start"). You won't be able to use TARPN HOME until TARPN is updated to handle TCP-based KISS TNCs natively. But you can manually connect to your node (e.g. with QtTerm), and get into chat that way, and test out your new LoRa link, etc.
Performance
With the above config.py settings, a throughput test on my own TARPN node with a "bench setup" (2 LoRa transceivers right next to each other), I achieved 135bytes/sec. That's about as good as a very solid 9600 baud link with a NinoTNC and high quality mobile radios.
It can probably be made to go much faster with additional optimization. It's unclear if the KISS service running in Python represents a cap on the throughput, or if it's just the settings.
Although the distance of this link hasn't been tested yet, the LoRa module claims to work between 4km and 40km (from city to perfectly flat ideal conditions). 33cm has the nice property of easily going through windows and walls, so it should be especially good for suburban links, or between nodes with a fairly clear shot.