122 lines
6.6 KiB
Python
122 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Capture an already positioned spectator through the authenticated local worker.
|
|
|
|
The server/operator must teleport the observer first. This helper cannot teleport
|
|
or edit a world and does not extend the MCP project's authorized world boundary.
|
|
Useful for operator-side checks of separate initial-generation worlds. The normal
|
|
MCP camera_capture tool continues to use Paper's project-scoped route.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
import urllib.request
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
RUNTIME = Path(os.environ.get('MCB_RUNTIME_DIR') or ROOT / '.runtime').expanduser().resolve()
|
|
# Share bounded transport/PNG checks; importing this module does not run its test.
|
|
spec = importlib.util.spec_from_file_location('camera_checks', ROOT / 'scripts/live-camera-test.py')
|
|
checks = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(checks)
|
|
|
|
|
|
def validate_request(pose):
|
|
"""Validate locally before reading credentials or sending a capture request."""
|
|
values = [pose.get(key) for key in ('x', 'y', 'z', 'yaw', 'pitch')]
|
|
if any(type(value) not in (int, float) for value in values):
|
|
raise checks.TestFailure('Pose values must all be numbers.')
|
|
checks.validate_pose(values)
|
|
dimension = pose.get('dimension')
|
|
if not isinstance(dimension, str) or len(dimension) > 128 or not re.fullmatch(r'[a-z0-9_.-]+:[a-z0-9_./-]+', dimension):
|
|
raise checks.TestFailure('Dimension must be a namespaced Minecraft dimension key.')
|
|
for key, fallback, low, high in (('fov', 70, 30, 110), ('width', 1280, 320, 1920), ('height', 720, 180, 1080)):
|
|
value = pose.get(key, fallback)
|
|
if type(value) is not int or not low <= value <= high:
|
|
raise checks.TestFailure(f'{key} must be an integer in {low}..{high}.')
|
|
|
|
|
|
def validated_image(result, capture_id, pose, token, started):
|
|
if result.get('captureId') != capture_id or result.get('mimeType') != 'image/png':
|
|
raise checks.TestFailure('Completed capture identity or MIME type did not match.')
|
|
if result.get('dimension') != pose['dimension']:
|
|
raise checks.TestFailure('Completed capture came from a different dimension.')
|
|
captured_at = checks.instant(result.get('capturedAt'), 'Capture timestamp')
|
|
if captured_at < started - 1 or captured_at > time.time() + 5:
|
|
raise checks.TestFailure('The frame timestamp does not correspond to this capture request.')
|
|
encoded = result.get('imageBase64')
|
|
if not isinstance(encoded, str) or len(encoded) > 12_000_000:
|
|
raise checks.TestFailure('Completed capture did not include a bounded image payload.')
|
|
try:
|
|
raw = base64.b64decode(encoded, validate=True)
|
|
except ValueError:
|
|
raise checks.TestFailure('Completed capture contained invalid image encoding.') from None
|
|
width, height = checks.inspect_png(raw)
|
|
if result.get('width') != width or result.get('height') != height:
|
|
raise checks.TestFailure('PNG dimensions disagree with the capture metadata.')
|
|
metadata = {key: value for key, value in result.items() if key in checks.SAFE_METADATA}
|
|
if any(not isinstance(value, (str, int, float, bool, type(None))) or isinstance(value, str) and len(value) > 1024
|
|
for value in metadata.values()):
|
|
raise checks.TestFailure('Capture metadata contains unexpected values.')
|
|
metadata.update(operator_direct_worker=True, imageSha256=hashlib.sha256(raw).hexdigest(), imageBytes=len(raw))
|
|
try:
|
|
receipt = json.dumps(metadata, indent=2, allow_nan=False) + '\n'
|
|
except ValueError:
|
|
raise checks.TestFailure('Capture metadata contains nonfinite values.') from None
|
|
if token in receipt:
|
|
raise checks.TestFailure('Refusing to save metadata containing a component secret.')
|
|
return raw, receipt
|
|
|
|
|
|
def capture(config, pose, output):
|
|
validate_request(pose)
|
|
try:
|
|
source = Path(config).expanduser().read_text()
|
|
token = checks.scalar(source, 'camera-token')
|
|
port = int(checks.scalar(source, 'camera-port'))
|
|
except (OSError, UnicodeError, ValueError):
|
|
raise checks.TestFailure('Cannot read a valid private camera configuration.') from None
|
|
if not 1024 <= port <= 65535 or not re.fullmatch(r'[A-Za-z0-9._~-]{32,512}', token):
|
|
raise checks.TestFailure('Invalid private camera configuration.')
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), checks.NoRedirect())
|
|
endpoint = f'http://127.0.0.1:{port}'
|
|
started, deadline = time.time(), time.monotonic() + 30
|
|
result = checks.request_json(opener, endpoint + '/v1/capture', token, pose, timeout=25)
|
|
capture_id = checks.identifier(result.get('captureId'), 'Capture ID')
|
|
while result.get('status') in ('pending', 'queued', 'capturing'):
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise checks.TestFailure('Camera did not complete in time.')
|
|
time.sleep(min(.3, remaining))
|
|
result = checks.request_json(opener, endpoint + '/v1/captures/' + capture_id, token, timeout=min(10, remaining))
|
|
if result.get('captureId') != capture_id:
|
|
raise checks.TestFailure('Camera returned a different capture identity.')
|
|
if result.get('status') != 'completed':
|
|
code = result.get('error', '')
|
|
safe_code = code if isinstance(code, str) and re.fullmatch(r'[a-z_]{1,64}', code) else 'capture_failed'
|
|
raise checks.TestFailure('Camera unavailable: ' + safe_code)
|
|
raw, receipt = validated_image(result, capture_id, pose, token, started)
|
|
output = Path(output).expanduser()
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_bytes(raw)
|
|
output.with_suffix('.capture.json').write_text(receipt)
|
|
print(json.dumps({'path': str(output.resolve()), 'dimension': result.get('dimension'), 'status': 'completed'}))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument('--config', type=Path, default=Path(os.environ.get('MCB_CAMERA_PAPER_CONFIG') or RUNTIME / 'server/plugins/MinecraftBuilderMCP/config.yml'))
|
|
p.add_argument('--pose', type=float, nargs=5, required=True, metavar=('X', 'Y', 'Z', 'YAW', 'PITCH'))
|
|
p.add_argument('--dimension', required=True)
|
|
p.add_argument('--fov', type=int, default=80)
|
|
p.add_argument('--out', type=Path, required=True)
|
|
a = p.parse_args()
|
|
try:
|
|
capture(a.config, dict(zip(('x', 'y', 'z', 'yaw', 'pitch'), a.pose)) | {'dimension': a.dimension, 'fov': a.fov, 'width': 1280, 'height': 720}, a.out)
|
|
except checks.TestFailure as error:
|
|
raise SystemExit(str(error)) from None
|