#!/usr/bin/env python3 import argparse import json import pathlib import requests import shutil import subprocess import sys import uuid from datetime import datetime from argparse import ArgumentDefaultsHelpFormatter, RawTextHelpFormatter import re class SmartFormatter(ArgumentDefaultsHelpFormatter, RawTextHelpFormatter): pass VERSION = "1.11.1" CLIENT_ID_FILE = pathlib.Path(__file__).parent / ".client-id" REGION_FILE = pathlib.Path(__file__).parent / ".region" AWS_REGIONS = { "us-east-1", "us-east-2", "us-west-1", "us-west-2", "af-south-1", "ap-east-1", "ap-south-1", "ap-south-2", "ap-southeast-1", "ap-southeast-2", "ap-southeast-3", "ap-southeast-4", "ap-northeast-1", "ap-northeast-2", "ap-northeast-3", "ca-central-1", "eu-central-1", "eu-central-2", "eu-north-1", "eu-south-1", "eu-south-2", "eu-west-1", "eu-west-2", "eu-west-3", "il-central-1", "me-central-1", "me-south-1", "sa-east-1", } def format_time(timestr: str) -> str: """Convert ISO8601 string to YYYY-MM-DD HH:MM:SS (UTC). Returns '' if input is empty or invalid.""" if not timestr: return "" try: dt = datetime.fromisoformat(timestr.replace("Z", "+00:00")) return dt.strftime("%Y-%m-%d %H:%M:%S") except Exception: return timestr # fallback: show raw def print_executions_table(title: str, executions: list[dict]): if not executions: print(f"{title}: none") return print(f"{title}:") header = ( f"{'Execution ID':<36} {'Start Time (UTC)':<19} {'Stop Time (UTC)':<19}" ) print(header) print("-" * len(header)) for exe in executions: print( f"{exe['execution_id']:<36} {format_time(exe['startTime']):<19} {format_time(exe['stopTime']):<19}" ) print() def parse_s3_path(s3_path: str): """ To avoid weird errors from AWS Step Functions, and SSMs, we validate the path allowedPattern: "^s3://[a-zA-Z0-9\\-_/]+$" The reason to add this function is to avoid characters like `.` or other invalid characters in the path that lead to hanging step functions and confusing errors later on. Returns a tuple (is_valid: bool, message_or_path: str) """ pattern = r"^s3://[a-zA-Z0-9\-_/]+$" if not re.match(pattern, s3_path): return ( False, f"s3-path '{s3_path}' is not valid. It must match the pattern {pattern}", ) return True, s3_path def validate_framewise_trinsics(s3_path: str, trinsics_name="extrinsics") -> str: """Validate that extrinsics or intrinsics files exist for all topbot images when using --framewise-extrinsics or --framewise-intrinsics flags. Returns error message if validation fails, empty string otherwise.""" if not shutil.which("aws"): return "" # Skip validation if AWS CLI not available try: # List topbot files topbot_prefix = s3_path.rstrip("/") + "/topbot/" try: topbot_output = subprocess.check_output( ["aws", "s3", "ls", topbot_prefix], text=True, stderr=subprocess.PIPE, ) except subprocess.CalledProcessError: return f"Topbot folder not found at {topbot_prefix}" topbot_files = [ line.split()[-1] for line in topbot_output.strip().split("\n") if line and line.split()[-1].endswith(".tiff") ] if not topbot_files: return f"No topbot images found in {topbot_prefix}" # List trinsics files trinsics_prefix = s3_path.rstrip("/") + f"/{trinsics_name}/" try: trinsics_output = subprocess.check_output( ["aws", "s3", "ls", trinsics_prefix], text=True, stderr=subprocess.PIPE, ) trinsics_files = [ line.split()[-1] for line in trinsics_output.strip().split("\n") if line and line.split()[-1].endswith(".yaml") ] except subprocess.CalledProcessError: # Extrinsics folder doesn't exist or is empty return f"{trinsics_name.capitalize()} folder not found at {trinsics_prefix}" # Check that each topbot has a corresponding trinsics file missing = [] for topbot in topbot_files: frame_id = topbot.replace(".tiff", "") trinsics_file = f"{frame_id}.yaml" if trinsics_file not in trinsics_files: missing.append(frame_id) if missing: return f"Missing {trinsics_name} files for {len(missing)} frame(s): {', '.join(missing)}" return "" except subprocess.CalledProcessError: return "" # Skip validation if topbot listing fails except Exception as e: print(f"Warning: could not validate framewise {trinsics_name}: {e}") return "" def validate_client_id(client_id: str): try: val = uuid.UUID(client_id, version=4) except ValueError: return False, f"{client_id} is not a valid UUID v4" if str(val) != client_id: return False, f"{client_id} is not in canonical UUID v4 format" return True, client_id def validate_target_subdomain(subdomain: str): if subdomain in AWS_REGIONS: return True, subdomain if subdomain.startswith("dev-") or subdomain.startswith("test-"): suffix = "-".join(subdomain.split("-")[-3:]) if suffix in AWS_REGIONS: return True, subdomain return False, f"{subdomain} is not a valid AWS region or dev/test prefixed region" def get_or_prompt(file_path: pathlib.Path, arg_value: str, prompt: str, validate_func): if arg_value: ok, result = validate_func(arg_value) if not ok: return None, result file_path.write_text(result) return result, None if file_path.exists(): value = file_path.read_text().strip() ok, result = validate_func(value) if not ok: return None, result return result, None value = input(f"{prompt}: ").strip() ok, result = validate_func(value) if not ok: return None, result file_path.write_text(result) return result, None def get_client_and_region(client_id, region): client_id, err = get_or_prompt( CLIENT_ID_FILE, client_id, "Enter client-id (UUID v4)", validate_client_id ) if not client_id: print(f"Error: {err}") return None, None, 1 region, err = get_or_prompt( REGION_FILE, region, "Enter region", validate_target_subdomain ) if not region: print(f"Error: {err}") return None, None, 1 return client_id, region, 0 def build_base_url(subdomain: str) -> str: return f"https://cloud.{subdomain}.nodarsensor.net" def print_error(url, response): print(f"Error: request to {url} failed with status {response.status_code}") try: data = response.json() if isinstance(data, dict) and "body" in data: print("\n\t", data["body"]) else: print("\n\t", data) except Exception: print("\n\t", response.text) if response.status_code == 400 and response.text == "Bad Request": print("\nIt is likely that either your UUID or region is incorrect.") FLAGS = { "common": [ "--save-left-disparity", "--save-right-disparity", "--save-left-rectified", "--save-right-rectified", "--save-left-valid-pixel-map", "--save-right-valid-pixel-map", "--save-details", "--save-pc", "--disable-autocal", "--framewise-extrinsics", "--framewise-intrinsics", ], "ground_truth_only": [], "hammerhead_only": ["--save-left-confidence-map", "--save-right-confidence-map"], "unsupported": [], } def generate_flags_help() -> str: def fmt(name, flags): return f" {name}:\n " + "\n ".join(flags) if flags else None sections = [ fmt("Common", FLAGS["common"]), fmt("Ground Truth Only", FLAGS["ground_truth_only"]), fmt( "Hammerhead Only (since ground-truth does not generate confidence maps)", FLAGS["hammerhead_only"], ), ] sections = [s for s in sections if s] # drop empty return ( "Processing flags (space-separated).\n" + "\n".join(f" {line}" for line in sections) + "\n" ) def validate_flags(matcher, flags): """Validate that flags are compatible with the chosen matcher.""" flag_list = flags.split() for flag in flag_list: if flag in FLAGS["unsupported"]: return f"Flag '{flag}' is not implemented" elif flag in FLAGS["ground_truth_only"] and matcher != "ground-truth": return f"Flag '{flag}' is only supported with ground-truth matcher, not {matcher}" elif flag in FLAGS["hammerhead_only"] and matcher != "hammerhead": return f"Flag '{flag}' is only supported with hammerhead matcher, not {matcher}" elif not any(flag in group for group in FLAGS.values()): return f"Unknown flag: '{flag}'" if ( "--framewise-intrinsics" in flag_list and "--framewise-extrinsics" in flag_list and "--disable-autocal" not in flag_list ): return "--framewise-intrinsics and --framewise-extrinsics require --disable-autocal to also be set" return None # No error def resolve_dataset_defaults(raw, defaults): """Fill in any field a manifest entry omits from the CLI-level defaults.""" return { "s3_path": raw.get("s3_path"), "start_frame": str(raw.get("start_frame", defaults["start_frame"])), "frame_count": str(raw.get("frame_count", defaults["frame_count"])), "matcher": raw.get("matcher", defaults["matcher"]), "flags": raw.get("flags", defaults["flags"]), "pixel_format": raw.get("pixel_format", defaults["pixel_format"]), "max_disp": str(raw.get("max_disp", defaults["max_disp"])), "split_network": raw.get("split_network", defaults["split_network"]), } def load_manifest_datasets(manifest_path, defaults): """Returns (datasets, error). datasets is a list of dicts with every field resolved (manifest entries may omit any field except s3_path).""" try: with open(manifest_path) as f: manifest = json.load(f) except Exception as e: return None, f"Could not read manifest file '{manifest_path}': {e}" raw_datasets = manifest.get("datasets") if isinstance(manifest, dict) else None if not isinstance(raw_datasets, list) or not raw_datasets: return None, f"Manifest '{manifest_path}' must contain a non-empty 'datasets' list" if not all(isinstance(d, dict) and d.get("s3_path") for d in raw_datasets): return None, "Each entry in 'datasets' must be an object with an 's3_path'" return [resolve_dataset_defaults(d, defaults) for d in raw_datasets], None # --------------------------------------------------------------------------- # "Results already exist" pre-check (used by `start` unless --force-reprocess). # # Results land in S3 at: # {s3_path}/{matcher_dir}/executions/{execution_id}/{subdir}/{frame:09d}.{ext} # execution_id is unique per run, so a reprocess never overwrites - this check # only avoids redundant GPU cost. It is entirely best-effort `aws s3 ls`: any # inability to read S3 resolves to "proceed with the job" (a genuinely broken # dataset, e.g. missing topbot/, is caught by the Lambda's structure validation). # --------------------------------------------------------------------------- # Maps each output-producing --save-* flag to the S3 subdirectory the container # writes it to. MUST stay in sync with IMAGE_OUTPUT_DIRS / OTHER_OUTPUT_DIRS in # src/nodar-gt/src/nodargt/data_repository.py. Flags absent here (--disable-autocal, # --framewise-*) produce no output and are ignored by the pre-check. FLAG_OUTPUT_DIRS = { "--save-left-disparity": "disparity", "--save-right-disparity": "right-disparity", "--save-left-rectified": "left-rect", "--save-right-rectified": "right-rect", "--save-left-confidence-map": "left-confidence-map", "--save-right-confidence-map": "right-confidence-map", "--save-left-valid-pixel-map": "left-valid-pixel-map", "--save-right-valid-pixel-map": "right-valid-pixel-map", "--save-details": "details", "--save-pc": "point_clouds", } # prior_run_status() states RUN_NONE = "NONE" # no prior attempt found (also the fail-open result) -> run RUN_COMPLETE = "COMPLETE" # a prior execution fully covers the request -> block RUN_INCOMPLETE = "INCOMPLETE" # a prior execution is partial/crashed -> block (loudly) def matcher_output_dir(matcher: str) -> str: """Top-level S3 output directory for a matcher (mirrors DataRepository).""" return "nodar-hh" if matcher == "hammerhead" else "nodar-gt" def flags_to_output_subdirs(flags: str) -> set: """The set of S3 output subdirs the requested --save-* flags will produce.""" return {FLAG_OUTPUT_DIRS[f] for f in flags.split() if f in FLAG_OUTPUT_DIRS} def _run_aws_ls(target: str, recursive: bool = False): """Run `aws s3 ls ` and return stdout text, or None on any failure (aws missing, no creds, prefix absent, transient error). None always means 'could not determine' and resolves to proceeding with the job.""" if not shutil.which("aws"): return None cmd = ["aws", "s3", "ls", target] if recursive: cmd.append("--recursive") try: return subprocess.check_output(cmd, text=True, stderr=subprocess.PIPE) except Exception: return None def list_topbot_frame_ids(s3_path: str): """Sorted list of integer frame ids from {s3_path}/topbot/*.tiff, or None if the folder cannot be listed.""" out = _run_aws_ls(f"{s3_path.rstrip('/')}/topbot/") if out is None: return None ids = [] for line in out.strip().split("\n"): tokens = line.split() if not tokens or not tokens[-1].endswith(".tiff"): continue try: ids.append(int(tokens[-1][: -len(".tiff")])) except ValueError: continue return sorted(ids) def list_execution_ids(s3_path: str, matcher_dir: str): """Names of prior execution folders under {s3_path}/{matcher_dir}/executions/. Returns [] if the prefix does not exist or cannot be listed.""" out = _run_aws_ls(f"{s3_path.rstrip('/')}/{matcher_dir}/executions/") if out is None: return [] ids = [] for line in out.strip().split("\n"): tokens = line.split() if len(tokens) >= 2 and tokens[0] == "PRE": ids.append(tokens[-1].rstrip("/")) return ids def execution_present_frames(s3_path: str, matcher_dir: str, exec_id: str) -> dict: """Return {subdir: set(frame_ids)} for one prior execution folder, via a single recursive listing. {} on any listing failure.""" prefix = f"{s3_path.rstrip('/')}/{matcher_dir}/executions/{exec_id}/" out = _run_aws_ls(prefix, recursive=True) present: dict = {} if out is None: return present marker = f"executions/{exec_id}/" for line in out.strip().split("\n"): tokens = line.split() if not tokens: continue key = tokens[-1] idx = key.find(marker) if idx < 0: continue parts = key[idx + len(marker):].split("/") if len(parts) < 2: continue subdir, filename = parts[0], parts[-1] try: fid = int(filename.rsplit(".", 1)[0]) except ValueError: continue present.setdefault(subdir, set()).add(fid) return present def expected_frame_ids(topbot_ids, start_frame: int, frame_count: int) -> set: """Frame ids the container would process: topbot ids with id >= start_frame (sorted), truncated to the first frame_count of them (frame_count <= 0 = all). Mirrors the start_frame/frame_count logic in nodargt.__main__.""" eligible = sorted(fid for fid in topbot_ids if fid >= start_frame) if frame_count > 0: eligible = eligible[:frame_count] return set(eligible) def prior_run_status(dataset: dict): """Best-effort check of whether `dataset` was already processed. Returns (state, message) with state one of RUN_NONE / RUN_COMPLETE / RUN_INCOMPLETE. Any inability to read S3 resolves to RUN_NONE ('proceed').""" s3_path = dataset["s3_path"].rstrip("/") requested = flags_to_output_subdirs(dataset["flags"]) if not requested: return RUN_NONE, "" # no output-producing flags -> nothing to detect matcher_dir = matcher_output_dir(dataset["matcher"]) exec_ids = list_execution_ids(s3_path, matcher_dir) if not exec_ids: return RUN_NONE, "" # no prior executions (or S3 unreadable) -> run topbot_ids = list_topbot_frame_ids(s3_path) if not topbot_ids: return RUN_NONE, "" # can't compute expected frames -> run (Lambda validates topbot/) try: expected = expected_frame_ids( topbot_ids, int(dataset["start_frame"]), int(dataset["frame_count"]) ) except (ValueError, KeyError): return RUN_NONE, "" if not expected: return RUN_NONE, "" # no frames would be processed -> run best_incomplete = None # (coverage_score, exec_id, present) for exec_id in exec_ids: present = execution_present_frames(s3_path, matcher_dir, exec_id) if not any(present.get(sd) for sd in requested): continue # this execution has none of the requested outputs if all(expected <= present.get(sd, set()) for sd in requested): msg = ( f"execution {exec_id} already has all requested outputs " f"({', '.join(sorted(requested))}) for all {len(expected)} frame(s)" ) return RUN_COMPLETE, msg score = sum(len(present.get(sd, set()) & expected) for sd in requested) if best_incomplete is None or score > best_incomplete[0]: best_incomplete = (score, exec_id, present) if best_incomplete is None: return RUN_NONE, "" _, exec_id, present = best_incomplete total = len(expected) lines = [f"an existing run appears INCOMPLETE: execution {exec_id}"] for sd in sorted(requested): have = present.get(sd, set()) & expected missing = expected - have if missing: lines.append(f" {sd}/: {len(have)}/{total} frame(s) (first missing frame {min(missing):09d})") else: lines.append(f" {sd}/: {len(have)}/{total} frame(s) (complete)") return RUN_INCOMPLETE, "\n".join(lines) def start( client_id, region, s3_path, start_frame, frame_count, matcher, flags, pixel_format, max_disp, split_network, force_reprocess=False, ): """ Start an execution. `s3_path` is either a single s3://... dataset path, or a path to a local JSON manifest file listing multiple datasets to process sequentially in one job (see load_manifest_datasets). Either way, this always submits a `datasets` list to the API - a single dataset is just a batch of one. All other arguments are batch-wide defaults: for a manifest run, any field a dataset entry omits falls back to the corresponding argument here. Returns a process exit code (0 on success, 1 on any validation or request failure). """ client_id, region, exit_code = get_client_and_region(client_id, region) if exit_code != 0: return exit_code defaults = { "start_frame": start_frame, "frame_count": frame_count, "matcher": matcher, "flags": flags, "pixel_format": pixel_format, "max_disp": max_disp, "split_network": "1" if split_network else "0", } if s3_path.startswith("s3://"): datasets = [resolve_dataset_defaults({"s3_path": s3_path}, defaults)] elif pathlib.Path(s3_path).is_file(): datasets, manifest_error = load_manifest_datasets(s3_path, defaults) if manifest_error: print(f"Error: {manifest_error}") return 1 else: print( f"Error: '{s3_path}' is neither an s3:// path nor an existing local manifest (.json) file." ) return 1 # A single dataset (the common case) gets unlabeled messages, same as before # batch support existed; a real batch gets each dataset labeled by index. multi = len(datasets) > 1 for index, dataset in enumerate(datasets): label = f"Dataset {index} ({dataset['s3_path']})" if multi else "Dataset" try: max_disp_int = int(dataset["max_disp"]) except ValueError: print(f"Error: {label}: --max-disp must be an integer (got '{dataset['max_disp']}')") return 1 if max_disp_int != 0 and max_disp_int % 32 != 0: down = (max_disp_int // 32) * 32 up = down + 32 print(f"Error: {label}: --max-disp must be a multiple of 32 (got {max_disp_int}). Try {down} or {up}.") return 1 if max_disp_int > 608: print(f"Error: {label}: --max-disp must be at most 608 (got {max_disp_int}).") return 1 flag_error = validate_flags(dataset["matcher"], dataset["flags"]) if flag_error: print(f"Error: {label}: {flag_error}") return 1 if not dataset["s3_path"].startswith("s3://"): print(f"Error: {label}: s3_path must start with s3:// (got '{dataset['s3_path']}')") return 1 if not parse_s3_path(dataset["s3_path"])[0]: print( f"Error: {label}: An S3 path may only contain letters, numbers, hyphens, underscores, " "and slashes -- no `.` or spaces" ) return 1 if "--framewise-extrinsics" in dataset["flags"]: print(f"{label}: Validating framewise extrinsics...") extrinsics_error = validate_framewise_trinsics( dataset["s3_path"], trinsics_name="extrinsics" ) if extrinsics_error: print(f"Error: {label}: {extrinsics_error}") return 1 print(f"{label}: All topbot images have corresponding extrinsics files") if "--framewise-intrinsics" in dataset["flags"]: print(f"{label}: Validating framewise intrinsics...") intrinsics_error = validate_framewise_trinsics( dataset["s3_path"], trinsics_name="intrinsics" ) if intrinsics_error: print(f"Error: {label}: {intrinsics_error}") return 1 print(f"{label}: All topbot images have corresponding intrinsics files") # Skip datasets that already have results in S3, unless --force-reprocess. # A partial/crashed prior run (INCOMPLETE) is flagged loudly and also blocked, # so it gets a human's attention rather than being silently reprocessed. if not force_reprocess: remaining = [] blocked_incomplete = False for index, dataset in enumerate(datasets): label = f"Dataset {index} ({dataset['s3_path']})" if multi else "Dataset" state, message = prior_run_status(dataset) if state == RUN_COMPLETE: print(f"{label}: results already exist -- {message}") print(f"{label}: skipping (pass --force-reprocess to reprocess anyway).") elif state == RUN_INCOMPLETE: blocked_incomplete = True print(f"\n⚠️ {label}: {message}") print( f"{label}: NOT reprocessing automatically. Inspect the run above, " "then pass --force-reprocess to reprocess.\n" ) else: remaining.append(dataset) if not remaining: print( "Nothing to submit: all requested datasets already have results " "(or an incomplete prior run). Use --force-reprocess to override." ) return 3 if blocked_incomplete else 0 datasets = remaining url = f"{build_base_url(region)}/start-execution" headers = {"Customer-ID": client_id} try: response = requests.post(url, headers=headers, json={"datasets": datasets}) except Exception as e: print(f"Error: could not connect to {url} - {e}") return 1 if response.ok: try: data = response.json() if msg := data.get("message", ""): print(msg) print(f"Started process with execution ID: {data.get('executionId')}") except Exception: print("Warning: response body was not valid JSON") else: print_error(url, response) return 0 if response.ok else 1 def stop(client_id, region, execution_id): client_id, region, exit_code = get_client_and_region(client_id, region) if exit_code != 0: return exit_code url = f"{build_base_url(region)}/stop-execution" headers = { "Customer-ID": client_id, "Execution-ID": execution_id, } try: response = requests.post(url, headers=headers) except Exception as e: print(f"Error: could not connect to {url} - {e}") return 1 if response.ok: print(f"Submitted stop request for process with execution ID: {execution_id}") else: print_error(url, response) return 0 if response.ok else 1 def principal(client_id, region): client_id, region, exit_code = get_client_and_region(client_id, region) if exit_code != 0: return exit_code url = f"{build_base_url(region)}/get-principal" headers = { "Customer-ID": client_id, } try: response = requests.get(url, headers=headers) except Exception as e: print(f"Error: could not connect to {url} - {e}") return 1 if response.ok: try: data = response.json() print(f"Roles that need access to your S3 bucket:\n") print(f"EC2_ROLE_ARN: {data.get('ec2RoleArn')}") except Exception: print("Warning: response body was not valid JSON") else: print_error(url, response) return 0 if response.ok else 1 def format_duration(seconds): """Format a duration in seconds as e.g. '1h 23m 4s'. Returns 'unknown' for None.""" if seconds is None: return "unknown" seconds = int(round(seconds)) hours, rem = divmod(seconds, 3600) minutes, secs = divmod(rem, 60) parts = [] if hours: parts.append(f"{hours}h") if hours or minutes: parts.append(f"{minutes}m") parts.append(f"{secs}s") return " ".join(parts) def print_progress(progress): """Print one dataset's progress/ETA, as returned in /get-status's "progress" field while an execution is RUNNING (see StatusReporter in the nodargt package for how this payload is produced).""" frame_index = progress.get("frame_index", 0) total_frames = progress.get("total_frames", 0) dataset_index = progress.get("dataset_index", 0) dataset_count = progress.get("dataset_count", 1) phase = progress.get("phase", "unknown") eta_seconds = progress.get("eta_seconds") pct = f"{(frame_index / total_frames * 100):.1f}%" if total_frames else "?" if dataset_count > 1: print(f" Dataset {dataset_index + 1}/{dataset_count}") print(f" Phase: {phase}, frame {frame_index}/{total_frames} ({pct})") print(f" ETA: {format_duration(eta_seconds)}") def status(client_id, region, execution_id): """ Print the status of an execution (and frame progress/ETA, if it's RUNNING and has written a status file yet). Returns a process exit code: 0 on SUCCEEDED, 2 if the execution FAILED because the EC2 instance or its container never started (reason="failed_to_start" - often a transient capacity issue, worth retrying), 1 for any other FAILED execution or request-level failure. """ client_id, region, exit_code = get_client_and_region(client_id, region) if exit_code != 0: return exit_code url = f"{build_base_url(region)}/get-status" headers = { "Customer-ID": client_id, "Execution-ID": execution_id, } try: response = requests.get(url, headers=headers) except Exception as e: print(f"Error: could not connect to {url} - {e}") return 1 if not response.ok: print_error(url, response) return 1 try: data = response.json() except Exception: print("Warning: response body was not valid JSON") return 1 job_status = data.get("status") reason = data.get("reason") print(f"Status of {execution_id}: {job_status}") progress = data.get("progress") if progress: print_progress(progress) if job_status != "FAILED": return 0 # Show error details if available error = data.get("error") cause = data.get("cause") if reason == "failed_to_start": print("\nThe job never started: the EC2 instance or its container failed to come up.") print("This may be a transient capacity issue - it's often worth retrying.") elif error: print("\nPlease report this error to support@nodarsensor.com") print("It is likely an internal AWS cloud error.\n") print(f"Error: {error}") if cause: print(f"Cause: {cause}") return 2 if reason == "failed_to_start" else 1 def cloud_version(client_id, region): """ Query the API /cloud-version endpoint and print the deployed version. """ client_id, region, exit_code = get_client_and_region(client_id, region) if exit_code != 0: return exit_code url = f"{build_base_url(region)}/cloud-version" headers = {"Customer-ID": client_id} try: response = requests.get(url, headers=headers) except Exception as e: print(f"Error: could not connect to {url} - {e}") return 1 if response.ok: try: data = response.json() print(f"Cloud API version: {data.get('VERSION')}") except Exception: print("Warning: response body was not valid JSON") else: print_error(url, response) return 0 if response.ok else 1 def list_executions(client_id, region, endpoint, label): client_id, region, exit_code = get_client_and_region(client_id, region) if exit_code != 0: return exit_code url = f"{build_base_url(region)}/{endpoint}" headers = {"Customer-ID": client_id} try: response = requests.get(url, headers=headers) except Exception as e: print(f"Error: could not connect to {url} - {e}") return 1 if response.ok: try: executions = response.json() # Always print in table form print_executions_table(f"{label.capitalize()} executions", executions) except Exception: print("Warning: response body was not valid JSON") else: print_error(url, response) return 0 if response.ok else 1 def add_common_args(p): p.add_argument("--client-id", metavar="ID", help="Customer UUID v4") p.add_argument("--region", help="AWS Region (e.g. us-east-1)") def main(): parser = argparse.ArgumentParser( description="Nodar Cloud CLI", formatter_class=SmartFormatter, ) parser.add_argument( "-v", "--version", action="version", version=f"%(prog)s {VERSION}" ) parsers = parser.add_subparsers(dest="command", required=True) add_common_args( parsers.add_parser("cloud-version", help="Show the version of the cloud API") ) start_parser = parsers.add_parser( "start", help="Start execution", description=( "Start execution.\n\n" "s3_path is either a single s3://bucket/prefix dataset, or a path to a\n" "local JSON manifest file listing several datasets to process\n" "sequentially in one job:\n\n" ' {"datasets": [{"s3_path": "s3://bucket/run1/"}, ' '{"s3_path": "s3://bucket/run2/", "matcher": "hammerhead"}]}\n\n' "Every field except s3_path is optional per dataset; whatever a dataset\n" "omits falls back to this command's own flags below (--matcher, --flags,\n" "etc), same as a single-dataset run." ), formatter_class=SmartFormatter, ) add_common_args(start_parser) # s3-path can be passed as positional or as a flag. It may also be a path to a # local JSON manifest file ({"datasets": [{"s3_path": "s3://...", ...}, ...]}) # to process several datasets sequentially in one job; missing per-dataset # fields fall back to this command's own flags (--matcher, --flags, etc). start_parser.add_argument( "s3_path", nargs="?", help="S3 path like s3://bucket/prefix, or a local manifest.json for a batch of datasets", ) start_parser.add_argument( "--s3-path", dest="s3_path_opt", metavar="S3_PATH", help="S3 path like s3://bucket/prefix, or a local manifest.json for a batch of datasets", ) start_parser.add_argument( "--start-frame", metavar="FRAME", default="0", help="Starting frame number. Batch-wide default for any manifest dataset that omits start_frame.", ) start_parser.add_argument( "--frame-count", metavar="COUNT", default="-1", help="Number of frames to process where -1 denotes 'all frames'. Batch-wide default for any manifest dataset that omits frame_count.", ) start_parser.add_argument( "--matcher", default="ground-truth", help="Matcher type: 'ground-truth' or 'hammerhead'. Batch-wide default for any manifest dataset that omits matcher.", ) start_parser.add_argument( "--pixel-format", metavar="FMT", default="BGR", help="Input image pixel format: BGR, Bayer_RGGB, Bayer_GRBG, Bayer_BGGR, Bayer_GBRG. Batch-wide default for any manifest dataset that omits pixel_format.", ) start_parser.add_argument( "--max-disp", metavar="DISP", default="0", help="Maximum disparity. Must be a multiple of 32, up to 608. 0 means use the model default (416). Batch-wide default for any manifest dataset that omits max_disp.", ) start_parser.add_argument( "--split-network", action="store_true", default=False, help="Force the use of the split network (for memory-constrained systems). Normally auto-detected. Batch-wide default for any manifest dataset that omits split_network.", ) start_parser.add_argument( "--flags", default="--save-left-disparity --save-details --save-left-rectified", help=generate_flags_help() + "Batch-wide default for any manifest dataset that omits flags.\n", ) start_parser.add_argument( "--force-reprocess", action="store_true", default=False, help=( "Reprocess even if results already exist in S3 for a dataset. By default,\n" "if a prior execution's outputs are already present -- whether complete or a\n" "partial/crashed run -- that dataset is not submitted, so it can be inspected\n" "first. In a batch, only the not-yet-processed datasets are submitted." ), ) stop_parser = parsers.add_parser("stop", help="Stop execution") add_common_args(stop_parser) # execution_id can be passed as a positional or as a flag stop_parser.add_argument("execution_id", nargs="?", help="Execution ID") stop_parser.add_argument( "--execution-id", dest="execution_id_opt", metavar="XID", help="Execution ID", ) add_common_args( parsers.add_parser( "principal", help="Get principal (specifically, its ARN). Add this to your S3 bucket", ) ) status_parser = parsers.add_parser("status", help="Get status of execution") add_common_args(status_parser) # execution_id can be passed as a positional or as a flag status_parser.add_argument("execution_id", nargs="?", help="Execution ID") status_parser.add_argument( "--execution-id", dest="execution_id_opt", metavar="XID", help="Execution ID" ) add_common_args( parsers.add_parser("running", help="List executions that are running") ) add_common_args( parsers.add_parser("succeeded", help="List executions that succeeded") ) add_common_args( parsers.add_parser("timed-out", help="List executions that timed-out") ) add_common_args( parsers.add_parser("aborted", help="List executions that were aborted") ) add_common_args(parsers.add_parser("failed", help="List executions that failed")) args = parser.parse_args() if args.command == "start": s3_path = args.s3_path or args.s3_path_opt if not s3_path: print("You must provide an s3-path") return 1 return start( client_id=args.client_id, region=args.region, s3_path=s3_path, start_frame=args.start_frame, frame_count=args.frame_count, matcher=args.matcher, flags=args.flags, pixel_format=args.pixel_format, max_disp=args.max_disp, split_network=args.split_network, force_reprocess=args.force_reprocess, ) elif args.command == "stop": execution_id = args.execution_id or args.execution_id_opt if not execution_id: print("You must provide an execution ID") return 1 return stop( client_id=args.client_id, region=args.region, execution_id=execution_id ) elif args.command == "principal": return principal(client_id=args.client_id, region=args.region) elif args.command == "status": execution_id = args.execution_id or args.execution_id_opt if not execution_id: print("You must provide an execution ID") return 1 return status( client_id=args.client_id, region=args.region, execution_id=execution_id ) elif args.command == "succeeded": return list_executions( args.client_id, args.region, "get-succeeded", "succeeded" ) elif args.command == "failed": return list_executions(args.client_id, args.region, "get-failed", "failed") elif args.command == "running": return list_executions(args.client_id, args.region, "get-running", "running") elif args.command == "timed-out": return list_executions( args.client_id, args.region, "get-timed-out", "timed-out" ) elif args.command == "aborted": return list_executions(args.client_id, args.region, "get-aborted", "aborted") elif args.command == "cloud-version": return cloud_version(client_id=args.client_id, region=args.region) else: print(f"Unknown command: {args.command}. Use --help for available commands.") return 1 if __name__ == "__main__": sys.exit(main())