sw: add console script prototype
This commit is contained in:
94
software/console.py
Normal file
94
software/console.py
Normal file
@ -0,0 +1,94 @@
|
||||
import argparse
|
||||
import socket
|
||||
|
||||
|
||||
def run_debug(args, sock):
|
||||
"""Debug run: send fixed values to test eth+ctrl on fpga."""
|
||||
print(f"DEBUG MODE: ip={args.ip} send_port={args.send_port}")
|
||||
|
||||
dest = (args.ip, args.send_port)
|
||||
|
||||
# reset
|
||||
sock.sendto(0x0f00.to_bytes(2), dest)
|
||||
print("Sent soft_reset!")
|
||||
|
||||
# config data
|
||||
sock.sendto(format_ctrl_data(0x12345678, 0x9abcdef0,
|
||||
0x0bea, 0xdead, dac_bits=args.dac_bits), dest)
|
||||
print("Config data sent!")
|
||||
|
||||
sock.sendto(0xf000.to_bytes(2), dest)
|
||||
print("Sent start!")
|
||||
|
||||
|
||||
def format_ctrl_data(pulse_width: int, pulse_period: int,
|
||||
pulse_height: int, pulse_num: int, dac_bits: int = 16) -> bytes:
|
||||
"""Format data packet for set_data command."""
|
||||
output = bytearray()
|
||||
|
||||
output += 0b10001000.to_bytes(1, 'little')
|
||||
|
||||
# no negative please
|
||||
assert pulse_width > 0, "pulse_width should be positive"
|
||||
assert pulse_period > 0, "pulse_period should be positive"
|
||||
assert pulse_num > 0, "pulse_num should be positive"
|
||||
assert pulse_height > 0, "pulse_height should be positive"
|
||||
|
||||
# overflow check
|
||||
assert pulse_width < 2**32-1, "pulse_width too high"
|
||||
assert pulse_period < 2**32-1, "pulse_period too high"
|
||||
assert pulse_num < 2**16-1, "pulse_num too high"
|
||||
assert pulse_height < 2**dac_bits-1, "pulse_height too high"
|
||||
|
||||
output += pulse_width.to_bytes(4, 'little')
|
||||
output += pulse_period.to_bytes(4, 'little')
|
||||
output += pulse_num.to_bytes(2, 'little')
|
||||
output += pulse_height.to_bytes(2, 'little')
|
||||
|
||||
assert len(output) == 13, "Config data should be 96 bits + 8 bit header"
|
||||
return output
|
||||
|
||||
|
||||
def run(args, sock):
|
||||
pass
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Консоль для рефлектометра"
|
||||
)
|
||||
|
||||
parser.add_argument("--debug", action='store_true',
|
||||
help="отладочная отправка пакета soft_reset, пакета с данными и пакета start")
|
||||
|
||||
parser.add_argument("--ip", type=str, default="192.168.0.2",
|
||||
help="IP рефлектометра, по умолчанию 192.168.0.2")
|
||||
|
||||
parser.add_argument("--send-port", type=int, default=8080,
|
||||
help="Порт для отправки команд")
|
||||
|
||||
parser.add_argument("--recv-port", type=int,
|
||||
default=8080, help="Порт для приема данных")
|
||||
|
||||
parser.add_argument("--dac-bits", type=int, default=12,
|
||||
help="Битность ЦАП (влияет на максимальный pulse_height)")
|
||||
|
||||
# передача параметров через аргументы
|
||||
for arg in ("pulse_width", "pulse_period", "pulse_num", "pulse_height"):
|
||||
parser.add_argument(f"--{arg}", type=int,
|
||||
default=0, help=f"Задать {arg}")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
|
||||
if args.debug:
|
||||
run_debug(args, sock)
|
||||
else:
|
||||
run(args, sock)
|
||||
|
||||
sock.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user