2020-04-22 10:07:13 +00:00
from . Interface import Interface
2018-04-04 12:14:22 +00:00
from time import sleep
import sys
import serial
import threading
import time
import RNS
class KISS ( ) :
2020-08-13 10:15:56 +00:00
FEND = 0xC0
FESC = 0xDB
TFEND = 0xDC
TFESC = 0xDD
CMD_UNKNOWN = 0xFE
CMD_DATA = 0x00
CMD_TXDELAY = 0x01
CMD_P = 0x02
CMD_SLOTTIME = 0x03
CMD_TXTAIL = 0x04
CMD_FULLDUPLEX = 0x05
CMD_SETHARDWARE = 0x06
CMD_READY = 0x0F
CMD_RETURN = 0xFF
@staticmethod
def escape ( data ) :
data = data . replace ( bytes ( [ 0xdb ] ) , bytes ( [ 0xdb , 0xdd ] ) )
data = data . replace ( bytes ( [ 0xc0 ] ) , bytes ( [ 0xdb , 0xdc ] ) )
return data
2020-05-03 21:06:53 +00:00
2018-04-08 10:26:57 +00:00
class AX25 ( ) :
2020-08-13 10:15:56 +00:00
PID_NOLAYER3 = 0xF0
CTRL_UI = 0x03
CRC_CORRECT = bytes ( [ 0xF0 ] ) + bytes ( [ 0xB8 ] )
HEADER_SIZE = 16
2018-04-08 10:26:57 +00:00
class AX25KISSInterface ( Interface ) :
2020-08-13 10:15:56 +00:00
MAX_CHUNK = 32768
owner = None
port = None
speed = None
databits = None
parity = None
stopbits = None
serial = None
def __init__ ( self , owner , name , callsign , ssid , port , speed , databits , parity , stopbits , preamble , txtail , persistence , slottime , flow_control ) :
2021-09-24 18:10:04 +00:00
self . rxb = 0
self . txb = 0
2020-08-13 10:15:56 +00:00
self . serial = None
self . owner = owner
self . name = name
self . src_call = callsign . upper ( ) . encode ( " ascii " )
self . src_ssid = ssid
self . dst_call = " APZRNS " . encode ( " ascii " )
self . dst_ssid = 0
self . port = port
self . speed = speed
self . databits = databits
self . parity = serial . PARITY_NONE
self . stopbits = stopbits
self . timeout = 100
self . online = False
self . packet_queue = [ ]
self . flow_control = flow_control
self . interface_ready = False
2021-09-03 08:56:49 +00:00
self . flow_control_timeout = 5
self . flow_control_locked = time . time ( )
2020-08-13 10:15:56 +00:00
if ( len ( self . src_call ) < 3 or len ( self . src_call ) > 6 ) :
raise ValueError ( " Invalid callsign for " + str ( self ) )
if ( self . src_ssid < 0 or self . src_ssid > 15 ) :
raise ValueError ( " Invalid SSID for " + str ( self ) )
self . preamble = preamble if preamble != None else 350 ;
self . txtail = txtail if txtail != None else 20 ;
self . persistence = persistence if persistence != None else 64 ;
self . slottime = slottime if slottime != None else 20 ;
if parity . lower ( ) == " e " or parity . lower ( ) == " even " :
self . parity = serial . PARITY_EVEN
if parity . lower ( ) == " o " or parity . lower ( ) == " odd " :
self . parity = serial . PARITY_ODD
try :
RNS . log ( " Opening serial port " + self . port + " ... " )
self . serial = serial . Serial (
port = self . port ,
baudrate = self . speed ,
bytesize = self . databits ,
parity = self . parity ,
stopbits = self . stopbits ,
xonxoff = False ,
rtscts = False ,
timeout = 0 ,
inter_byte_timeout = None ,
write_timeout = None ,
dsrdtr = False ,
)
except Exception as e :
RNS . log ( " Could not open serial port for interface " + str ( self ) , RNS . LOG_ERROR )
raise e
if self . serial . is_open :
# Allow time for interface to initialise before config
sleep ( 2.0 )
thread = threading . Thread ( target = self . readLoop )
thread . setDaemon ( True )
thread . start ( )
self . online = True
RNS . log ( " Serial port " + self . port + " is now open " )
RNS . log ( " Configuring AX.25 KISS interface parameters... " )
self . setPreamble ( self . preamble )
self . setTxTail ( self . txtail )
self . setPersistence ( self . persistence )
self . setSlotTime ( self . slottime )
self . setFlowControl ( self . flow_control )
self . interface_ready = True
RNS . log ( " AX.25 KISS interface configured " )
sleep ( 2 )
else :
raise IOError ( " Could not open serial port " )
def setPreamble ( self , preamble ) :
preamble_ms = preamble
preamble = int ( preamble_ms / 10 )
if preamble < 0 :
preamble = 0
if preamble > 255 :
preamble = 255
kiss_command = bytes ( [ KISS . FEND ] ) + bytes ( [ KISS . CMD_TXDELAY ] ) + bytes ( [ preamble ] ) + bytes ( [ KISS . FEND ] )
written = self . serial . write ( kiss_command )
if written != len ( kiss_command ) :
raise IOError ( " Could not configure AX.25 KISS interface preamble to " + str ( preamble_ms ) + " (command value " + str ( preamble ) + " ) " )
def setTxTail ( self , txtail ) :
txtail_ms = txtail
txtail = int ( txtail_ms / 10 )
if txtail < 0 :
txtail = 0
if txtail > 255 :
txtail = 255
kiss_command = bytes ( [ KISS . FEND ] ) + bytes ( [ KISS . CMD_TXTAIL ] ) + bytes ( [ txtail ] ) + bytes ( [ KISS . FEND ] )
written = self . serial . write ( kiss_command )
if written != len ( kiss_command ) :
raise IOError ( " Could not configure AX.25 KISS interface TX tail to " + str ( txtail_ms ) + " (command value " + str ( txtail ) + " ) " )
def setPersistence ( self , persistence ) :
if persistence < 0 :
persistence = 0
if persistence > 255 :
persistence = 255
kiss_command = bytes ( [ KISS . FEND ] ) + bytes ( [ KISS . CMD_P ] ) + bytes ( [ persistence ] ) + bytes ( [ KISS . FEND ] )
written = self . serial . write ( kiss_command )
if written != len ( kiss_command ) :
raise IOError ( " Could not configure AX.25 KISS interface persistence to " + str ( persistence ) )
def setSlotTime ( self , slottime ) :
slottime_ms = slottime
slottime = int ( slottime_ms / 10 )
if slottime < 0 :
slottime = 0
if slottime > 255 :
slottime = 255
kiss_command = bytes ( [ KISS . FEND ] ) + bytes ( [ KISS . CMD_SLOTTIME ] ) + bytes ( [ slottime ] ) + bytes ( [ KISS . FEND ] )
written = self . serial . write ( kiss_command )
if written != len ( kiss_command ) :
raise IOError ( " Could not configure AX.25 KISS interface slot time to " + str ( slottime_ms ) + " (command value " + str ( slottime ) + " ) " )
def setFlowControl ( self , flow_control ) :
kiss_command = bytes ( [ KISS . FEND ] ) + bytes ( [ KISS . CMD_READY ] ) + bytes ( [ 0x01 ] ) + bytes ( [ KISS . FEND ] )
written = self . serial . write ( kiss_command )
if written != len ( kiss_command ) :
if ( flow_control ) :
raise IOError ( " Could not enable AX.25 KISS interface flow control " )
else :
raise IOError ( " Could not enable AX.25 KISS interface flow control " )
def processIncoming ( self , data ) :
if ( len ( data ) > AX25 . HEADER_SIZE ) :
2021-09-24 18:10:04 +00:00
self . rxb + = len ( data )
2020-08-13 10:15:56 +00:00
self . owner . inbound ( data [ AX25 . HEADER_SIZE : ] , self )
def processOutgoing ( self , data ) :
2021-09-24 18:10:04 +00:00
datalen = len ( data )
2020-08-13 10:15:56 +00:00
if self . online :
if self . interface_ready :
if self . flow_control :
self . interface_ready = False
2021-09-03 08:56:49 +00:00
self . flow_control_locked = time . time ( )
2020-08-13 10:15:56 +00:00
encoded_dst_ssid = bytes ( [ 0x60 | ( self . dst_ssid << 1 ) ] )
encoded_src_ssid = bytes ( [ 0x60 | ( self . src_ssid << 1 ) | 0x01 ] )
addr = b " "
for i in range ( 0 , 6 ) :
if ( i < len ( self . dst_call ) ) :
addr + = bytes ( [ self . dst_call [ i ] << 1 ] )
else :
addr + = bytes ( [ 0x20 ] )
addr + = encoded_dst_ssid
for i in range ( 0 , 6 ) :
if ( i < len ( self . src_call ) ) :
addr + = bytes ( [ self . src_call [ i ] << 1 ] )
else :
addr + = bytes ( [ 0x20 ] )
addr + = encoded_src_ssid
data = addr + bytes ( [ AX25 . CTRL_UI ] ) + bytes ( [ AX25 . PID_NOLAYER3 ] ) + data
data = data . replace ( bytes ( [ 0xdb ] ) , bytes ( [ 0xdb ] ) + bytes ( [ 0xdd ] ) )
data = data . replace ( bytes ( [ 0xc0 ] ) , bytes ( [ 0xdb ] ) + bytes ( [ 0xdc ] ) )
kiss_frame = bytes ( [ KISS . FEND ] ) + bytes ( [ 0x00 ] ) + data + bytes ( [ KISS . FEND ] )
written = self . serial . write ( kiss_frame )
2021-09-24 18:10:04 +00:00
self . txb + = datalen
2020-08-13 10:15:56 +00:00
if written != len ( kiss_frame ) :
if self . flow_control :
self . interface_ready = True
raise IOError ( " AX.25 interface only wrote " + str ( written ) + " bytes of " + str ( len ( kiss_frame ) ) )
else :
self . queue ( data )
def queue ( self , data ) :
self . packet_queue . append ( data )
def process_queue ( self ) :
if len ( self . packet_queue ) > 0 :
data = self . packet_queue . pop ( 0 )
self . interface_ready = True
self . processOutgoing ( data )
elif len ( self . packet_queue ) == 0 :
self . interface_ready = True
def readLoop ( self ) :
try :
in_frame = False
escape = False
command = KISS . CMD_UNKNOWN
data_buffer = b " "
last_read_ms = int ( time . time ( ) * 1000 )
while self . serial . is_open :
if self . serial . in_waiting :
byte = ord ( self . serial . read ( 1 ) )
last_read_ms = int ( time . time ( ) * 1000 )
if ( in_frame and byte == KISS . FEND and command == KISS . CMD_DATA ) :
in_frame = False
self . processIncoming ( data_buffer )
elif ( byte == KISS . FEND ) :
in_frame = True
command = KISS . CMD_UNKNOWN
data_buffer = b " "
elif ( in_frame and len ( data_buffer ) < RNS . Reticulum . MTU + AX25 . HEADER_SIZE ) :
if ( len ( data_buffer ) == 0 and command == KISS . CMD_UNKNOWN ) :
# We only support one HDLC port for now, so
# strip off the port nibble
byte = byte & 0x0F
command = byte
elif ( command == KISS . CMD_DATA ) :
if ( byte == KISS . FESC ) :
escape = True
else :
if ( escape ) :
if ( byte == KISS . TFEND ) :
byte = KISS . FEND
if ( byte == KISS . TFESC ) :
byte = KISS . FESC
escape = False
data_buffer = data_buffer + bytes ( [ byte ] )
elif ( command == KISS . CMD_READY ) :
# TODO: add timeout and reset if ready
# command never arrives
self . process_queue ( )
else :
time_since_last = int ( time . time ( ) * 1000 ) - last_read_ms
if len ( data_buffer ) > 0 and time_since_last > self . timeout :
data_buffer = b " "
in_frame = False
command = KISS . CMD_UNKNOWN
escape = False
2021-09-03 08:56:49 +00:00
sleep ( 0.05 )
if self . flow_control :
if not self . interface_ready :
if time . time ( ) > self . flow_control_locked + self . flow_control_timeout :
RNS . log ( " Interface " + str ( self ) + " is unlocking flow control due to time-out. This should not happen. Your hardware might have missed a flow-control READY command, or maybe it does not support flow-control. " , RNS . LOG_WARNING )
self . process_queue ( )
2020-08-13 10:15:56 +00:00
except Exception as e :
self . online = False
RNS . log ( " A serial port error occurred, the contained exception was: " + str ( e ) , RNS . LOG_ERROR )
2021-09-18 20:49:04 +00:00
RNS . log ( " The interface " + str ( self ) + " experienced an unrecoverable error and is being torn down. Restart Reticulum to attempt to open this interface again. " , RNS . LOG_ERROR )
if RNS . Reticulum . panic_on_interface_error :
RNS . panic ( )
2020-08-13 10:15:56 +00:00
def __str__ ( self ) :
return " AX25KISSInterface[ " + self . name + " ] "