import json import os import re import subprocess import uuid import typer from pyfzf.pyfzf import FzfPrompt from celilo.Adb import Adb from celilo.common import saru_home # Configuration directory and file path CONFIG_DIR = os.path.expanduser("~/.somewear/.gridpilot") CONFIG_FILE = os.path.join(CONFIG_DIR, "config.json") def load_config(): """Load configuration from file.""" if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE, 'r') as f: return json.load(f) return {} def save_config(config): """Save configuration to file.""" # Create config directory if it doesn't exist os.makedirs(CONFIG_DIR, exist_ok=True) with open(CONFIG_FILE, 'w') as f: json.dump(config, f, indent=2) def select_workspace(): """ Fetch workspaces from the organization using Somewear CLI and let the user select one using fzf fuzzy finder. Saves the selected workspace ID and name. """ # Load org_id from config config = load_config() org_id = config.get('org_id') if not org_id: typer.echo("Error: No organization ID found.", err=True) typer.echo("Please set it first with 'gp set-org-id '", err=True) raise typer.Exit(code=1) typer.echo(f"Fetching workspaces for organization ID: {org_id}...") # Call somewear CLI to get workspaces try: result = subprocess.run( ['somewear', 'workspace', 'get', '--org-id', org_id], capture_output=True, text=True ) output = result.stdout.strip() if not output or "No workspace found" in output: typer.echo("No workspaces found for this organization.", err=True) typer.echo("Make sure you've joined workspaces in this organization.", err=True) raise typer.Exit(code=1) # Parse the output - it's a table format with │ separators workspace_map = {} workspace_options = [] try: # Split into lines lines = output.split('\n') # Parse table rows for line in lines: # Skip empty lines, header lines, and separator lines if not line.strip(): continue if '─' in line or 'Workspaces' in line or 'WorkspaceRecord' in line: continue if line.strip().startswith('│ id'): # This is the header row continue # Match table rows with format: │ id │ name │ organizationId │ styleSettings │ # Using regex to extract the columns match = re.match(r'│\s*(\d+)\s*│\s*([^│]+?)\s*│\s*(\d+)\s*│', line) if match: ws_id = match.group(1).strip() ws_name = match.group(2).strip() org_id_in_row = match.group(3).strip() if ws_id and ws_name: display = f"{ws_name} (ID: {ws_id})" workspace_options.append(display) workspace_map[display] = {'id': str(ws_id), 'name': ws_name} # If no workspaces found in table format, try JSON format if not workspace_options: typer.echo("Table format parsing yielded no results, trying JSON format...") # Split by "Workspace {" and process each block workspace_blocks = re.split(r'Workspace\s*\{', output) for block in workspace_blocks: if not block.strip(): continue # Add back the opening brace and try to extract the JSON json_str = '{' + block # Find the closing brace that matches (handle nested braces) brace_count = 0 end_idx = -1 for i, char in enumerate(json_str): if char == '{': brace_count += 1 elif char == '}': brace_count -= 1 if brace_count == 0: end_idx = i + 1 break if end_idx > 0: json_str = json_str[:end_idx] try: ws_data = json.loads(json_str) ws_id = ws_data.get('id') ws_name = ws_data.get('name') if ws_id and ws_name: display = f"{ws_name} (ID: {ws_id})" workspace_options.append(display) workspace_map[display] = {'id': str(ws_id), 'name': ws_name} except json.JSONDecodeError: # Skip this block if JSON parsing fails continue except Exception as e: typer.echo(f"Error parsing workspace data: {e}", err=True) typer.echo(f"Raw output:\n{output}", err=True) raise typer.Exit(code=1) if not workspace_options: typer.echo("Error: No valid workspaces found in output.", err=True) typer.echo(f"Raw output:\n{output}", err=True) raise typer.Exit(code=1) # Use fzf to select workspace fzf = FzfPrompt() try: selected = fzf.prompt(workspace_options, '--prompt="Select workspace: "') if not selected: typer.echo("No workspace selected.") raise typer.Exit(code=0) # Get the selected workspace data selected_display = selected[0] selected_workspace = workspace_map[selected_display] # Save to config config['workspace_id'] = selected_workspace['id'] config['workspace_name'] = selected_workspace['name'] save_config(config) typer.echo(f"\n✓ Workspace selected and saved:") typer.echo(f" Name: {selected_workspace['name']}") typer.echo(f" ID: {selected_workspace['id']}") typer.echo(f" Saved to: {CONFIG_FILE}") # Generate workspace key typer.echo(f"\nGenerating workspace key...") try: # Generate random UUID for workspace key workspace_key = str(uuid.uuid4()) key_result = subprocess.run( ['bash', '-c', f"echo 'y' | somewear workspace-key create --workspace-id {selected_workspace['id']} --key {workspace_key}"], capture_output=True, text=True ) print("out" + key_result.stdout.strip()) print("err" + key_result.stderr.strip()) if key_result.returncode == 0: typer.echo(f"\n✓ Workspace Key: {workspace_key}") # Save workspace key to config config['workspace_key'] = workspace_key save_config(config) # Use default config path config_path = "config.json" if os.path.exists(config_path): # Use default app package app_package = "com.somewearlabs.atak.debug" # Build workspace-key URI workspace_uri = f"api.somewear.co:443/workspace-key?key={workspace_key}&name={selected_workspace['name']}" typer.echo(f"\nLaunching app with workspace key...") # Call launch_app with the workspace key URI try: # Load config for device settings if not os.path.exists(config_path): typer.echo(f"Error: Config file not found: {config_path}", err=True) else: with open(config_path, 'r') as f: device_config = json.load(f) # Initialize ADB adb = Adb(app_package) devices = adb.devices() if not devices or not devices[0]: typer.echo("Error: No ADB devices found.", err=True) else: typer.echo(f"Found {len(devices)} device(s): {', '.join(devices)}") # Determine host based on package is_atak = app_package == "com.atakmap.app.civ" host = "somewear" if is_atak else "atak_somewear_test" # Launch on each device for device in devices: # Get device-specific config if device in device_config: msgInterval = device_config[device]["message_interval"] callsign = device_config[device].get("callsign", "") else: msgInterval = 0 callsign = "" typer.echo(f"Warning: Device {device} not found in config file", err=True) # Build full URI full_uri = f"{host}://{workspace_uri}&msgInterval={msgInterval}" if callsign: full_uri += f"&callsign={callsign}" # Escape & characters escaped_uri = full_uri.replace('&', '\\&') # Build and execute ADB command command = ( f'adb -s {device} shell am start -a android.intent.action.VIEW -d ' f'"{escaped_uri}" --ez collectAnalytics true' ) typer.echo(f"\n→ Launching on device: {device}") typer.echo(f" Command: {command}") result = os.system(command) if result == 0: typer.echo(f" ✓ Successfully launched on {device}") else: typer.echo(f" ✗ Failed to launch on {device}", err=True) typer.echo(f"\n✓ Launch complete") except Exception as e: typer.echo(f"Error launching app: {e}", err=True) else: typer.echo(f"\nconfig.json not found. Skipping app launch.") typer.echo("To launch manually, run:") typer.echo(f" gdp launch-app \"api.somewear.co:443/workspace-key?key={workspace_key}&name={selected_workspace['name']}\" --config-path ") else: typer.echo(f"Error: Failed to create workspace key. Command returned exit code {key_result.returncode}", err=True) except Exception as e: typer.echo(f"Note: Could not generate workspace key: {e}") except Exception as e: typer.echo(f"Error during selection: {e}", err=True) raise typer.Exit(code=1) except FileNotFoundError: typer.echo("Error: 'somewear' CLI not found. Please ensure it is installed and in your PATH.", err=True) raise typer.Exit(code=1)