nova-sim / scripts /discover_nova_simple.py
Georg
Add environment configuration and Nova API integration support
dcf1b21
#!/usr/bin/env python3
"""Discover Nova API configuration - simple version without dependencies."""
import json
import os
import sys
import urllib.request
from pathlib import Path
def load_env_file(filepath):
"""Simple .env file loader."""
env_vars = {}
if not filepath.exists():
return env_vars
with open(filepath, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#') and '=' in line:
key, value = line.split('=', 1)
env_vars[key.strip()] = value.strip()
return env_vars
def request_json(url, access_token):
"""Make a GET request to the Nova API."""
request = urllib.request.Request(url)
request.add_header("Authorization", f"Bearer {access_token}")
request.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(request, timeout=10) as response:
body = response.read().decode("utf-8")
return json.loads(body) if body else {}
except urllib.error.HTTPError as exc:
body = exc.read().decode("utf-8") if exc.fp else ""
print(f"Error: Nova API returned status {exc.code}")
print(f"Response: {body}")
return None
except Exception as exc:
print(f"Error: {exc}")
return None
def main():
# Try to load from .env.local first, then .env
env_local = Path("nova-sim/.env.local")
env_file = Path("nova-sim/.env")
env_vars = {}
if env_local.exists():
env_vars = load_env_file(env_local)
print(f"Loaded configuration from {env_local}")
elif env_file.exists():
env_vars = load_env_file(env_file)
print(f"Loaded configuration from {env_file}")
else:
print("Error: No .env or .env.local file found")
print("Create one with at least NOVA_INSTANCE_URL and NOVA_ACCESS_TOKEN")
return 1
# Get configuration
instance_url = env_vars.get("NOVA_INSTANCE_URL") or env_vars.get("NOVA_API")
access_token = env_vars.get("NOVA_ACCESS_TOKEN")
cell_id = env_vars.get("NOVA_CELL_ID", "cell")
if not instance_url:
print("Error: NOVA_INSTANCE_URL (or NOVA_API) not set in .env file")
return 1
if not access_token:
print("Error: NOVA_ACCESS_TOKEN not set in .env file")
return 1
print(f"\nConnecting to Nova instance: {instance_url}")
print(f"Cell ID: {cell_id}")
print()
# Fetch controllers
print("Fetching controllers...")
controllers_url = f"{instance_url}/api/v2/cells/{cell_id}/controllers"
controllers_data = request_json(controllers_url, access_token)
if not controllers_data:
return 1
# Handle both list and dict responses
if isinstance(controllers_data, list):
controllers = controllers_data
else:
controllers = controllers_data.get("instances", [])
if not controllers:
print("No controllers found")
return 1
print(f"Found {len(controllers)} controller(s):")
print()
for controller in controllers:
# Handle both string and dict controller responses
if isinstance(controller, str):
controller_id = controller
else:
controller_id = controller.get("controller", "unknown")
print(f"Controller ID: {controller_id}")
# Fetch motion groups - try cell-level endpoint first
motion_groups_url = f"{instance_url}/api/v2/cells/{cell_id}/motion-groups"
motion_groups_data = request_json(motion_groups_url, access_token)
if not motion_groups_data:
# Try controller-specific endpoint
motion_groups_url = f"{instance_url}/api/v2/cells/{cell_id}/controllers/{controller_id}/motion-groups"
motion_groups_data = request_json(motion_groups_url, access_token)
if not motion_groups_data:
print(" Motion groups: Could not fetch")
continue
motion_groups = motion_groups_data.get("motion_groups", [])
print(f" Motion groups: {len(motion_groups)}")
for mg in motion_groups:
mg_id = mg.get("motion_group", "unknown")
model = mg.get("model_from_controller", "unknown")
name = mg.get("name_from_controller", "")
print(f" - ID: {mg_id}")
print(f" Model: {model}")
if name:
print(f" Name: {name}")
# Fetch motion group description
desc_url = f"{instance_url}/api/v2/cells/{cell_id}/controllers/{controller_id}/motion-groups/{mg_id}/description"
desc_data = request_json(desc_url, access_token)
if desc_data:
tcps = desc_data.get("tcps", {})
if tcps:
print(f" Available TCPs: {', '.join(tcps.keys())}")
print()
# Generate recommended .env configuration
print("\n" + "=" * 60)
print("RECOMMENDED CONFIGURATION")
print("=" * 60)
if controllers:
# Handle both string and dict controller responses
controller = controllers[0]
if isinstance(controller, str):
controller_id = controller
else:
controller_id = controller.get("controller", "unknown")
# Get motion groups for first controller
motion_groups_url = f"{instance_url}/api/v2/cells/{cell_id}/controllers/{controller_id}/motion-groups"
motion_groups_data = request_json(motion_groups_url, access_token)
if motion_groups_data:
motion_groups = motion_groups_data.get("motion_groups", [])
if motion_groups:
mg = motion_groups[0]
mg_id = mg.get("motion_group", "unknown")
model = mg.get("model_from_controller", "unknown")
print("\nAdd/update these in your .env.local file:")
print(f"""
NOVA_INSTANCE_URL={instance_url}
NOVA_ACCESS_TOKEN={access_token}
NOVA_CELL_ID={cell_id}
NOVA_CONTROLLER_ID={controller_id}
NOVA_MOTION_GROUP_ID={mg_id}
NOVA_MOTION_GROUP_MODEL={model}
NOVA_TCP_NAME=Flange
NOVA_RESPONSE_RATE_MS=200
""")
return 0
if __name__ == "__main__":
sys.exit(main())