1"""Run without arguments for simulated data; use --port for read-only serial I/O."""23from __future__ import annotations45import argparse6import json7import math8import struct9import sys10import time1112try:13 from .hall_protocol import (BAUD_TABLE, CMD_GET_DATA, CMD_GET_INFO, CMD_RESPONSE, HOST,14 TYPE_ANY, TYPE_ENCODER, TYPE_POSITION, Parser,15 ProtocolError, build_frame, parse_info, parse_sample)16except ImportError:17 from hall_protocol import (BAUD_TABLE, CMD_GET_DATA, CMD_GET_INFO, CMD_RESPONSE, HOST,18 TYPE_ANY, TYPE_ENCODER, TYPE_POSITION, Parser,19 ProtocolError, build_frame, parse_info, parse_sample)202122def make_parser() -> argparse.ArgumentParser:23 parser = argparse.ArgumentParser(description=__doc__)24 source = parser.add_mutually_exclusive_group()25 source.add_argument("--demo", action="store_true", help="simulated data (the default)")26 source.add_argument("--port", help="serial port, for example COM7 or /dev/ttyUSB0")27 parser.add_argument("--id", type=int, default=1, help="device address 1..254")28 parser.add_argument("--baud", type=int, choices=BAUD_TABLE, default=115200)29 parser.add_argument("--profile", choices=("encoder", "line"), default="line",30 help="demo profile only; serial mode discovers the device profile")31 parser.add_argument("--count", type=int, default=20, help="number of samples, 0 means until Ctrl+C")32 parser.add_argument("--interval-ms", type=float, default=100.0,33 help="delay after each sample, not a guaranteed hardware sample period")34 parser.add_argument("--timeout", type=float, default=0.4, help="serial transaction timeout in seconds")35 return parser363738def emit(sample, demo: bool) -> None:39 row = sample.to_dict()40 row["source"] = "simulated_not_hardware" if demo else "serial"41 print(json.dumps(row, ensure_ascii=False), flush=True)424344def run_demo(args) -> int:45 profile = TYPE_ENCODER if args.profile == "encoder" else TYPE_POSITION46 index = 047 while args.count == 0 or index < args.count:48 # Feed the wire parser too, so this path exercises the actual byte format.49 theta = index * 0.1550 value = int(round((math.degrees(theta) % 360) * 100)) if profile == TYPE_ENCODER else int(round(800 * math.sin(theta)))51 if profile == TYPE_POSITION:52 position_mm = value / 100.053 adc = tuple(int(round(2048 + 1000 * math.exp(-((position_mm - spot) ** 2) / (2 * 8.0 ** 2))))54 for spot in (-10.0, 0.0, 10.0))55 else:56 adc = tuple(int(round(2048 + 1000 * math.cos(theta + shift)))57 for shift in (0, -2 * math.pi / 3, 2 * math.pi / 3))58 payload = b"\x00" + struct.pack("<ihHHHB", value, 1000, *adc, 0)59 wire = build_frame(HOST, args.id, profile, CMD_GET_DATA | CMD_RESPONSE, index & 255, payload)60 frame = Parser().feed(wire)[0]61 emit(parse_sample(frame.payload, profile), demo=True)62 index += 163 if args.count == 0 or index < args.count:64 time.sleep(args.interval_ms / 1000.0)65 return 0666768def transact(port, parser: Parser, dev_id: int, dev_type: int,69 cmd: int, seq: int, timeout: float):70 deadline = time.monotonic() + timeout71 wire = build_frame(dev_id, HOST, dev_type, cmd, seq)72 if port.write(wire) != len(wire):73 raise ProtocolError("serial write ended before the complete request was queued")74 # Waiting for a reply already establishes transmission completion. Avoid75 # flush(), whose driver-level drain may outlive a transaction deadline.76 while (remaining := deadline - time.monotonic()) > 0:77 port.timeout = min(0.02, remaining)78 chunk = port.read(min(port.in_waiting or 1, 256))79 received_at = time.monotonic()80 if received_at > deadline:81 break82 for frame in parser.feed(chunk, now=received_at):83 if (frame.dst == HOST and frame.src == dev_id and84 frame.cmd == (cmd | CMD_RESPONSE) and frame.seq == seq and85 (dev_type == TYPE_ANY or frame.dev_type == dev_type)):86 return frame87 raise TimeoutError(f"device {dev_id}, command 0x{cmd:02x}: no matching reply; no sample emitted")888990def run_serial(args) -> int:91 try:92 import serial93 except ImportError:94 print("Serial mode needs pyserial: python -m pip install pyserial", file=sys.stderr)95 return 296 try:97 with serial.Serial(args.port, baudrate=args.baud, bytesize=8, parity="N",98 stopbits=1, timeout=min(0.02, args.timeout),99 write_timeout=args.timeout) as port:100 parser = Parser()101 info_frame = transact(port, parser, args.id, TYPE_ANY, CMD_GET_INFO, 1, args.timeout)102 info = parse_info(info_frame.payload)103 if info["dev_id"] != args.id or info_frame.dev_type != info["dev_type"]:104 raise ProtocolError("information payload and frame identity disagree")105 if BAUD_TABLE[info["baud_index"]] != args.baud:106 raise ProtocolError("reported baud does not match this serial connection")107 print(json.dumps({"device_info": info, "source": "serial"}, ensure_ascii=False), file=sys.stderr)108 index = 0109 while args.count == 0 or index < args.count:110 frame = transact(port, parser, args.id, info["dev_type"], CMD_GET_DATA,111 (index + 2) & 255, args.timeout)112 emit(parse_sample(frame.payload, info["dev_type"]), demo=False)113 index += 1114 if args.count == 0 or index < args.count:115 time.sleep(args.interval_ms / 1000.0)116 except (serial.SerialException, OSError, TimeoutError, ProtocolError) as exc:117 print(f"Stopped: {exc}", file=sys.stderr)118 return 2119 return 0120121122def main() -> int:123 parser = make_parser()124 args = parser.parse_args()125 if not 1 <= args.id <= 254:126 parser.error("--id must be in 1..254")127 if (args.count < 0 or not math.isfinite(args.interval_ms) or args.interval_ms < 0 or128 not math.isfinite(args.timeout) or args.timeout <= 0):129 parser.error("timeout must be finite and positive; count and interval must be finite and nonnegative")130 try:131 return run_serial(args) if args.port else run_demo(args)132 except KeyboardInterrupt:133 return 0134135136if __name__ == "__main__":137 raise SystemExit(main())
元驭机器人Hall 传感器教学下载原文件