mirror of
https://github.com/xssfox/freedvtnc2
synced 2026-08-14 19:31:47 -04:00
update shell and other fixes
This commit is contained in:
xssfox
parent
2c59a7db2d
commit
190fe84bb8
6 changed files with 351 additions and 66 deletions
|
|
@ -11,19 +11,6 @@ import sys,struct,fcntl,termios
|
|||
|
||||
logging.basicConfig()
|
||||
|
||||
|
||||
def blank_current_readline():
|
||||
# Next line said to be reasonably portable for various Unixes
|
||||
(rows,cols) = struct.unpack('hh', fcntl.ioctl(sys.stdout, termios.TIOCGWINSZ,'1234'))
|
||||
|
||||
text_len = len(readline.get_line_buffer())+2
|
||||
|
||||
# ANSI escape sequences (All VT100 except ESC[0G)
|
||||
sys.stdout.write('\x1b[2K') # Clear current line
|
||||
sys.stdout.write('\x1b[1A\x1b[2K'*(text_len//cols)) # Move cursor up and clear line
|
||||
sys.stdout.write('\x1b[0G') # Move to start of line
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
p = configargparse.ArgParser(default_config_files=['/etc/freedvtnc2.conf', '~/.freedvtnc2.conf'])
|
||||
p.add('-c', '-config', required=False, is_config_file=True, help='config file path')
|
||||
|
|
@ -35,6 +22,7 @@ if __name__ == '__main__':
|
|||
|
||||
p.add('--input-device', type=str, default=None, env_var="FREEDVTNC2_INPUT_DEVICE")
|
||||
p.add('--output-device', type=str, default=None, env_var="FREEDVTNC2_OUTPUT_DEVICE")
|
||||
p.add('--output-volume', type=float, default=0, env_var="FREEDVTNC2_OUTPUT_DB", help="in db. postive = louder, negative = quiter")
|
||||
|
||||
p.add('--mode', type=str, choices=[x.name for x in Modems], default=Modems.DATAC1.name, help="The TX mode for the modem. The modem will receive all modes at once")
|
||||
|
||||
|
|
@ -65,6 +53,10 @@ if __name__ == '__main__':
|
|||
def tx(data):
|
||||
logging.debug(f"Sending {str(data)}")
|
||||
output_device.write(modem_tx.write(data))
|
||||
|
||||
def progress(total:int, remaining:int, mode:str):
|
||||
if not options.no_cli:
|
||||
shell.progress(total, remaining, mode)
|
||||
|
||||
def rx(data: Packet):
|
||||
logging.debug(f"Received {str(data.header)} - {str(data.data)}")
|
||||
|
|
@ -76,13 +68,7 @@ if __name__ == '__main__':
|
|||
# ignoring debug messages - this is the only place where we have this issues - if we add more threaded output
|
||||
# we should move this into a dedicated function
|
||||
if not options.no_cli:
|
||||
blank_current_readline()
|
||||
print(f"<{call.decode()}> {message.decode()}")
|
||||
if readline.get_line_buffer()[-1:] == "\n": # the readline buffer doesn't get cleared on libedit - I haven't tested this on gnureadline
|
||||
sys.stdout.write(shell.prompt)
|
||||
else:
|
||||
sys.stdout.write(shell.prompt + readline.get_line_buffer())
|
||||
sys.stdout.flush()
|
||||
shell.add_text(f"<{call.decode()}> {message.decode()}\n")
|
||||
else:
|
||||
print(f"\n<{call.decode()}> {message.decode()}")
|
||||
|
||||
|
|
@ -91,7 +77,10 @@ if __name__ == '__main__':
|
|||
else:
|
||||
tnc_interface = tnc.KissTCPInterface(tx, port=options.kiss_tcp_port, address=options.kiss_tcp_address)
|
||||
|
||||
modem_rx = FreeDVRX(callback=rx)
|
||||
def inhibit(state):
|
||||
output_device.inhibit = state
|
||||
|
||||
modem_rx = FreeDVRX(callback=rx, progress=progress, inhibit=inhibit)
|
||||
|
||||
input_device_name_or_id = options.input_device
|
||||
output_device_name_or_id = options.output_device
|
||||
|
|
@ -117,22 +106,16 @@ if __name__ == '__main__':
|
|||
ptt_release=ptt_release,
|
||||
ptt_trigger=ptt_trigger,
|
||||
ptt_on_delay_ms=options.ptt_on_delay_ms,
|
||||
ptt_off_delay_ms=options.ptt_off_delay_ms
|
||||
ptt_off_delay_ms=options.ptt_off_delay_ms,
|
||||
db=options.output_volume
|
||||
)
|
||||
|
||||
try:
|
||||
if not options.no_cli:
|
||||
if 'libedit' in readline.__doc__: # macos hack
|
||||
readline.parse_and_bind ("bind ^I rl_complete")
|
||||
|
||||
shell = FreeDVShell()
|
||||
shell.modem_rx = modem_rx
|
||||
shell.modem_tx = modem_tx
|
||||
shell.input_device = input_device
|
||||
shell.output_device = output_device
|
||||
shell = FreeDVShell(modem_rx, modem_tx, output_device, input_device)
|
||||
if options.callsign:
|
||||
shell.callsign = options.callsign
|
||||
shell.cmdloop()
|
||||
shell.shell_commands.callsign = options.callsign
|
||||
shell.run()
|
||||
else:
|
||||
while 1:
|
||||
time.sleep(0.1)
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ from threading import Lock
|
|||
from typing import Callable
|
||||
#from pydub import pyaudioop
|
||||
import pydub
|
||||
|
||||
import math
|
||||
|
||||
p = pyaudio.PyAudio()
|
||||
|
||||
|
|
@ -71,6 +71,7 @@ class InputDevice():
|
|||
"""
|
||||
|
||||
rate_state = None # used for sample rate conversions
|
||||
input_level = -99
|
||||
|
||||
def __init__(self, callback: Callable[[bytes], None], sample_rate:int, name_or_id:str|int|None=None):
|
||||
self.sample_rate = sample_rate
|
||||
|
|
@ -110,7 +111,6 @@ class InputDevice():
|
|||
frames_per_buffer=4096
|
||||
)
|
||||
|
||||
|
||||
def close(self):
|
||||
self.stream.close()
|
||||
|
||||
|
|
@ -121,6 +121,12 @@ class InputDevice():
|
|||
self.close()
|
||||
|
||||
def pa_callback(self, in_data: bytes, frame_count: int, time_info, status_flag):
|
||||
max_audio = pyaudioop.max(in_data,pyaudio.get_sample_size(FORMAT))
|
||||
if max_audio:
|
||||
self.input_level = 20*math.log10(max_audio/(2**(self.bit_depth*8-1)))
|
||||
else:
|
||||
self.input_level = -99
|
||||
|
||||
if self.device.input_channels == 2:
|
||||
in_data = pyaudioop.tomono(
|
||||
in_data,
|
||||
|
|
@ -156,11 +162,26 @@ class OutputDevice():
|
|||
|
||||
output_buffer_lock = Lock()
|
||||
|
||||
def __init__(self, sample_rate: int, name_or_id:int|str|None=None, ptt_trigger:Callable[[],None]=None, ptt_release:Callable[[],None]=None, ptt_on_delay_ms:int=0, ptt_off_delay_ms:int=0):
|
||||
inhibit = False
|
||||
|
||||
@property
|
||||
def queue_ms(self):
|
||||
return (len(self.buffer)/self.bit_depth/self.device.sample_rate)*1000
|
||||
|
||||
def __init__(self,
|
||||
sample_rate: int,
|
||||
name_or_id:int|str|None=None,
|
||||
ptt_trigger:Callable[[],None]=None,
|
||||
ptt_release:Callable[[],None]=None,
|
||||
ptt_on_delay_ms:int=0,
|
||||
ptt_off_delay_ms:int=0,
|
||||
db:float=0
|
||||
):
|
||||
self.sample_rate = sample_rate
|
||||
self.bit_depth = pyaudio.get_sample_size(FORMAT)
|
||||
self.ptt_on_delay_ms = ptt_on_delay_ms
|
||||
self.ptt_off_delay_ms = ptt_off_delay_ms
|
||||
self.db = db
|
||||
|
||||
if name_or_id:
|
||||
try:
|
||||
|
|
@ -214,6 +235,10 @@ class OutputDevice():
|
|||
self.device.sample_rate,
|
||||
self.rate_state,
|
||||
)
|
||||
|
||||
if self.db:
|
||||
data = pyaudioop.mul(data, 2, 10**(self.db/20.0))
|
||||
|
||||
if self.device.output_channels == 2:
|
||||
data = pyaudioop.tostereo(
|
||||
data,
|
||||
|
|
@ -240,6 +265,10 @@ class OutputDevice():
|
|||
|
||||
ptt = False
|
||||
|
||||
# if we aren't transmitting and we have inhibited tx then skip
|
||||
if self.inhibit == True and self.ptt == False:
|
||||
return (bytes(output), pyaudio.paContinue)
|
||||
|
||||
with self.output_buffer_lock:
|
||||
chunk_size = min(len(self.buffer), buffer_size)
|
||||
output[:chunk_size] = self.buffer[:chunk_size]
|
||||
|
|
|
|||
|
|
@ -198,10 +198,13 @@ class Modem():
|
|||
class Packet():
|
||||
data: bytes
|
||||
header: int
|
||||
mode: str
|
||||
|
||||
class FreeDVRX():
|
||||
def __init__(self, callback: Callable[[bytes],None]):
|
||||
def __init__(self, callback: Callable[[bytes],None], progress: Callable[[int,int],None], inhibit: Callable[[bool],None]):
|
||||
self.callback = callback
|
||||
self.progress = progress
|
||||
self.inhibit = inhibit
|
||||
|
||||
# we RX all the modems at once
|
||||
self.modems = [Modem(x, callback=self.rx) for x in Modems]
|
||||
|
|
@ -221,14 +224,20 @@ class FreeDVRX():
|
|||
"""
|
||||
Accepts bytes of data that will be read by tge modem and demodulated
|
||||
"""
|
||||
sync = False
|
||||
for modem in self.modems:
|
||||
modem.write(data)
|
||||
if modem.sync:
|
||||
sync = True
|
||||
self.inhibit(sync)
|
||||
|
||||
def rx(self, data_frame: FreeDVFrame):
|
||||
logging.debug(f"Received data. snr:{data_frame.snr}")
|
||||
data = bytearray(data_frame.data)
|
||||
header = data.pop(0)
|
||||
if header > 200: # start of packet
|
||||
self.remaining_bytes = int.from_bytes(data[0:2])
|
||||
self.total_bytes = self.remaining_bytes
|
||||
del data[0:2]
|
||||
logging.debug(f"Found packet start - Expecting {self.remaining_bytes} bytes")
|
||||
self.next_seq_number = 0
|
||||
|
|
@ -254,10 +263,11 @@ class FreeDVRX():
|
|||
|
||||
logging.debug(f"Seq: {header} Remaining data: {self.remaining_bytes}")
|
||||
|
||||
self.progress(self.total_bytes, self.remaining_bytes, data_frame.modem)
|
||||
if self.remaining_bytes == 0:
|
||||
self.next_seq_number = None
|
||||
self.remaining_bytes = None
|
||||
self.callback(Packet(header=self.header, data=self.partial_data))
|
||||
self.callback(Packet(header=self.header, data=self.partial_data, mode=data_frame.modem))
|
||||
|
||||
class FreeDVTX():
|
||||
def __init__(self, modem: Modem = Modems.DATAC1):
|
||||
|
|
|
|||
27
freedvtnc2/poetry.lock
generated
27
freedvtnc2/poetry.lock
generated
|
|
@ -93,6 +93,20 @@ files = [
|
|||
[package.dependencies]
|
||||
pyserial = ">=3.4"
|
||||
|
||||
[[package]]
|
||||
name = "prompt-toolkit"
|
||||
version = "3.0.43"
|
||||
description = "Library for building powerful interactive command lines in Python"
|
||||
optional = false
|
||||
python-versions = ">=3.7.0"
|
||||
files = [
|
||||
{file = "prompt_toolkit-3.0.43-py3-none-any.whl", hash = "sha256:a11a29cb3bf0a28a387fe5122cdb649816a957cd9261dcedf8c9f1fef33eacf6"},
|
||||
{file = "prompt_toolkit-3.0.43.tar.gz", hash = "sha256:3527b7af26106cbc65a040bcc84839a3566ec1b051bb0bfe953631e704b0ff7d"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
wcwidth = "*"
|
||||
|
||||
[[package]]
|
||||
name = "pyaudio"
|
||||
version = "0.2.14"
|
||||
|
|
@ -166,7 +180,18 @@ files = [
|
|||
[package.extras]
|
||||
widechars = ["wcwidth"]
|
||||
|
||||
[[package]]
|
||||
name = "wcwidth"
|
||||
version = "0.2.12"
|
||||
description = "Measures the displayed width of unicode strings in a terminal"
|
||||
optional = false
|
||||
python-versions = "*"
|
||||
files = [
|
||||
{file = "wcwidth-0.2.12-py2.py3-none-any.whl", hash = "sha256:f26ec43d96c8cbfed76a5075dac87680124fa84e0855195a6184da9c187f133c"},
|
||||
{file = "wcwidth-0.2.12.tar.gz", hash = "sha256:f01c104efdf57971bcb756f054dd58ddec5204dd15fa31d6503ea57947d97c02"},
|
||||
]
|
||||
|
||||
[metadata]
|
||||
lock-version = "2.0"
|
||||
python-versions = "^3.11"
|
||||
content-hash = "0c76f5a37399d9d07cc95a4fee91abffff6ea8efbd181335dc1e685678534acb"
|
||||
content-hash = "04edbe4763c33088c098a98a234e973ec663bd489f67383785eacd5b74b242d3"
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ pyaudio = "^0.2.14"
|
|||
tabulate = "^0.9.0"
|
||||
pydub = "^0.25.1"
|
||||
kissfix = "^7.0.11"
|
||||
prompt-toolkit = "^3.0.43"
|
||||
|
||||
|
||||
[build-system]
|
||||
|
|
|
|||
|
|
@ -1,79 +1,131 @@
|
|||
import logging
|
||||
import cmd
|
||||
from .modem import Modems, lib, ffi
|
||||
import readline
|
||||
import rlcompleter
|
||||
import code
|
||||
import sys
|
||||
from prompt_toolkit import Application
|
||||
from prompt_toolkit.buffer import Buffer
|
||||
from prompt_toolkit.key_binding import KeyBindings
|
||||
from prompt_toolkit.layout.containers import HSplit, Window, VSplit
|
||||
from prompt_toolkit.layout.controls import BufferControl, FormattedTextControl
|
||||
from prompt_toolkit.layout.layout import Layout
|
||||
from prompt_toolkit.layout.dimension import LayoutDimension
|
||||
from prompt_toolkit.widgets import TextArea, ProgressBar
|
||||
from prompt_toolkit.document import Document
|
||||
from prompt_toolkit.styles import Style
|
||||
from prompt_toolkit.key_binding.bindings.page_navigation import scroll_page_up, scroll_page_down
|
||||
import time
|
||||
import logging
|
||||
from prompt_toolkit.completion import WordCompleter
|
||||
import sys
|
||||
from . import audio
|
||||
|
||||
import readline
|
||||
import code
|
||||
import rlcompleter
|
||||
import pydub.generators
|
||||
from .modem import Modems
|
||||
import traceback
|
||||
|
||||
class FreeDVShell(cmd.Cmd):
|
||||
intro = "FreeDVTNC2 Shell - type help or ? to list commands\n"
|
||||
prompt = "(freedvtnc2) "
|
||||
class LogHandler(logging.StreamHandler):
|
||||
def __init__(self, text_area:TextArea):
|
||||
self.text_area = text_area
|
||||
super().__init__()
|
||||
def emit(self, record):
|
||||
|
||||
msg = self.text_area.text + self.format(record) + "\n"
|
||||
self.text_area.buffer.document = Document(
|
||||
text=msg, cursor_position=len(msg)
|
||||
)
|
||||
|
||||
class FreeDVShellCommands():
|
||||
callsign = None
|
||||
|
||||
def __init__(self, modem_tx, output_device):
|
||||
self.modem_tx = modem_tx
|
||||
self.output_device = output_device
|
||||
|
||||
@property
|
||||
def commands(self):
|
||||
return [func[3:] for func in dir(self) if func.startswith("do_")]
|
||||
|
||||
@property
|
||||
def help(self):
|
||||
return {
|
||||
func[3:] : getattr(self, f"help_{func[3:]}")() if hasattr(self, f"help_{func[3:]}") else getattr(self,func).__doc__
|
||||
for func in dir(self) if func.startswith("do_")
|
||||
}
|
||||
|
||||
def do_log_level(self, arg):
|
||||
"Set the log level"
|
||||
arg=arg.upper()
|
||||
if arg not in logging._nameToLevel.keys():
|
||||
return f"Must be one of : {','.join(logging._nameToLevel.keys())}"
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(level=arg.upper())
|
||||
logger.setLevel(level=arg)
|
||||
return f"Set log level to {arg}"
|
||||
|
||||
def do_test_ptt(self, arg):
|
||||
"Turns on PTT for 2 seconds"
|
||||
print("Starting TX")
|
||||
sin_wave = pydub.generators.Sine(
|
||||
440,
|
||||
sample_rate=self.modem_tx.modem.sample_rate,
|
||||
bit_depth=16,
|
||||
).to_audio_segment(2000)
|
||||
).to_audio_segment(2000, volume=-6)
|
||||
sin_wave.set_channels(1)
|
||||
self.output_device.write(sin_wave.raw_data)
|
||||
print("Stopping TX")
|
||||
|
||||
def help_mode(self):
|
||||
return f"Change TX Mode: mode [{', '.join([x.name for x in Modems])}]"
|
||||
def do_mode(self, arg):
|
||||
if arg == "":
|
||||
print(f"Current mode: {self.modem_tx.modem.modem_name}")
|
||||
return
|
||||
return f"Current mode: {self.modem_tx.modem.modem_name}"
|
||||
|
||||
|
||||
arg = arg.upper()
|
||||
|
||||
if arg not in [x.name for x in Modems]:
|
||||
print(f"Mode must be {', '.join([x.name for x in Modems])}")
|
||||
return f"Mode must be {', '.join([x.name for x in Modems])}"
|
||||
else:
|
||||
modem = {x.name:x for x in Modems}[arg]
|
||||
self.modem_tx.set_mode(modem)
|
||||
print(f"Set mode {arg}")
|
||||
|
||||
def help_mode(self):
|
||||
print(f"Change TX Mode: mode [{', '.join([x.name for x in Modems])}]")
|
||||
return f"Set mode {arg}"
|
||||
|
||||
def do_clear(self, arg):
|
||||
"Clears TX queues"
|
||||
self.output_device.clear()
|
||||
print("TX buffer cleared")
|
||||
return "TX buffer cleared"
|
||||
|
||||
def do_list_audio_devices(self, arg):
|
||||
"Lists audio device parameters"
|
||||
print(audio.devices)
|
||||
return audio.devices
|
||||
|
||||
def do_help(self, arg):
|
||||
"This help"
|
||||
header = "\nFreeDVTNC2 Help\n---------------\n"
|
||||
commands = "\n".join([f"{command}\n {help_string}" for command, help_string in self.help.items()])
|
||||
return header+commands+"\n"
|
||||
|
||||
def do_send_string(self, arg):
|
||||
"Sends string over the modem"
|
||||
|
||||
self.output_device.write(self.modem_tx.write(arg.encode()))
|
||||
return "Queued for sending"
|
||||
|
||||
def do_volume(self,arg):
|
||||
"Set the volume gain in db for output level - you probably want to use soundcard configuration or radio configuration rather than this."
|
||||
self.output_device.db = float(arg)
|
||||
return f"Set TX volume to {float(arg)} db"
|
||||
|
||||
def do_callsign(self,arg):
|
||||
"Sets callsign - example: callsign N0CALL"
|
||||
self.callsign=arg
|
||||
return f"Callsign set to {arg}"
|
||||
def do_msg(self, arg):
|
||||
"Send a message"
|
||||
|
||||
if not self.callsign:
|
||||
self.callsign = input("Your callsign:")
|
||||
return "Set callsign with the callsign command\n"
|
||||
|
||||
data = self.callsign.encode() + b"\xff" + arg.encode()
|
||||
|
||||
self.output_device.write(self.modem_tx.write(data, header_byte=b"\xfe"))
|
||||
|
||||
def do_exit(self, arg):
|
||||
"Exits FreeDVTNC2"
|
||||
raise KeyboardInterrupt
|
||||
|
||||
def do_debug(self, arg):
|
||||
|
|
@ -95,5 +147,190 @@ class FreeDVShell(cmd.Cmd):
|
|||
shell.interact(banner="freedvtnc2 debug console")
|
||||
except SystemExit:
|
||||
pass
|
||||
def emptyline(self):
|
||||
pass
|
||||
class FreeDVShell():
|
||||
def __init__(self, modem_rx, modem_tx, output_device, input_device):
|
||||
self.modem_tx = modem_tx
|
||||
self.modem_rx = modem_rx
|
||||
self.output_device = output_device
|
||||
self.input_device = input_device
|
||||
|
||||
self.logger = logging.getLogger()
|
||||
self.shell_commands = FreeDVShellCommands(modem_tx, output_device)
|
||||
self.log = TextArea(
|
||||
text="",
|
||||
scrollbar=True,
|
||||
line_numbers=False,
|
||||
)
|
||||
while self.logger.hasHandlers(): # remove existing handlers
|
||||
self.logger.removeHandler(self.logger.handlers[0])
|
||||
self.log_handler = LogHandler(self.log)
|
||||
self.log_handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s"))
|
||||
self.logger.addHandler(self.log_handler)
|
||||
def add_text(self, text):
|
||||
new_text = self.log.text + text
|
||||
|
||||
# Add text to output buffer.
|
||||
self.log.buffer.document = Document(
|
||||
text=new_text, cursor_position=len(new_text)
|
||||
)
|
||||
def progress(self, total:int, remaining:int, mode:str):
|
||||
self.pb.percentage = ((total - remaining)/total)*100
|
||||
self.pb_text.buffer.document = Document(f" {(total - remaining)}/{total} bytes [{mode}]")
|
||||
def run(self):
|
||||
|
||||
def accept(buff):
|
||||
try:
|
||||
command, arg = input_field.text.split(" ", 1)
|
||||
except ValueError:
|
||||
command = input_field.text
|
||||
arg = ""
|
||||
try:
|
||||
command = getattr(self.shell_commands, "do_" + command)
|
||||
try:
|
||||
command_result = command(arg)
|
||||
if command_result:
|
||||
output = str(command_result) + "\n"
|
||||
else:
|
||||
output = ""
|
||||
except KeyboardInterrupt:
|
||||
raise KeyboardInterrupt
|
||||
except:
|
||||
output = traceback.format_exc() + "\n"
|
||||
except Exception:
|
||||
output = "Invalid command. Valid commands: " + ", ".join(self.shell_commands.commands) + "\n"
|
||||
|
||||
new_text = self.log.text + output
|
||||
|
||||
# Add text to output buffer.
|
||||
self.log.buffer.document = Document(
|
||||
text=new_text, cursor_position=len(new_text)
|
||||
)
|
||||
|
||||
input_field = TextArea(
|
||||
height=3,
|
||||
prompt="(freedvtnc2) ",
|
||||
style="class:input-field",
|
||||
multiline=False,
|
||||
wrap_lines=False,
|
||||
accept_handler=accept,
|
||||
completer=WordCompleter(self.shell_commands.commands)
|
||||
)
|
||||
|
||||
|
||||
|
||||
|
||||
def get_statusbar_text():
|
||||
if self.input_device.input_level > -5.0:
|
||||
dbfs_color = "red"
|
||||
elif self.input_device.input_level < -90:
|
||||
dbfs_color = "red"
|
||||
elif self.input_device.input_level < -50:
|
||||
dbfs_color = "yellow"
|
||||
else:
|
||||
dbfs_color = "green"
|
||||
statuses = [
|
||||
# input level
|
||||
# ptt status
|
||||
# tx queue (in seconds?)
|
||||
# each modem snr
|
||||
("class:status", f"Input level: "),
|
||||
|
||||
(f"class:status.{dbfs_color}",f"{self.input_device.input_level:6.2f}"),
|
||||
("class:status",f" dBFS | "),
|
||||
|
||||
("class:status", f"PTT: "),
|
||||
(f"class:status.{ 'red' if self.output_device.ptt else 'green' }", f"{ ' on' if self.output_device.ptt else 'off' }"),
|
||||
("class:status", f" | "),
|
||||
|
||||
("class:status", f"TX Queue: { (self.output_device.queue_ms / 1000) :5.1f}s | "),
|
||||
("class:status", f"Channel: "),
|
||||
(f"class:status.{'red' if self.output_device.inhibit else 'green'}", f"{'busy' if self.output_device.inhibit else 'clear'}"),
|
||||
("class:status", " |\n"),
|
||||
]
|
||||
|
||||
nl = "\n"
|
||||
snrs = [
|
||||
("class:status", f'{x[1].modem_name}: {x[1].snr:6.2f}db {"|" +nl if x[0] == len(self.modem_rx.modems)-1 else "| "}' ) for x in enumerate(self.modem_rx.modems)
|
||||
]
|
||||
|
||||
|
||||
syncs = []
|
||||
for x in self.modem_rx.modems:
|
||||
syncs.append(("class:status", f"{x.modem_name}: "))
|
||||
syncs.append((f"class:status.{'red' if x.sync == 0 else 'green'}",f"{x.sync:8}"))
|
||||
syncs.append(("class:status",f" | " ))
|
||||
|
||||
|
||||
statuses += snrs
|
||||
statuses += syncs
|
||||
return statuses
|
||||
self.pb = ProgressBar(
|
||||
|
||||
)
|
||||
self.pb_text = TextArea(height=1, width=30,multiline=False,wrap_lines=False,)
|
||||
self.pb.percentage = 0
|
||||
|
||||
self.pbsplit = VSplit(
|
||||
[
|
||||
self.pb,
|
||||
self.pb_text
|
||||
]
|
||||
)
|
||||
|
||||
root_container = HSplit([
|
||||
Window(
|
||||
content=FormattedTextControl(get_statusbar_text),
|
||||
height=LayoutDimension.exact(3),
|
||||
style="class:status",
|
||||
),
|
||||
Window(height=1, char="-", style="class:line"),
|
||||
self.log,
|
||||
Window(height=1, char="-", style="class:line"),
|
||||
self.pbsplit,
|
||||
input_field
|
||||
])
|
||||
|
||||
kb = KeyBindings()
|
||||
|
||||
@kb.add("c-c")
|
||||
@kb.add("c-q")
|
||||
def _(event):
|
||||
"Pressing Ctrl-Q or Ctrl-C will exit the user interface."
|
||||
raise KeyboardInterrupt
|
||||
|
||||
@kb.add("pageup")
|
||||
def _(event):
|
||||
w = event.app.layout.current_window
|
||||
event.app.layout.focus(self.log.window)
|
||||
scroll_page_up(event)
|
||||
event.app.layout.focus(w)
|
||||
|
||||
@kb.add("pagedown")
|
||||
def _(event):
|
||||
w = event.app.layout.current_window
|
||||
event.app.layout.focus(self.log.window)
|
||||
scroll_page_down(event)
|
||||
event.app.layout.focus(w)
|
||||
|
||||
style = Style(
|
||||
[
|
||||
("output-field", "bg:#000044 #ffffff"),
|
||||
("status.red", "bg:#000000 #ff0000"),
|
||||
("status.green", "bg:#000000 #00ff00"),
|
||||
("status.yellow", "bg:#000000 #ffff00"),
|
||||
("input-field", "bg:#000000 #ffffff"),
|
||||
("line", "#004400"),
|
||||
("progress-bar.used","bg:#ffffff"),
|
||||
("progress-bar","bg:#444444"),
|
||||
]
|
||||
)
|
||||
|
||||
app = Application(
|
||||
layout=Layout(root_container, focused_element=input_field),
|
||||
full_screen=True,
|
||||
key_bindings=kb,
|
||||
mouse_support=False,
|
||||
refresh_interval=0.2,
|
||||
style=style
|
||||
)
|
||||
app.run()
|
||||
Loading…
Add table
Add a link
Reference in a new issue