feat: screenshot API, camera control, pipe/wire crossing detection
New endpoints: - POST /api/action/screenshot: capture screenshot to file - GET /api/screenshot/latest: serve PNG binary - POST /api/action/camera: move camera to (x,y) + set zoom - GET /api/state/camera: get current camera position/zoom Pipe/Wire crossing intelligence: - build_pipe/build_wire now detect existing pipes/wires at each cell - mode='line' (default): new pipe merges with existing - mode='cross': auto-places bridges at intersections to avoid connection - mode='single': one segment CLI: snapshot [file.png], camera <x> <y> [zoom], camera_status All high-level commands: auto pause, execute, resume
This commit is contained in:
root
114
tools/oni_api.py
114
tools/oni_api.py
@ -702,24 +702,60 @@ def cmd_assign_job(args):
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def cmd_build_pipe_line(args):
|
||||
"""Build a pipe line. Usage: build_pipe_line <gas|liquid> <x1> <y1> <x2> <y2> [bridge]"""
|
||||
"""Build a pipe line with crossing mode. Usage: build_pipe_line <gas|liquid> <x1> <y1> <x2> <y2> [mode]"""
|
||||
if len(args) < 5:
|
||||
print("Usage: build_pipe_line gas|liquid <x1> <y1> <x2> <y2> [bridge]")
|
||||
print("Usage: build_pipe_line gas|liquid <x1> <y1> <x2> <y2> [mode]")
|
||||
print(" mode: 'line' (default, auto-merge), 'cross' (use bridges at intersections), 'single' (one segment)")
|
||||
return
|
||||
payload = {"type": args[0], "x1": int(args[1]), "y1": int(args[2]),
|
||||
"x2": int(args[3]), "y2": int(args[4]), "bridge": len(args) > 5 and args[5] == 'bridge'}
|
||||
result = api_post('/api/action/build_pipe', payload)
|
||||
_print_feedback(result)
|
||||
ptype = args[0]
|
||||
x1, y1, x2, y2 = int(args[1]), int(args[2]), int(args[3]), int(args[4])
|
||||
mode = args[5] if len(args) > 5 else 'line'
|
||||
|
||||
if mode not in ('line', 'cross', 'single'):
|
||||
print("[!] mode must be 'line', 'cross', or 'single'")
|
||||
return
|
||||
|
||||
result = api_post('/api/action/build_pipe', {
|
||||
"type": ptype, "x1": x1, "y1": y1,
|
||||
"x2": x2, "y2": y2, "mode": mode
|
||||
})
|
||||
if result.get('success'):
|
||||
d = result.get('data', {})
|
||||
segs = d.get('segmentCount', 0)
|
||||
bridges = d.get('bridgesPlaced', 0)
|
||||
xing = sum(1 for s in d.get('segments', []) if s.get('crossing'))
|
||||
print(f"[OK] {ptype} pipe: {segs} segments, {bridges} bridges, {xing} crossings handled")
|
||||
if xing > 0:
|
||||
print(f" Mode '{mode}': {'bridges used at crossings' if mode == 'cross' else 'merged at crossings'}")
|
||||
else:
|
||||
_print_feedback(result)
|
||||
|
||||
def cmd_build_wire_line(args):
|
||||
"""Build a wire line. Usage: build_wire_line <type> <x1> <y1> <x2> <y2> [bridge]"""
|
||||
"""Build a wire line with crossing mode. Usage: build_wire_line <type> <x1> <y1> <x2> <y2> [mode]"""
|
||||
if len(args) < 5:
|
||||
print("Usage: build_wire_line regular|heavy|conductive <x1> <y1> <x2> <y2> [bridge]")
|
||||
print("Usage: build_wire_line regular|heavy|conductive <x1> <y1> <x2> <y2> [mode]")
|
||||
print(" mode: 'line' (default, auto-merge), 'cross' (use bridges at intersections), 'single' (one segment)")
|
||||
return
|
||||
payload = {"type": args[0], "x1": int(args[1]), "y1": int(args[2]),
|
||||
"x2": int(args[3]), "y2": int(args[4]), "bridge": len(args) > 5 and args[5] == 'bridge'}
|
||||
result = api_post('/api/action/build_wire', payload)
|
||||
_print_feedback(result)
|
||||
wtype = args[0]
|
||||
x1, y1, x2, y2 = int(args[1]), int(args[2]), int(args[3]), int(args[4])
|
||||
mode = args[5] if len(args) > 5 else 'line'
|
||||
|
||||
if mode not in ('line', 'cross', 'single'):
|
||||
print("[!] mode must be 'line', 'cross', or 'single'")
|
||||
return
|
||||
|
||||
result = api_post('/api/action/build_wire', {
|
||||
"type": wtype, "x1": x1, "y1": y1,
|
||||
"x2": x2, "y2": y2, "mode": mode
|
||||
})
|
||||
if result.get('success'):
|
||||
d = result.get('data', {})
|
||||
segs = d.get('segmentCount', 0)
|
||||
bridges = d.get('bridgesPlaced', 0)
|
||||
xing = sum(1 for s in d.get('segments', []) if s.get('crossing'))
|
||||
print(f"[OK] {wtype} wire: {segs} segments, {bridges} bridges, {xing} crossings handled")
|
||||
else:
|
||||
_print_feedback(result)
|
||||
|
||||
def _print_feedback(result):
|
||||
"""Pretty-print action feedback."""
|
||||
@ -734,6 +770,52 @@ def _print_feedback(result):
|
||||
if sug:
|
||||
print(f" -> {sug}")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Screenshot / Camera
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def cmd_snapshot(args):
|
||||
"""Take a screenshot and save locally. Usage: snapshot [output.png]"""
|
||||
output = args[0] if args else f"oni_snapshot_{int(__import__('time').time())}.png"
|
||||
result = api_post('/api/action/screenshot', {})
|
||||
if not result.get('success'):
|
||||
print(f"[!] Screenshot failed: {result.get('errorMessage', '')}")
|
||||
return
|
||||
filename = result.get('data', {}).get('filename', '?')
|
||||
print(f"[OK] Screenshot taken: {filename}")
|
||||
|
||||
# Download the image
|
||||
try:
|
||||
import urllib.request
|
||||
from oni_api import api_url
|
||||
url = api_url('/api/screenshot/latest')
|
||||
urllib.request.urlretrieve(url, output)
|
||||
print(f"[OK] Saved to {output} ({os.path.getsize(output)} bytes)")
|
||||
except Exception as e:
|
||||
print(f"[!] Download failed: {e}")
|
||||
|
||||
def cmd_camera(args):
|
||||
"""Control camera. Usage: camera <x> <y> [zoom]"""
|
||||
if len(args) < 2:
|
||||
print("Usage: camera <x> <y> [zoom]")
|
||||
print(" zoom: 5 (close) to 80 (far), default 30")
|
||||
return
|
||||
x, y = int(args[0]), int(args[1])
|
||||
zoom = float(args[2]) if len(args) > 2 else None
|
||||
payload = {"x": x, "y": y}
|
||||
if zoom is not None:
|
||||
payload["zoom"] = zoom
|
||||
result = api_post('/api/action/camera', payload)
|
||||
_print_feedback(result)
|
||||
|
||||
def cmd_camera_status(args):
|
||||
"""Get current camera position. Usage: camera_status"""
|
||||
data = api_get('/api/state/camera')
|
||||
if 'error' in data:
|
||||
print(f"Error: {data['error']}")
|
||||
return
|
||||
print(f"Camera at ({data.get('x', 0):.0f}, {data.get('y', 0):.0f}), zoom={data.get('zoom', 30):.0f}")
|
||||
|
||||
def cmd_explore(args):
|
||||
"""AI-friendly exploration: reads a region and returns structured text summary."""
|
||||
x = int(args[0]) if len(args) > 0 else 0
|
||||
@ -849,6 +931,9 @@ COMMANDS = {
|
||||
'assign_job': cmd_assign_job,
|
||||
'build_pipe_line': cmd_build_pipe_line,
|
||||
'build_wire_line': cmd_build_wire_line,
|
||||
'snapshot': cmd_snapshot,
|
||||
'camera': cmd_camera,
|
||||
'camera_status': cmd_camera_status,
|
||||
}
|
||||
|
||||
if __name__ == '__main__':
|
||||
@ -920,6 +1005,11 @@ if __name__ == '__main__':
|
||||
print(" assign_job <name> <job> Assign dupe to a chore group")
|
||||
print(" morale Duplicant morale levels")
|
||||
print("")
|
||||
print("=== Screenshot / Camera ===")
|
||||
print(" snapshot [file.png] Take screenshot + save locally")
|
||||
print(" camera <x> <y> [zoom] Move camera view to coordinates")
|
||||
print(" camera_status Get current camera position/zoom")
|
||||
print("")
|
||||
print("=== Batch / Priority (Advanced) ===")
|
||||
print(" batch <json_file> Execute batch plan")
|
||||
print(" priority_global <t> <p> Set global default priority")
|
||||
|
||||
Reference in New Issue
Block a user