file transfer script cleanup + automatic EOT detection

This commit is contained in:
flowenol 2023-01-08 01:31:35 +01:00
parent 78a9792692
commit d53257f53b

View File

@ -4,10 +4,12 @@ import select
import signal import signal
import time import time
bytes_processed: int = 0
def print_usage(): def print_usage_and_exit():
print('Usage: python serialfile.py -in <serial device> > file', file=sys.stderr) print('Usage: python serialfile.py -in <serial device> [ > file ]', file=sys.stderr)
print('Usage: python serialfile.py -out <serial device> < file', file=sys.stderr) print('Usage: python serialfile.py -out <serial device> < file', file=sys.stderr)
exit(1)
def get_serial(port: str) -> serial.Serial: def get_serial(port: str) -> serial.Serial:
@ -24,6 +26,44 @@ def get_serial(port: str) -> serial.Serial:
def data_ready_on_stdin(): def data_ready_on_stdin():
return bool(select.select([sys.stdin], [], [], 0)[0]) return bool(select.select([sys.stdin], [], [], 0)[0])
def data_ready_on_port(port: serial.Serial):
return bool(select.select([port.fileno()], [], [], 0)[0])
def handle_inbound_transmission(port: str):
global bytes_processed
with get_serial(port) as s:
print('Waiting for data... ', file=sys.stderr, end='', flush=True)
while not data_ready_on_port(s):
pass
print('OK', file=sys.stderr)
s.timeout = 1
while True:
data = s.read()
if len(data) > 0:
sys.stdout.buffer.write(data)
bytes_processed += len(data)
else:
break
print('Transmission finished', file=sys.stderr)
def handle_outbound_transmission(port: str):
global bytes_processed
if not data_ready_on_stdin():
print_usage_and_exit()
with get_serial(port) as s:
data = sys.stdin.buffer.read()
while len(data) > 0:
s.write(data)
bytes_processed += len(data)
data = sys.stdin.buffer.read()
""" """
This utility receives/sends files via selected serial port to the This utility receives/sends files via selected serial port to the
Apple-1 Serial Interface card. Apple-1 Serial Interface card.
@ -35,28 +75,23 @@ pip3 install pyserial
if __name__ == "__main__": if __name__ == "__main__":
if len(sys.argv) != 3: if len(sys.argv) != 3:
print_usage() print_usage_and_exit()
exit(1)
def print_bytes_processed_and_exit(status: int):
print(f'Total bytes processed: {bytes_processed}', file=sys.stderr)
exit(status)
# terminate silently on SIGINT
signal.signal(signal.SIGINT, lambda s, fr: print_bytes_processed_and_exit(1))
if sys.argv[1] == '-in': if sys.argv[1] == '-in':
print('Press Ctrl+C after the transmission ends', file=sys.stderr) handle_inbound_transmission(sys.argv[2])
# terminate silently on SIGINT
signal.signal(signal.SIGINT, lambda s, fr: exit(0))
with get_serial(sys.argv[2]) as s:
while True:
sys.stdout.buffer.write(s.read())
elif sys.argv[1] == '-out': elif sys.argv[1] == '-out':
# check if the data is available on stdin handle_outbound_transmission(sys.argv[2])
if not data_ready_on_stdin():
exit(0)
with get_serial(sys.argv[2]) as s:
data = sys.stdin.buffer.read()
while len(data) > 0:
s.write(data)
data = sys.stdin.buffer.read()
else: else:
print(f'Unknown option: {sys.argv[1]}', file=sys.stderr) print(f'Unknown option: {sys.argv[1]}', file=sys.stderr)
print_usage() print_usage_and_exit()
exit(1)
print_bytes_processed_and_exit(0)