GREEN, RED = 2, 160 # Unix terminal colors

@staticmethod

def fg(text, color):

'''

if sys.platform.startswith('win'):

return text

else:

return "\33[38;5;" + str(color) + "m" + text + "\33[0m"

@staticmethod

def green(text):

return Terminal.fg(text, Terminal.GREEN)

@staticmethod

def red(text):

SYNC = 0xc8

# CRSF Device Addresses

BROADCAST_ADDR = 0x00

CLOUD_ADDR = 0x0E # MQTT server

WIFI_ADDR = 0x12

REMOTE_ADDR = 0xEA

RX_ADDR = 0xEC

TX_ADDR = 0xEE

FC_ADDR = 0xC8 # flight controller

VTX_ADDR = 0xCE

# CRSF Frame Types

MSG_TYPE_GPS = 0x02

MSG_TYPE_GPST = 0x03

MSG_TYPE_BATT = 0x08

MSG_TYPE_VTX_TEL = 0x10

MSG_TYPE_LINK_STATS = 0x14

MSG_TYPE_PPM = 0x16 # channel values

MSG_TYPE_PPM3 = 0x17 # CRSF V3 (packed channel values)

MSG_TYPE_LINK_STATS_RX = 0x1C # CRSF V3

MSG_TYPE_LINK_STATS_TX = 0x1D # CRSF V3

MSG_TYPE_ATTD = 0x1E

MSG_TYPE_MADD = 0x1F

MSG_TYPE_PING = 0x28

MSG_TYPE_DEVICE_INFO = 0x29

MSG_TYPE_PARAM_ENTRY = 0x2B

MSG_TYPE_PARAM_READ = 0x2C

MSG_TYPE_PARAM_WRITE = 0x2D

MSG_TYPE_CMD = 0x32

MSG_TYPE_LOG = 0x34

MSG_TYPE_REMOTE = 0x3A # Remote-related frames

MSG_TYPE_MAVLINK_ENV = 0xAA

# CRSF menu parameter types

PARAM_TYPE_UINT8 = 0

PARAM_TYPE_INT8 = 1

PARAM_TYPE_UINT16 = 2

PARAM_TYPE_INT16 = 3

PARAM_TYPE_UINT32 = 4

PARAM_TYPE_INT32 = 5

PARAM_TYPE_FLOAT = 8

PARAM_TYPE_TEXT_SELECTION = 9

PARAM_TYPE_STRING = 10

PARAM_TYPE_FOLDER = 11

PARAM_TYPE_INFO = 12

PARAM_TYPE_COMMAND = 13

PARAM_TYPE_OUT_OF_RANGE = 127

# CRSF frame structure (field offsets)

OFFSET_LENGTH = 1

OFFSET_MSG_TYPE = 2

# Menu

MENU_COMMAND_READY = 0 # --> feedback

MENU_COMMAND_START = 1 # <-- input

MENU_COMMAND_PROGRESS = 2 # --> feedback

MENU_COMMAND_CONFIRMATION_NEEDED = 3 # --> feedback

MENU_COMMAND_CONFIRM = 4 # <-- input

MENU_COMMAND_CANCEL = 5 # <-- input

MENU_COMMAND_POLL = 6 # <-- input

@staticmethod

def decode_int32(data):

uint32 = (data[0] << 24) | (data[1] << 16) | (data[2] << 8) | data[3]

if uint32 & 0x80000000:

return -(~uint32) - 1

else:

return uint32

@staticmethod

def encode_int32(val):

return [

(val >> 24) & 0xFF,

(val >> 16) & 0xFF,

(val >> 8) & 0xFF,

val & 0xFF,

RESET_VALUE = 0x00

POLYNOM_1 = 0xD5 # CRC8 DVB-S2 (polynomial used for validating CRSF frames)

POLYNOM_2 = 0xBA # custom polynomial used for validating CRSF commands

MSB_SET = 0x80

FINISH_VALUE = 0x00

def __init__(self, poly = None):

self.val = self.RESET_VALUE

self.poly = poly if poly is not None else self.POLYNOM_1

def _calc(self, byte):

for i in range(8):

msb_flag = self.val & self.MSB_SET

self.val <<= 1

if byte & self.MSB_SET:

self.val += 1

byte <<= 1

if msb_flag:

self.val ^= self.poly

self.val &= 0xFF

def digest(self, data):

for x in data:

self._calc(x)

def finish(self):

self._calc(self.FINISH_VALUE)

'''Calculate CRC8 as in CRSF frames'''

crc = crsf_crc8(poly = crsf_crc8.POLYNOM_1)

crc.digest(data)

def __init__(self, data):

self.data = bytes(data)

@property

def len(self):

return self.data[1] + 2

@property

def type(self):

return self.data[2]

@property

def bytes(self):

return self.data

@property

def is_extended(self):

'''Extended CRSF frames include their ORIGIN and DESTINATION'''

return self.type >= 0x28 and self.type <= 0x96 or self.type == CRSF.MSG_TYPE_MAVLINK_ENV

@property

def origin(self):

return self.data[4] if self.is_extended else None

@property

def destination(self):

return self.data[3] if self.is_extended else None

@property

def payload(self):

return self.data[5 if self.is_extended else 3:-1]

def parse(self):

if self.type == CRSF.MSG_TYPE_DEVICE_INFO:

delim = self.payload.index(0x00)

device_name, tail = self.payload[:delim], self.payload[delim+1:]

device_name = bytes(device_name).decode()

sn = (tail[0] << 24) | (tail[1] << 16) | (tail[2] << 8) | tail[3]

hw_id = (tail[4] << 24) | (tail[5] << 16) | (tail[6] << 8) | tail[7]

sw_id = (tail[8] << 24) | (tail[9] << 16) | (tail[10] << 8) | tail[11]

param_count = tail[12]

version = tail[13]

return device_name, sn, hw_id, sw_id, param_count, version

else:

raise ValueError("cannot parse frame of type 0x%02x" % self.type)

def __str__(self):

s = 'SYNC ' if self.bytes[0] == CRSF.SYNC else ('%02x ' % self.bytes[0])

s += 'L=%d ' % self.len

s += ('(%s) ' % msg_name[self.type]) if self.type in msg_name else ('(t=%02x) ' % self.type)

i = 3

if self.is_extended:

s += '%s->%s ' % (dev_name.get(self.bytes[i+1], '%02x' % self.bytes[i+1]),

dev_name.get(self.bytes[i ], '%02x' % self.bytes[i ]))

i += 2

s += ' '.join(map(lambda x: "%02x" % x, self.bytes[i:]))

def __init__(self, silent):

self.data = bytearray()

self.silent = silent

def digest(self, data):

'''Digests incoming bytes, yields complete CRSF frames'''

self.data += bytearray(data)

#print(len(self.data))

while len(self.data) >= 4:

while self.data and self.data[0] != CRSF.SYNC:

if not self.silent:

sys.stderr.write("byte %02x discarded\n" % self.data[0])

self.data = self.data[1:]

if len(self.data) > 1:

frame_len = self.data[1] + 2

if len(self.data) >= frame_len:

crc_byte = self.data[frame_len - 1]

calc_crc = calc_crc8(self.data[2:frame_len - 1])

if crc_byte == calc_crc:

frame, self.data = self.data[:frame_len], self.data[frame_len:]

yield crsf_frame(frame)

else:

if not self.silent:

sys.stderr.write("crc mismatch; byte discarded\n")

self.data = self.data[1:]

else:

break

else:

'''Decode Run-Length Encoded number'''

off = 0

res = 0

ch = 0x80

while (ch & 0x80) and off < 6:

ch = data[off]

tmp = ch & 0x7F

res |= (tmp << (7*off))

off += 1

'''Decode debug message for XLOG frame type'''

if (data[0] & 0x80) == 0:

msg = (bytes([data[0] & 0x7F]).decode(), 1)

else:

i = data.find(0)

data = bytearray(data)

data[0] &= 0x7F

msg = (bytes(data[:i]).decode(), i+1)

data = ''.join(map(bin_byte, reversed(data)))

assert(len(data) % 11 == 0)

ticks = [int(data[x:x+11], 2) for x in range(len(data)-11, 0, -11)]

global up_lqi_history, down_lqi_history

cur_time = time.time()

up_lqi_history.append((cur_time, data[2]))

down_lqi_history.append((cur_time, data[8]))

up_lqi_history = up_lqi_history[-HIST_SIZE:]

down_lqi_history = down_lqi_history[-HIST_SIZE:]

s = 'Uplink: RSSI={}/{}'.format(-data[0], -data[1])

s += ', LQI={:3d}%, SNR={}, Ant.={}'.format(data[2], data[3], data[4])

s += ', RFmode={}, RFpwr={}'.format(data[5], data[6])

s += '; Downlink: RSSI={}, LQI={:3d}%, SNR={}'.format(-data[7], data[8], data[9])

# Show average LQI (both uplink and downlink)

if len(up_lqi_history) > 1 and len(down_lqi_history) > 1:

def get_hist_lqi(lqi_hist):

avg = sum([x[1] for x in lqi_hist])/len(lqi_hist)

if colorize:

green, red = Terminal.green, Terminal.red

else:

green = red = lambda x: x

return '{}/{:.1f}'.format(

green('100') if avg == 100 else red('{:.2f}'.format(avg)),

(lqi_hist[-1][0] - lqi_hist[0][0])

)

short_up = get_hist_lqi(up_lqi_history[-SHORT_HIST_SIZE:])

long_up = get_hist_lqi(up_lqi_history[-HIST_SIZE:])

short_down = get_hist_lqi(down_lqi_history[-SHORT_HIST_SIZE:])

long_down = get_hist_lqi(down_lqi_history[-HIST_SIZE:])

s += '\n History: Uplink LQI={}, {}; Downlink LQI={}, {}'.format(short_up, long_up, short_down, long_down)

'''

