PoC of remote CW keyer operation

remote_cw_keyer.py

# SPDX-License-Identifier: GPL-3.0-or-later

import serial
import sys
import time
import threading
import select

# Shared state
wpm = 28
running = True
wpm_lock = threading.Lock()

def wpm_to_millis(wpm_val):
    return 60000 / (wpm_val * 50)

def dit(ser, millis):
    ser.rts = True
    time.sleep(millis)
    ser.rts = False
    time.sleep(millis)

def dah(ser, millis):
    ser.rts = True
    time.sleep(millis * 3)
    ser.rts = False
    time.sleep(millis)

def keying_thread(phys, virt):
    global running, wpm
    while running:
        with wpm_lock:
            millis = wpm_to_millis(wpm)

        if phys.dsr:
            dit(virt, millis)
        elif phys.cts:
            dah(virt, millis)
        else:
            time.sleep(0.005)  # Small delay to reduce CPU usage

def speed_control_thread():
    global wpm, running
    print("Use W/w to change WPM, q to quit")
    while running:
        if select.select([sys.stdin], [], [], 0.1)[0]:
            char = sys.stdin.read(1)
            with wpm_lock:
                if char == 'W':
                    wpm += 1
                elif char == 'w':
                    wpm = max(1, wpm - 1)
                elif char == 'q':
                    running = False
        with wpm_lock:
            print(f"Speed: {wpm} wpm    ", end='\r', flush=True)

def main():
    global running

    try:
        physical_port = serial.Serial('COM1', 9600, timeout=0)
        virtual_port = serial.Serial('COM2', 9600, timeout=0)
    except serial.SerialException as e:
        print(f"Error opening ports: {e}")
        return

    key_thread = threading.Thread(target=keying_thread, args=(physical_port, virtual_port))
    speed_thread = threading.Thread(target=speed_control_thread)

    key_thread.start()
    speed_thread.start()

    key_thread.join()
    speed_thread.join()

    physical_port.close()
    virtual_port.close()
    print("\nKeyer stopped.")

if __name__ == "__main__":
    main()