Send batch commands from a JSON file.
(api: SerialStudioAPI, args)
| 864 | |
| 865 | |
| 866 | def cmd_batch(api: SerialStudioAPI, args) -> int: |
| 867 | """Send batch commands from a JSON file.""" |
| 868 | try: |
| 869 | with open(args.file, "r") as f: |
| 870 | data = json.load(f) |
| 871 | except FileNotFoundError: |
| 872 | print(f"[ERROR] File not found: {args.file}", file=sys.stderr) |
| 873 | return 1 |
| 874 | except json.JSONDecodeError as e: |
| 875 | print(f"[ERROR] Invalid JSON in file: {e}", file=sys.stderr) |
| 876 | return 1 |
| 877 | |
| 878 | if isinstance(data, list): |
| 879 | commands = data |
| 880 | elif isinstance(data, dict) and "commands" in data: |
| 881 | commands = data["commands"] |
| 882 | else: |
| 883 | print( |
| 884 | '[ERROR] Expected array of commands or {"commands": [...]}', file=sys.stderr |
| 885 | ) |
| 886 | return 1 |
| 887 | |
| 888 | response = api.send_batch(commands) |
| 889 | |
| 890 | if response is None: |
| 891 | print("[ERROR] No response received", file=sys.stderr) |
| 892 | return 1 |
| 893 | |
| 894 | if args.json: |
| 895 | print(json.dumps(response, indent=2)) |
| 896 | else: |
| 897 | results = response.get("results", []) |
| 898 | success_count = sum(1 for r in results if r.get("success")) |
| 899 | print( |
| 900 | f"Executed {len(results)} commands: {success_count} succeeded, " |
| 901 | f"{len(results) - success_count} failed" |
| 902 | ) |
| 903 | for i, result in enumerate(results): |
| 904 | status = chr(10003) if result.get("success") else chr(10007) |
| 905 | cmd_id = result.get("id", f"#{i+1}") |
| 906 | if result.get("success"): |
| 907 | print(f" {status} {cmd_id}") |
| 908 | else: |
| 909 | err = result.get("error", {}) |
| 910 | print( |
| 911 | f" {status} {cmd_id}: {err.get('code')} - " f"{err.get('message')}" |
| 912 | ) |
| 913 | |
| 914 | return 0 if response.get("success") else 1 |
| 915 | |
| 916 | |
| 917 | def _print_command_list(commands: list[dict], use_color: bool): |
no test coverage detected