global last_time, ppm_times

s = ''

data = frame.bytes

# Parse certain kinds of frames

if frame.type == CRSF.MSG_TYPE_PPM:

channels = ppm_channels_decode(frame.payload)

s += '\n CH1..16: ' + ', '.join(map(str, channels))

curr = time.time()*1000

ppm_times = [curr] + ppm_times

ppm_times = ppm_times[:500]

s += ' {:.2f} {:.2f}'.format(curr - last_time,

(ppm_times[0] - ppm_times[-1])/len(ppm_times))

last_time = curr

elif frame.type == CRSF.MSG_TYPE_PPM3:

s += '\n CRSFv3'

elif frame.type == CRSF.MSG_TYPE_LINK_STATS:

s += '\n ' + explain_link_stats(frame.payload, colorize)

elif frame.type == CRSF.MSG_TYPE_DEVICE_INFO:

device_name, sn, hw_id, fw_id, param_count, version = frame.parse()

s += '\n Device: {name}, S/N=0x{sn:x} HW_ID=0x{hw_id:x}, SW_ID=0x{sw_id:x}, param count={cnt}, v={v}'.format(

sn=sn, name=device_name, hw_id=hw_id, sw_id=fw_id, cnt=param_count, v=version)

elif frame.type == CRSF.MSG_TYPE_PARAM_ENTRY:

