From 190fe84bb8bd743afaffc13e9ef02631a50662e2 Mon Sep 17 00:00:00 2001 From: xssfox Date: Wed, 27 Dec 2023 00:00:24 +1100 Subject: [PATCH] update shell and other fixes --- freedvtnc2/__main__.py | 47 ++---- freedvtnc2/audio.py | 35 ++++- freedvtnc2/modem.py | 14 +- freedvtnc2/poetry.lock | 27 +++- freedvtnc2/pyproject.toml | 1 + freedvtnc2/shell.py | 293 ++++++++++++++++++++++++++++++++++---- 6 files changed, 351 insertions(+), 66 deletions(-) diff --git a/freedvtnc2/__main__.py b/freedvtnc2/__main__.py index 680280e..f871cb7 100644 --- a/freedvtnc2/__main__.py +++ b/freedvtnc2/__main__.py @@ -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) diff --git a/freedvtnc2/audio.py b/freedvtnc2/audio.py index 58b731a..3ad69bb 100644 --- a/freedvtnc2/audio.py +++ b/freedvtnc2/audio.py @@ -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] diff --git a/freedvtnc2/modem.py b/freedvtnc2/modem.py index 4b62bd1..6207c38 100644 --- a/freedvtnc2/modem.py +++ b/freedvtnc2/modem.py @@ -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): diff --git a/freedvtnc2/poetry.lock b/freedvtnc2/poetry.lock index 495769b..151f5d6 100644 --- a/freedvtnc2/poetry.lock +++ b/freedvtnc2/poetry.lock @@ -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" diff --git a/freedvtnc2/pyproject.toml b/freedvtnc2/pyproject.toml index 6923594..f69ba2d 100644 --- a/freedvtnc2/pyproject.toml +++ b/freedvtnc2/pyproject.toml @@ -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] diff --git a/freedvtnc2/shell.py b/freedvtnc2/shell.py index a8e3141..024b565 100644 --- a/freedvtnc2/shell.py +++ b/freedvtnc2/shell.py @@ -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 \ No newline at end of file +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() \ No newline at end of file