"""Native PySide Oasis Action Player, backed by the connected Oasis project.""" from __future__ import annotations import importlib.util import io import json import math import sys import threading import time import traceback from pathlib import Path from PIL import Image, ImageDraw, ImageOps from PySide6.QtCore import QEvent, QThread, Qt, QTimer, Signal from PySide6.QtGui import QPixmap from PySide6.QtWidgets import ( QCheckBox, QComboBox, QFileDialog, QFormLayout, QFrame, QHBoxLayout, QLabel, QLineEdit, QMessageBox, QPushButton, QScrollArea, QSlider, QSpinBox, QVBoxLayout, QWidget, ) from adam.oasis_player import capture_path, discover_models, frame_size, model_info, random_roblox_dataset_frame def _external_player_module(root: Path): script = root / "roblox_action_flow_app.py" if not script.is_file(): raise RuntimeError("Oasis player code was not found. Connect Oasis-Game-Trainer in Settings.") # The connected project owns oasis_input.py, temporal_oasis.py and its VAE. # Its directory must be importable before executing roblox_action_flow_app.py. if str(root) not in sys.path: sys.path.insert(0, str(root)) spec = importlib.util.spec_from_file_location("adam_oasis_external_player", script) if spec is None or spec.loader is None: raise RuntimeError("ADAM could not load the Oasis player helpers.") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module class OasisPlaybackWorker(QThread): status = Signal(str) frames_ready = Signal(object, float, str, str) comparison_ready = Signal(bytes, str) failed = Signal(str) def __init__(self, adam_root: Path, oasis_root: Path, model: Path, image_bytes: bytes, settings: dict, state: dict, lock: threading.Lock, input_timeline=None): super().__init__() # The player code and trained models intentionally stay in Oasis, while # captures belong to ADAM. These paths must never be interchangeable. self.adam_root, self.oasis_root = adam_root, oasis_root self.model, self.image_bytes = model, image_bytes self.settings, self.state, self.lock = settings, state, lock self.input_timeline = input_timeline self._running = True self._compare = threading.Event() def stop(self): self._running = False def request_comparison(self): self._compare.set() @staticmethod def _png(image: Image.Image) -> bytes: data = io.BytesIO() image.save(data, format="PNG") return data.getvalue() def _snapshot(self): with self.lock: settings = dict(self.settings) state = dict(self.state) self.state["mouse_dx"] = 0.0 self.state["mouse_dy"] = 0.0 self.state["zoom"] = 0.0 return settings, state def _action(self, external, info: dict, state: dict): binary = external.BINARY_ACTION_NAMES enabled = set(info.get("enabled_action_names") or info.get("action_names") or external.ACTION_NAMES) require_drag = bool(self.settings.get("require_drag", True)) camera_allowed = not require_drag or bool(state.get("right_mouse", False)) mouse_dx = max(-1.0, min(1.0, float(state.get("mouse_dx", 0.0)) / external.PLAYER_MOUSE_SCALE)) if camera_allowed else 0.0 mouse_dy = max(-1.0, min(1.0, float(state.get("mouse_dy", 0.0)) / external.PLAYER_MOUSE_SCALE)) if camera_allowed else 0.0 zoom = max(-1.0, min(1.0, float(state.get("zoom", 0.0)) / external.PLAYER_ZOOM_SCALE)) held = set(state.get("keys", set())) values = [1.0 if ({"jump": "space"}.get(name, name) in held) else 0.0 for name in binary] values += [mouse_dx, mouse_dy, zoom] values = [value if name in enabled else 0.0 for name, value in zip(external.ACTION_NAMES, values)] moving = any(value > .5 for value in values[:len(binary)]) or any(abs(value) > .02 for value in values[-3:]) labels = [external.ACTION_DISPLAY_NAMES.get(name, name.upper()) for name, value in zip(binary, values) if value > .5] return values, moving, labels def _comparison(self, external, source, model, previous, info, torch, dtype, device, generator): actions = [("Idle", [0.0] * external.ACTION_DIM)] enabled = set(info.get("enabled_action_names") or info.get("action_names") or external.ACTION_NAMES) for index, name in enumerate(external.BINARY_ACTION_NAMES): if name in enabled: values = [0.0] * external.ACTION_DIM values[index] = 1.0 actions.append((external.ACTION_DISPLAY_NAMES.get(name, name.upper()), values)) actions = actions[:9] width, height = frame_size(info) panel = Image.new("RGB", (width * 3, (height + 24) * math.ceil(len(actions) / 3)), (9, 19, 29)) draw = ImageDraw.Draw(panel) base_noise = torch.randn(previous.shape, generator=generator, device=device, dtype=dtype) for index, (label, values) in enumerate(actions): action = torch.tensor([values], device=device, dtype=dtype) generated = external.sample_guided_action_frame( model, previous, action, int(self.settings["steps"]), device, dtype, generator, method=self.settings["method"], initial_noise=base_noise.clone(), action_input_scale=float(info.get("action_input_scale", 1.0)), guidance=float(self.settings["guidance"]), ) image = source.tensor_to_pil(generated[0].clamp(-1, 1)) x, y = (index % 3) * width, (index // 3) * (height + 24) panel.paste(image, (x, y + 24)); draw.text((x + 5, y + 5), label, fill=(237, 245, 251)) path = capture_path(self.adam_root, "oasis_action_compare") panel.save(path) self.comparison_ready.emit(self._png(panel), str(path)) def _run_temporal(self, external, info: dict, current: Image.Image, torch) -> None: """Run the external temporal format without applying legacy guidance/noise UI tricks.""" if self.input_timeline is None: raise RuntimeError("Temporal input timing was not initialized.") if info.get("action_names") != external.ACTION_NAMES: raise RuntimeError("This temporal Oasis model uses a control schema ADAM does not recognize.") from oasis_input import interval_vector pixel_engine = info.get("model_type") == "action_conditioned_temporal_pixel_flow" if pixel_engine: from temporal_pixel_oasis import TemporalPixelPlayback as TemporalPlayback, load_model else: from temporal_oasis import TemporalPlayback, load_model from vae_cpu_lite import load_vae import numpy as np required_source = info.get("mouse_input_source") if required_source == "windows_raw_input" and not bool(self.settings.get("raw_mouse_ready")): raise RuntimeError("This model was trained with raw mouse counts. Raw mouse input is unavailable in ADAM, so cursor pixels will not be substituted.") device, dtype = torch.device("cuda" if torch.cuda.is_available() else "cpu"), torch.float32 self.status.emit("Loading temporal Oasis model…") model = load_model(self.model, device, dtype) vae = None if pixel_engine else load_vae(self.model, device, dtype) pixels = torch.from_numpy(np.array(current.convert("RGB"), copy=True)).permute(2, 0, 1).float() / 127.5 - 1 generator = torch.Generator(device=device).manual_seed(int(self.settings["seed"])) world = (TemporalPlayback(model, info, pixels[None], device, dtype, generator) if pixel_engine else TemporalPlayback(model, vae, info, pixels[None], device, dtype, generator)) dt = float(info["step_seconds"]) self.status.emit(f"Temporal preview ready: {info['context_frames']} frames of history, {dt:.4f}s per world step.") world_step = 0 while self._running: deadline = time.perf_counter() + dt while self._running and time.perf_counter() < deadline: time.sleep(min(0.01, max(0.0, deadline - time.perf_counter()))) if not self._running: break settings, _state = self._snapshot() interval = self.input_timeline.snapshot(time.perf_counter()) if not interval["valid"]: raise RuntimeError("Temporal input timing overflowed. Stop and restart the player.") values = interval_vector(interval, external.BINARY_ACTION_NAMES, info["mouse_scale"], info["scroll_scale"]) has_input = any(value > 1e-6 for value in values[:len(external.BINARY_ACTION_NAMES)]) or any( abs(value) > 1e-6 for value in values[len(external.BINARY_ACTION_NAMES):] ) if not settings.get("temporal_advance_idle", True) and not has_input: self.status.emit("Temporal preview paused for no input. Move the mouse or press a control to advance.") continue snapshot = world.snapshot() rgb = world.step(values, int(settings["steps"]), str(settings["method"])) if not self._running: world.restore(snapshot) break array = ((rgb[0].detach().float().cpu().clamp(-1, 1) + 1) * 127.5).round().byte().permute(1, 2, 0).numpy() current = Image.fromarray(array) world_step += 1 text = f"Temporal input | mouse {values[-3]:+.2f}, {values[-2]:+.2f} | world time {world_step * dt:.2f}s" self.frames_ready.emit([self._png(current)], 1 / dt, text, "Learned history; raw generated frames") def run(self): try: import torch external = _external_player_module(self.oasis_root) info = model_info(self.model) width, height = frame_size(info) current = ImageOps.fit(Image.open(io.BytesIO(self.image_bytes)).convert("RGB"), (width, height), method=Image.Resampling.LANCZOS) if info.get("model_type") in {"action_conditioned_temporal_latent_flow", "action_conditioned_temporal_pixel_flow"}: import torch self._run_temporal(external, info, current, torch) return source, _ = external.load_source_module() device = torch.device("cuda" if torch.cuda.is_available() else "cpu") dtype = torch.float16 if device.type == "cuda" else torch.float32 self.status.emit("Loading Oasis action model…") model = external.load_action_unet(self.model, device, dtype) generator = torch.Generator(device=device).manual_seed(int(self.settings["seed"])) noise_state = None self.status.emit(f"Running {info.get('name', self.model.name)} on {device.type.upper()}.") while self._running: started = time.perf_counter() settings, state = self._snapshot() previous_frame = current.copy() previous = source.pil_to_normalized_tensor(current, device, dtype) values, moving, labels = self._action(external, info, state) if not moving and not settings["animate_idle"]: result = current.copy() else: refresh = float(settings["motion_noise"]) * (float(settings["active_noise"]) if moving else float(settings["idle_noise"])) if noise_state is None or tuple(noise_state.shape) != tuple(previous.shape): noise_state = torch.randn(previous.shape, generator=generator, device=device, dtype=dtype) elif refresh: fresh = torch.randn(previous.shape, generator=generator, device=device, dtype=dtype) noise_state = noise_state * math.sqrt(max(0.0, 1 - refresh * refresh)) + fresh * refresh action = torch.tensor([external.subdivide_action_values(values, int(settings["subdivision"]))], device=device, dtype=dtype) generated = external.sample_guided_action_frame(model, previous, action, int(settings["steps"]), device, dtype, generator, method=settings["method"], initial_noise=noise_state.detach().clone(), action_input_scale=float(info.get("action_input_scale", 1.0)), guidance=float(settings["guidance"])) result = source.tensor_to_pil(generated[0]) if settings["stability"]: try: result = external.motion_aware_stabilize(current, result, float(settings["stability"])) except Exception: pass current = result if self._compare.is_set(): self._compare.clear(); self._comparison(external, source, model, previous, info, torch, dtype, device, generator) fps = max(.5, float(settings["fps"])); period = 1 / fps try: display = external.interpolate_motion_frames(previous_frame, current, max(1, int(settings["display_multiplier"]) if moving else 1)) except Exception: display = [current] text = "Guided input: " + (" + ".join(labels) or "IDLE") + f" | yaw {values[-3]:+.2f} pitch {values[-2]:+.2f} zoom {values[-1]:+.2f}" self.frames_ready.emit([self._png(frame) for frame in display], fps, text, f"Motion-aware stability {float(settings['stability']) * 100:.0f}%") remaining = period - (time.perf_counter() - started) if remaining > 0: time.sleep(remaining) except Exception: self.failed.emit(traceback.format_exc()) class OasisPlayerPage(QWidget): """The playable Oasis experience; training deliberately remains in Training Studio.""" def __init__(self, root, assets, config): super().__init__() self.root, self.assets, self.config = Path(root), assets, config self.models: dict[str, Path] = {}; self.frame: bytes | None = None; self.worker = None self.state = {"keys": set(), "mouse_dx": 0.0, "mouse_dy": 0.0, "zoom": 0.0, "right_mouse": False} self.lock = threading.Lock(); self.sequence = 0 self.temporal_timeline = None; self.raw_mouse_input = None; self.temporal_mouse_source = None self._restoring = True self._build(); self.refresh(); self._restore() self._restoring = False self._model_changed() def _build(self): layout = QVBoxLayout(self); layout.setContentsMargins(22, 18, 22, 18) title = QLabel("Oasis Player"); title.setObjectName("PageTitle"); layout.addWidget(title) sub = QLabel("Explore a trained action-world model from a starting screenshot. Training stays in Training Studio."); sub.setProperty("muted", True); layout.addWidget(sub) body = QHBoxLayout(); layout.addLayout(body, 1) controls = QFrame(); controls.setProperty("card", True); controls.setMaximumWidth(350) cl = QVBoxLayout(controls); cl.setContentsMargins(15, 15, 15, 15) label = QLabel("PLAYER CONTROLS"); label.setObjectName("CardTitle"); cl.addWidget(label) self.model = QComboBox(); self.model.currentIndexChanged.connect(self._model_changed); cl.addWidget(self.model) refresh = QPushButton("Refresh models"); refresh.clicked.connect(self.refresh); cl.addWidget(refresh) self.image = QLineEdit(); self.image.setPlaceholderText("Choose a starting game screenshot"); cl.addWidget(self.image) browse = QPushButton("Choose starting image"); browse.clicked.connect(self._browse_image); cl.addWidget(browse) self.random_dataset = QCheckBox("Use a random Roblox Dataset frame when starting"); self.random_dataset.setChecked(True); cl.addWidget(self.random_dataset) self.random_dataset.toggled.connect(lambda _checked: self._save()) find_frame = QPushButton("Find New Frame"); find_frame.clicked.connect(self._find_new_frame); cl.addWidget(find_frame) form = QFormLayout(); self.steps = self._combo(["1", "2", "4", "6", "8", "10", "12"], "4"); self.method = self._combo(["Euler", "Heun"], "Euler"); self.guidance = self._combo(["1.0", "1.5", "2.0", "2.5", "3.0", "3.5"], "2.0"); self.fps = self._combo(["2", "4", "6", "8", "12", "24"], "12"); self.subdivision = self._combo(["1", "2", "3", "4"], "3"); self.multiplier = self._combo(["1", "2", "3", "4"], "1") for name, field in (("Flow steps", self.steps), ("Denoise method", self.method), ("Action guidance", self.guidance), ("Gameplay FPS", self.fps), ("Motion subdivision", self.subdivision), ("Display interpolation", self.multiplier)): form.addRow(name, field) cl.addLayout(form) self.stability = QSlider(Qt.Horizontal); self.stability.setRange(0, 100); self.stability.setValue(25); cl.addWidget(QLabel("Motion-aware stability")); cl.addWidget(self.stability) self.idle = QCheckBox("Keep no-input scenes animated"); self.idle.setChecked(True); self.drag = QCheckBox("Camera requires right-mouse drag"); self.drag.setChecked(True); cl.addWidget(self.idle); cl.addWidget(self.drag) self.temporal_idle = QCheckBox("Temporal: advance during no-input"); self.temporal_idle.setChecked(True); cl.addWidget(self.temporal_idle) self.temporal_hint = QLabel("When enabled, idle time tests coasting, falling, and other learned dynamics. When disabled, the preview waits for an input."); self.temporal_hint.setProperty("muted", True); self.temporal_hint.setWordWrap(True); cl.addWidget(self.temporal_hint) self.start = QPushButton("Start Action Player"); self.start.setProperty("primary", True); self.start.clicked.connect(self._toggle); cl.addWidget(self.start) reset = QPushButton("Reset frame"); reset.clicked.connect(self._reset_frame); cl.addWidget(reset) capture = QPushButton("Capture frame"); capture.clicked.connect(self._capture); compare = QPushButton("Compare actions (same noise)"); compare.clicked.connect(self._compare); cl.addWidget(capture); cl.addWidget(compare); cl.addStretch() body.addWidget(controls) display = QFrame(); display.setProperty("card", True); dl = QVBoxLayout(display); dl.setContentsMargins(12, 12, 12, 12) top = QHBoxLayout(); self.status = QLabel("Choose an Oasis model and starting image."); self.status.setProperty("muted", True); self.metrics = QLabel("AI FPS —"); self.metrics.setProperty("accent", True); top.addWidget(self.status); top.addStretch(); top.addWidget(self.metrics); dl.addLayout(top) self.preview = QLabel("Select a model and screenshot to begin."); self.preview.setAlignment(Qt.AlignCenter); self.preview.setMinimumSize(620, 420); self.preview.setStyleSheet("background: #03080d; border: 1px solid #183246; border-radius: 8px;"); self.preview.setFocusPolicy(Qt.StrongFocus); self.preview.installEventFilter(self); dl.addWidget(self.preview, 1) self.input = QLabel("Guided input: IDLE"); self.input.setProperty("accent", True); dl.addWidget(self.input); body.addWidget(display, 1) for control in (self.steps, self.method, self.guidance, self.fps, self.subdivision, self.multiplier): control.currentTextChanged.connect(lambda _value: self._save()) self.stability.valueChanged.connect(lambda _value: self._save()) self.idle.toggled.connect(lambda _checked: self._save()) self.temporal_idle.toggled.connect(lambda _checked: self._save()) self.drag.toggled.connect(lambda _checked: self._save()) @staticmethod def _combo(items, current): box = QComboBox(); box.addItems(items); box.setCurrentText(current); return box def refresh(self): oasis = Path(str(self.config.get("tool_folders", {}).get("oasis_trainer", ""))) current = self.model.currentData(); self.models = dict(discover_models(oasis, self.assets.assets)) if oasis.is_dir() else {} self.model.blockSignals(True); self.model.clear() for name, path in self.models.items(): self.model.addItem(name, str(path)) if current: index = self.model.findData(current) if index >= 0: self.model.setCurrentIndex(index) self.model.blockSignals(False); self._model_changed() def _restore(self): saved = self.config.get("oasis_player_settings", {}) if isinstance(saved, dict): self.image.setText(str(saved.get("starting_image", ""))) self.random_dataset.setChecked(bool(saved.get("random_roblox_dataset", True))) for key, control in ( ("steps", self.steps), ("method", self.method), ("guidance", self.guidance), ("fps", self.fps), ("subdivision", self.subdivision), ("display_multiplier", self.multiplier), ): value = str(saved.get(key, "")) if control.findText(value) >= 0: control.setCurrentText(value) try: self.stability.setValue(max(0, min(100, int(saved.get("stability", 25))))) except (TypeError, ValueError): pass self.idle.setChecked(bool(saved.get("animate_idle", True))) self.temporal_idle.setChecked(bool(saved.get("temporal_advance_idle", True))) self.drag.setChecked(bool(saved.get("require_drag", True))) index = self.model.findData(str(saved.get("model", ""))) if index >= 0: self.model.setCurrentIndex(index) def _save(self): if getattr(self, "_restoring", False): return self.config.update({"oasis_player_settings": { "model": self.model.currentData() or "", "starting_image": self.image.text(), "random_roblox_dataset": self.random_dataset.isChecked(), "steps": self.steps.currentText(), "method": self.method.currentText(), "guidance": self.guidance.currentText(), "fps": self.fps.currentText(), "subdivision": self.subdivision.currentText(), "display_multiplier": self.multiplier.currentText(), "stability": self.stability.value(), "animate_idle": self.idle.isChecked(), "temporal_advance_idle": self.temporal_idle.isChecked(), "require_drag": self.drag.isChecked(), }}) def _model_changed(self): path = self.model.currentData() temporal = False if path: try: temporal = model_info(path).get("model_type") in {"action_conditioned_temporal_latent_flow", "action_conditioned_temporal_pixel_flow"} except ValueError: pass self.status.setText(f"Ready: {self.model.currentText()} · choose a matching starting screenshot.") self.temporal_idle.setVisible(temporal); self.temporal_hint.setVisible(temporal) self._save() def _browse_image(self): path, _ = QFileDialog.getOpenFileName(self, "Choose starting screenshot", "", "Images (*.png *.jpg *.jpeg *.webp *.bmp)") if path: self.image.setText(path); self._reset_frame() def _reset_frame(self): try: path = Path(self.image.text()) info = model_info(self.model.currentData()); width, height = frame_size(info) with Image.open(path) as image: image = ImageOps.fit(image.convert("RGB"), (width, height), method=Image.Resampling.LANCZOS) output = io.BytesIO(); image.save(output, format="PNG"); self.frame = output.getvalue() self._show(self.frame); self.status.setText("Frame reset. Press Start Action Player."); self._save() except Exception as exc: QMessageBox.warning(self, "Starting image needed", str(exc)) def _settings(self): return {"seed": int(time.time() * 1000) % 1_000_000_000, "steps": int(self.steps.currentText()), "method": self.method.currentText(), "guidance": float(self.guidance.currentText()), "fps": float(self.fps.currentText()), "subdivision": int(self.subdivision.currentText()), "display_multiplier": int(self.multiplier.currentText()), "stability": self.stability.value() / 100, "animate_idle": self.idle.isChecked(), "temporal_advance_idle": self.temporal_idle.isChecked(), "require_drag": self.drag.isChecked(), "motion_noise": .06, "active_noise": 1.0, "idle_noise": .18, "raw_mouse_ready": bool(self.raw_mouse_input and self.raw_mouse_input.available)} def _oasis_root(self) -> Path: raw = self.config.get("tool_folders", {}).get("oasis_trainer", "") folder = Path(str(raw)).expanduser() if not (folder / "roblox_action_flow_app.py").is_file(): raise ValueError("Oasis Game Trainer is not configured with roblox_action_flow_app.py. Check Settings → Tool Folders.") return folder.resolve() def _find_new_frame(self) -> bool: """Find and immediately prepare a new nested Roblox recording frame.""" try: selected = random_roblox_dataset_frame(self._oasis_root()) self.image.setText(str(selected)) self._reset_frame() self.status.setText(f"Random Roblox Dataset frame ready: {selected.name}") return self.frame is not None except (OSError, ValueError) as exc: QMessageBox.warning(self, "Roblox Dataset frame unavailable", str(exc)) return False def _toggle(self): if self.worker and self.worker.isRunning(): self._stop(); return if self.random_dataset.isChecked() and not self._find_new_frame(): return if not self.model.currentData() or not self.frame: QMessageBox.warning(self, "Player needs setup", "Choose a valid Oasis model and reset a starting image first."); return try: oasis_root = self._oasis_root() except ValueError as exc: QMessageBox.warning(self, "Oasis connection needed", str(exc)) return try: info = model_info(self.model.currentData()) self._prepare_temporal_inputs(oasis_root, info) except (ImportError, OSError, RuntimeError, ValueError) as exc: QMessageBox.warning(self, "Temporal Oasis setup", str(exc)); return self.worker = OasisPlaybackWorker(self.root, oasis_root, Path(self.model.currentData()), self.frame, self._settings(), self.state, self.lock, self.temporal_timeline) self.worker.status.connect(self.status.setText); self.worker.frames_ready.connect(self._frames); self.worker.comparison_ready.connect(self._comparison); self.worker.failed.connect(self._failed); self.worker.finished.connect(lambda: self.start.setText("Start Action Player")); self.worker.start(); self.start.setText("Stop Action Player"); self.preview.setFocus() def _stop(self): if self.worker: self.worker.stop(); self.status.setText("Action player paused.") self._close_temporal_inputs() self.start.setText("Start Action Player") def shutdown(self): """Stop inference before Qt tears down the page and its CUDA worker.""" if self.worker and self.worker.isRunning(): self.worker.stop() self.worker.wait(3000) self._close_temporal_inputs() def _frames(self, frames, fps, text, mode): self.sequence += 1; token = self.sequence; self.input.setText(text); self.metrics.setText(f"AI playback FPS: {fps:.1f} | {mode}") for index, frame in enumerate(frames): QTimer.singleShot(round(index * 1000 / max(1, fps * len(frames))), lambda data=frame, current=token: self._show(data) if current == self.sequence else None) def _show(self, data): self.frame = data; pixmap = QPixmap(); pixmap.loadFromData(data, "PNG"); self.preview.setPixmap(pixmap.scaled(self.preview.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation)) def _capture(self): if not self.frame: return path = capture_path(self.root) path.write_bytes(self.frame); self.status.setText(f"Captured {path.name}") def _compare(self): try: if model_info(self.model.currentData()).get("model_type") in {"action_conditioned_temporal_latent_flow", "action_conditioned_temporal_pixel_flow"}: QMessageBox.information(self, "Temporal model", "Temporal models are checked through their history and generated-rollout metrics. The single-image comparison does not apply.") return except ValueError: pass if not self.worker or not self.worker.isRunning(): QMessageBox.information(self, "Start player first", "Start the player to load the model, then compare actions."); return self.status.setText("Generating same-noise action comparison…"); self.worker.request_comparison() def _comparison(self, data, path): self._show(data); self.status.setText(f"Saved same-noise action comparison: {Path(path).name}") def _failed(self, error): self._close_temporal_inputs(); self.start.setText("Start Action Player"); self.status.setText("Action player failed."); QMessageBox.critical(self, "Oasis player error", error[-2500:]) def _prepare_temporal_inputs(self, oasis_root: Path, info: dict) -> None: self._close_temporal_inputs() if info.get("model_type") not in {"action_conditioned_temporal_latent_flow", "action_conditioned_temporal_pixel_flow"}: return _external_player_module(oasis_root) from oasis_input import InputTimeline from oasis_raw_mouse import RawMouseInput source = str(info.get("mouse_input_source", "")) if source not in {"windows_raw_input", "absolute_cursor_fallback"}: raise ValueError("This temporal model has no recognized mouse input source.") with self.lock: buttons = {("jump" if key == "space" else key): 1 for key in self.state["keys"]} buttons["mouse_right"] = int(self.state["right_mouse"]) self.temporal_timeline = InputTimeline(); self.temporal_timeline.reset(time.perf_counter(), buttons) self.temporal_mouse_source = source if source == "windows_raw_input": self.raw_mouse_input = RawMouseInput(self._raw_mouse) if not self.raw_mouse_input.available: error = self.raw_mouse_input.error or "Raw mouse input could not start." self._close_temporal_inputs() raise RuntimeError(error) def _close_temporal_inputs(self) -> None: if self.raw_mouse_input is not None: self.raw_mouse_input.close() self.raw_mouse_input = None; self.temporal_timeline = None; self.temporal_mouse_source = None def _raw_mouse(self, dx, dy, source_time=None) -> None: timeline = self.temporal_timeline if timeline is not None and self.temporal_mouse_source == "windows_raw_input": timeline.push("mouse", dx=float(dx), dy=float(dy), source_time=source_time) def eventFilter(self, watched, event): if watched is self.preview: with self.lock: if event.type() == QEvent.KeyPress: key = {Qt.Key_W: "w", Qt.Key_A: "a", Qt.Key_S: "s", Qt.Key_D: "d", Qt.Key_Space: "space", Qt.Key_Up: "arrow_up", Qt.Key_Left: "arrow_left", Qt.Key_Down: "arrow_down", Qt.Key_Right: "arrow_right"}.get(event.key()) if key: self.state["keys"].add(key) if self.temporal_timeline is not None: self.temporal_timeline.push("button", name="jump" if key == "space" else key, value=1) return True if event.type() == QEvent.KeyRelease: key = {Qt.Key_W: "w", Qt.Key_A: "a", Qt.Key_S: "s", Qt.Key_D: "d", Qt.Key_Space: "space", Qt.Key_Up: "arrow_up", Qt.Key_Left: "arrow_left", Qt.Key_Down: "arrow_down", Qt.Key_Right: "arrow_right"}.get(event.key()) if key: self.state["keys"].discard(key) if self.temporal_timeline is not None: self.temporal_timeline.push("button", name="jump" if key == "space" else key, value=0) return True if event.type() == QEvent.MouseButtonPress and event.button() == Qt.RightButton: self.state["right_mouse"] = True; self.state["position"] = event.position() if self.temporal_timeline is not None: self.temporal_timeline.push("button", name="mouse_right", value=1) return True if event.type() == QEvent.MouseButtonRelease and event.button() == Qt.RightButton: self.state["right_mouse"] = False if self.temporal_timeline is not None: self.temporal_timeline.push("button", name="mouse_right", value=0) return True if event.type() == QEvent.MouseMove and self.state.get("right_mouse"): old = self.state.get("position", event.position()); dx, dy = event.position().x() - old.x(), event.position().y() - old.y() self.state["mouse_dx"] += dx; self.state["mouse_dy"] += dy; self.state["position"] = event.position() if self.temporal_timeline is not None and self.temporal_mouse_source == "absolute_cursor_fallback": self.temporal_timeline.push("mouse", dx=dx, dy=dy) return True if event.type() == QEvent.Wheel: scroll = event.angleDelta().y() / 120; self.state["zoom"] += scroll if self.temporal_timeline is not None: self.temporal_timeline.push("scroll", value=scroll) return True return super().eventFilter(watched, event) def closeEvent(self, event): self.shutdown(); super().closeEvent(event)