i = 5 # payload start

if data[i+3:i+4] == [CRSF.PARAM_TYPE_INFO]:

# "Info" parameter (unmodifiable string)

name = data[i+4:]

delim = name.index(0x00)

name, val = name[:delim], data[i+4+delim+1:]

val = val[:val.index(0x00)]

name = bytes(name).decode()

val = bytes(val).decode()

s += '\n ' + name + ': ' + val

elif frame.type == CRSF.MSG_TYPE_LOG:

tm = functools.reduce(lambda v,e: v*256 + e, (data[5:9]))

val = ''

if data[-2] == 0:

val = data[9:-2].decode()

else:

val = ' '.join(map(lambda x: "%02x" % x, data[9:-1]))

s += '\n tick {ticks} ({ms} ms): {val}'.format(ticks=tm, ms=tm//(TICK_SPEED*1000), val=val)

'''Print CRSF frame in a partially parsed way'''

s = str(header) + ': ' + str(frame)

s += explain_frame(frame, COLORIZE)

'''Takes CRSF frame data, sets correct length byte and adds CRC8 byte'''

frame = bytearray([CRSF.SYNC, 0]) + bytearray(data)

frame[CRSF.OFFSET_LENGTH] = len(frame) - 1

frame.append(calc_crc8(frame[2:]))

'''Create broadcast PING frame from bytes'''

return create_frame([

CRSF.MSG_TYPE_PING, # type

dest, # destination

orig # origin

return create_frame([

CRSF.MSG_TYPE_PARAM_READ, # type

dest, # destination

orig, # origin

param_num,

chunk

return create_frame([

CRSF.MSG_TYPE_PARAM_WRITE, # type

dest, # destination

orig, # origin

param_num] +

list(value)

'''Return "fake" device information about Agent Python'''

name = "Agent Python"

return create_frame([

CRSF.MSG_TYPE_DEVICE_INFO, # type

dest, # destination

orig, # origin

# Payload

] + list(name.encode()) + [ # name of entity

0x00, # NUL to terminate string

0x12, 0x34, 0x56, 0x78, # S/N

0x01, 0x23, 0x45, 0x02, # HW ID

0x00, 0x00, 0x11, 0x11, # FW ID

0x00, # parameter total

0x01 # parameter version number

def read_crsf(self):

pass

def write_crsf(self, frame):

'''Class for exchanging CRSF over TCP'''

TCP_TIMEOUT_MS = 1000

TCP_RECV = 2048

def __init__(self, silent):

self.parser = crsf_parser(silent)

self.frames = [] # incoming frames that were already parsed

self.silent = silent

# Connect via TCP socket

self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

self.socket.settimeout(TCPConnection.TCP_TIMEOUT_MS)

self.socket.connect((TCP_HOST, TCP_PORT))

def read_crsf(self):

# Receive data from serial

data = None

try:

data = self.socket.recv(TCPConnection.TCP_RECV)

except socket.timeout:

if not self.silent:

sys.stderr.write('Timeout\n')

except KeyboardInterrupt:

if not self.silent:

print('KeyboardInterrup: Quit')

sys.exit(0)

except Exception as e:

if not self.silent:

sys.stderr.write('ERROR: TCP disconnected\n')

del self.socket

raise(e)

# TODO: move to super

# Parse data

if data and len(data):

#print(len(data))

for frame in self.parser.digest(data):

self.frames.append(frame)

# Return next frame

return self.frames.pop(0) if self.frames else None

def write_crsf(self, frame):

'''Class for exchanging CRSF over UART'''

SERIAL_SLEEP = 0.001

def __init__(self, silent):

self.parser = crsf_parser(silent)

self.serial = serial.Serial(SERIAL_PORT, baudrate=SERIAL_BAUD)

self.in_queue = queue.Queue()

# Receiving thread

self.thread = threading.Thread(target=self.read_thread)

self.thread.setDaemon(1)

self.alive = threading.Event()

self.alive.set()

self.thread.start()

def read_thread(self):

'''Separate thread for receiving frames asynchronously'''

while self.alive.isSet():

# How many bytes are in waiting?

waiting = self.serial.in_waiting

data = self.serial.read(waiting if waiting else 1)

if data and len(data):

for frame in self.parser.digest(data):

self.in_queue.put(frame, block=False)

time.sleep(SerialConnection.SERIAL_SLEEP)

print('Exit')

def read_crsf(self):

try:

frame = self.in_queue.get(block=False)

except queue.Empty:

frame = None

return frame

def write_crsf(self, frame):

'''Parse command line arguments'''

import argparse, pathlib

arg_parse = argparse.ArgumentParser()

arg_parse.add_argument('--tcp', action = 'store_true',

help = 'use TCP connection instead of UART' )

arg_parse.add_argument('--menu', action = 'store_true',

help = 'CRSF menu mode (otherwise - logs mode)' )

opts = arg_parse.parse_args()

if not silent:

print('Connecting with {}...'.format('TCP' if use_tcp else 'UART'))

if use_tcp:

return TCPConnection(silent)

else:

print('Press Ctrl+C to exit')

last_ping = last_read = 0

# Outer loop to reconnect on disconnect

while True:

conn = get_crsf_connection(use_tcp, False)

# Update state machine

while True:

try:

frame = conn.read_crsf()

except Exception as e:

print("err", e)

break

# Display data

if frame:

crsf_log_frame('Received', frame)

# Process frame

if frame.type == CRSF.MSG_TYPE_PING:

resp_frame = create_device_info_frame(dest=frame.origin, orig=ORIGIN_ADDR)

conn.write_crsf(resp_frame)

# Send ping frame (need to send something periodically so that connection is not reset)

if time.time() - last_ping > 10:

last_ping = time.time()

# Send ping frame

frame = create_ping_frame()

crsf_log_frame('Sending ping', frame)

conn.write_crsf(frame)

#frame = create_frame([CRSF.MSG_TYPE_PARAM_READ, CRSF.TX_ADDR, CRSF.CLOUD_ADDR, 1, 0])

#s.send(frame.bytes)

'''

# Parameter types determine the type of the value

TEXT_SELECTION_TYPE = 0x09 # value is uint8_t, with a list of options

STRING_TYPE = 0x0a # value is a string, entered by user directly

FOLDER_TYPE = 0x0b # no modifiable value; groups other parameters

INFO_TYPE = 0x0c # value is constant, CANNOT be modified

COMMAND_TYPE = 0x0d

# Numeric types (in practice, rarely used)

UINT8_TYPE = 0

INT8_TYPE = 1

UINT16_TYPE = 2

INT16_TYPE = 3

UINT32_TYPE = 4

INT32_TYPE = 5

FLOAT_TYPE = 8

def __init__(self, param_num, debug_cb=None):

# TODO: track if parameter is changing frequently or always stable

# TODO: poll stable parameters less often

# TODO: ensure that different chunks are from short time frame

# TODO: detect changes in chunks, invalidate parameter if some chunk changed

# Common CRSF-related attributes

self.num = param_num

self.parent_folder = None

self.type = None

self.name = '...'

self.chunks = [] # payloads of all chunks

self.value = None # except for FOLDER_TYPE

self.hidden = False

# Type-specific attributes

self.min = self.default = self.max = None # numeric types and TEXT_SELECTION_TYPE

self.unit = None # only in numeric and TEXT_SELECTION_TYPE

self.children = None # only for FOLDER_TYPE

self.options = [] # only for TEXT_SELECTION_TYPE

self.max_length = None # only for STRING_TYPE

self.decimal_point = None # only for FLOAT_TYPE

self.step_size = None # only for FLOAT_TYPE

self.status = None # only for COMMAND_TYPE

self.timeout = None # only for COMMAND_TYPE

self.info = None # only for COMMAND_TYPE

# Additional attributes

self.debug_cb = debug_cb

self.created_time = time.time()

self.obtained_time = 0 # when was this parameter last refreshed?

def is_folder(self):

return self.type == self.FOLDER_TYPE

def is_info(self):

return self.type == self.INFO_TYPE

def is_string_input(self):

return self.type == self.STRING_TYPE

def is_selection(self):

return self.type == self.TEXT_SELECTION_TYPE

def is_command(self):

return self.type == self.COMMAND_TYPE

def is_float(self):

return self.type == self.FLOAT_TYPE

def is_number_input(self):

return self.type in [self.FLOAT_TYPE,

self.UINT8_TYPE, self.INT8_TYPE,

self.UINT16_TYPE, self.INT16_TYPE,

self.UINT32_TYPE, self.INT32_TYPE]

def is_input(self):

return self.is_command() or self.is_string_input() or self.is_number_input() or self.is_selection()

def create_param_read_frame(self, origin):

if self.chunks:

# Determine if any chunks are missing or need updating

if None in self.chunks:

chunk = self.chunks.index(None)

else:

self.debug("error 5")

return None

else:

chunk = 0

frame = create_param_read_frame(origin, self.num, chunk)

return frame

def create_write_frame(self, origin, value: bytearray):

if value is not None:

return create_param_write_frame(origin, self.num, value)

else:

self.debug('error: value is None')

def get_float_decimal(self):

return eval(('0.' + '0'*self.decimal_point + '1') if self.decimal_point else '1')

def render_float(self, val=None):

'''Convert integer int32 to float based on self.decimal_point'''

if val is None:

val = self.value

if self.type == self.FLOAT_TYPE:

return val * self.get_float_decimal()

else:

raise ValueError("not float")

def encode_float(self, val):

'''Convert float to int32 based on self.decimal_point'''

val = int(round(val / self.get_float_decimal()))

if val < self.min or val > self.max:

self.debug("error: {} out of range {}..{}".format(val, self.min, self.max))

return None

return val

def debug(self, txt):

if self.debug_cb:

self.debug_cb(txt)

def process_frame(self, frame):

'''Parse parameter information frame'''

if frame.type != CRSF.MSG_TYPE_PARAM_ENTRY:

self.debug('invalid frame type')

return

payload = frame.payload

param_num, chunks_remain = payload[:2]

# Handle chunking

if chunks_remain:

# More chunks expected.

if not self.chunks:

# Assumption! Received first chunk

self.chunks = [None]*(chunks_remain + 1)

self.chunks[-chunks_remain - 1] = frame.payload

return

elif self.chunks:

# Last chunk received. Reassemble payload from chunks.

self.chunks[-1] = frame.payload

if None not in self.chunks and \

list(reversed([x[1] for x in self.chunks])) == list(range(len(self.chunks))) and \

len([1 for x in self.chunks if x[0] == self.num]) == len(self.chunks):

payload = self.chunks[0][:2]

for x in self.chunks:

payload += x[2:]

else:

self.debug("error: chunks error")

return

self.chunks = []

else:

# Trivial case: no chunking.

# CRSF frame contains the parameter entry entirely.

pass

parent_folder, data_type = payload[2:4]

hidden, data_type = data_type & 0x80, data_type & 0x7F

try:

nul = 4 + payload[4:].index(0x00)

name = bytes(payload[4:nul]).decode()

except:

# Invalid frame?

self.debug('frame error: ' + str(frame))

return

# Type-specific value

if data_type == self.FOLDER_TYPE:

try:

end = nul + 1 + payload[nul+1:].index(0xFF)

children = list(payload[nul+1:end])

except:

self.debug('frame error 2: ' + str(frame))

return

# Folder frame parsed successfully

if chunks_remain == 0:

self.children = children

self.parent_folder = parent_folder

self.type = data_type

self.name = name

self.hidden = hidden

self.debug("folder %d OK" % param_num)

self.obtained_time = time.time()

else:

self.debug("error: folder %d: multichunk folders not supported" % param_num)

return

elif data_type == self.COMMAND_TYPE:

try:

end = nul + 3 + payload[nul+3:].index(0x00)

info = bytes(payload[nul+3:end]).decode()

except:

self.debug('frame error 7: ' + str(frame))

return

if chunks_remain == 0:

self.parent_folder = parent_folder

self.type = data_type

self.name = name

self.hidden = hidden

self.status = payload[nul + 1]

self.timeout = payload[nul + 2]*0.1 if payload[nul + 2] % 10 else payload[nul + 2]//10

self.info = info

self.debug("command %d OK" % param_num)

self.obtained_time = time.time()

else:

self.debug("error: folder %d: multichunk folders not supported" % param_num)

return

elif data_type == self.FLOAT_TYPE:

try:

end = nul + 22 + payload[nul+22:].index(0x00)

unit = bytes(payload[nul+22:end]).decode()

except:

self.debug('frame error 5: ' + str(frame))

return

if chunks_remain == 0:

self.parent_folder = parent_folder

self.type = data_type

self.hidden = hidden

self.name = name

self.value = CRSF.decode_int32(payload[nul+1:nul+5])

self.min = CRSF.decode_int32(payload[nul+5:nul+9])

self.max = CRSF.decode_int32(payload[nul+9:nul+13])

self.default = CRSF.decode_int32(payload[nul+13:nul+17])

self.decimal_point = payload[nul+17]

self.step_size = CRSF.decode_int32(payload[nul+18:nul+22])

self.unit = unit

self.debug("float {} = {}, {}, {} OK".format(param_num, self.value, self.max, self.unit))

self.obtained_time = time.time()

else:

self.debug("error: float %d: multichunk floats not supported" % param_num)

return

elif data_type in [self.STRING_TYPE, self.INFO_TYPE]:

try:

end = nul + 1 + payload[nul+1:].index(0x00)

value = bytes(payload[nul+1:end]).decode()

except:

self.debug('frame error 4: ' + str(frame))

return

self.max_length = payload[end + 1] if data_type == self.STRING_TYPE else len(value)

if chunks_remain == 0:

self.parent_folder = parent_folder

self.type = data_type

self.name = name

self.hidden = hidden

self.value = value

self.debug("{} {} {} OK".format('info' if data_type == self.INFO_TYPE else 'string',

param_num, self.max_length))

self.obtained_time = time.time()

else:

self.debug("error: string/info %d: multichunk strings not supported" % param_num)

return

elif data_type == self.TEXT_SELECTION_TYPE:

try:

end = nul + 1 + payload[nul+1:].index(0x00)

options = bytes(payload[nul+1:end]).decode()

options = options.split(';')

except:

self.debug('frame error 6: ' + str(frame))

return

value, val_min, val_max, val_def = payload[end+1:end+5]

# TODO: unit

if chunks_remain == 0:

self.parent_folder = parent_folder

self.type = data_type

self.name = name

self.hidden = hidden

self.value = value

self.options = options

self.min = val_min

self.default = val_def

self.max = val_max

self.debug("selection %d OK" % (val_max + 1))

self.obtained_time = time.time()

else:

self.debug("error: folder %d: multichunk folders not supported" % param_num)

return

else:

self.debug("error: data_type %d" % data_type)

self.parent_folder = parent_folder

self.type = data_type

self.name = name

self.type = data_type

self.hidden = hidden

if chunks_remain == 0:

'''Holds all the information obtained from a device'''

POLL_PERIOD_S = 2.0

POLL_RESPONSE_SPEED_UP_FACTOR = 0.95

FOLDER_PERIOD_S = 10

PARAM_TIMEOUT_S = 120

class InvalidType(ValueError):

pass

def __init__(self, frame, menu_ui, debug_cb=None):

self.origin = 0 # network address of this device

self.last_seen = 0 # when was the last time "device information" was obtained from the device?

self.last_read_time = 0 # when was the last time "param read" was sent to the device?

self.last_read_frame = 0

# Parsed from "device information" frame

self.name = 0x00

self.sn = 0x00

self.hwid = 0x00

self.fwid = 0x00

self.param_count = 0

self.param_version = 0

if frame and frame.type == CRSF.MSG_TYPE_DEVICE_INFO:

self.name, self.sn, self.hwid, self.fwid, self.param_count, self.param_ver = frame.parse()

self.origin = frame.origin

elif frame:

raise ValueError("frame of wrong type")

# List of CRSFParam objects, first object - root folder

self.menu = [None]*(self.param_count + 1)

self.menu_ui = menu_ui

self.debug_cb = debug_cb

def debug(self, txt):

if self.debug_cb:

self.debug_cb(txt)

def match(self, frame):

'''Compare device information of this device to "device information" frame'''

if frame and frame.type == CRSF.MSG_TYPE_DEVICE_INFO:

return frame.origin == self.origin and \

frame.parse() == \

(self.name, self.sn, self.hwid, self.fwid, self.param_count, self.param_ver)

else:

raise ValueError("DEVICE_INFO frame expected")

def poll_params(self, conn, folder):

'''Periodically send out "param read" frames for the parameters in current folder'''

if not (0 <= folder < len(self.menu)):

self.debug('invalid folder: not 0<=%s<%d' % (str(folder), len(self.menu)))

return

if time.time() - self.last_read_time < self.POLL_PERIOD_S:

return

self.last_read_time = time.time()

if self.menu[folder] is None:

self.menu[folder] = CRSFParam(folder, debug_cb=self.debug_cb)

elif self.menu[folder].obtained_time and not self.menu[folder].is_folder():

raise CRSFDevice.InvalidType('folder type expected')

# Decide which parameter to load

frame = None

if not self.menu[folder].obtained_time or time.time() - self.menu[folder].obtained_time > self.FOLDER_PERIOD_S:

# Load the current folder itself

frame = self.menu[folder].create_param_read_frame(self.origin)

elif self.menu[folder].obtained_time:

oldest_time, oldest_child = None, None

for child in self.menu[folder].children:

if self.menu[child] is None or not self.menu[child].obtained_time and \