goyimnose/Medios-Macina
1
0
Fork 0
Code Issues Pull Requests Packages Releases 1 Wiki Activity

updated reply to mpv

Browse Source
This commit is contained in:
Nose
2026-03-21 14:22:48 -07:00
parent 6bb3875185
commit 11384266e3
2 changed files with 118 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
CLI.py
View File

@@ -115,26 +115,21 @@ from SYS.cli_parsing import SelectionSyntax, SelectionFilterSyntax, MedeiaLexer
def _notify_mpv_osd(text: str, *, duration_ms: int = 3500) -> bool:
message = str(text or "").strip()
if not message:
def _send_mpv_ipc_command(command: List[Any], *, ipc_path: Optional[str] = None, timeout: float = 0.75) -> bool:
if not isinstance(command, list) or not command:
return False
try:
from MPV.mpv_ipc import MPVIPCClient, get_ipc_pipe_path
client = MPVIPCClient(
socket_path=get_ipc_pipe_path(),
timeout=0.75,
socket_path=str(ipc_path or get_ipc_pipe_path()),
timeout=max(0.1, float(timeout or 0.75)),
silent=True,
)
try:
response = client.send_command({
"command": [
"show-text",
message,
max(0, int(duration_ms)),
]
"command": command,
})
finally:
try:
@@ -144,14 +139,62 @@ def _notify_mpv_osd(text: str, *, duration_ms: int = 3500) -> bool:
return bool(response and response.get("error") == "success")
except Exception as exc:
debug(f"mpv notify failed: {exc}")
debug(f"mpv ipc command failed: {exc}")
return False
def _notify_mpv_osd(text: str, *, duration_ms: int = 3500, ipc_path: Optional[str] = None) -> bool:
message = str(text or "").strip()
if not message:
return False
return _send_mpv_ipc_command(
[
"show-text",
message,
max(0, int(duration_ms)),
],
ipc_path=ipc_path,
)
def _notify_mpv_callback(metadata: Dict[str, Any], execution_result: Dict[str, Any]) -> bool:
callback = metadata.get("mpv_callback") if isinstance(metadata, dict) else None
if not isinstance(callback, dict):
return False
script_name = str(callback.get("script") or "").strip()
message_name = str(callback.get("message") or "").strip()
ipc_path = str(callback.get("ipc_path") or "").strip() or None
if not script_name or not message_name:
return False
payload = {
"phase": "completed",
"success": bool(execution_result.get("success")),
"status": str(execution_result.get("status") or "completed"),
"error": str(execution_result.get("error") or "").strip(),
"command_text": str(execution_result.get("command_text") or "").strip(),
"kind": str(metadata.get("kind") or "").strip(),
}
return _send_mpv_ipc_command(
[
"script-message-to",
script_name,
message_name,
json.dumps(payload, ensure_ascii=False),
],
ipc_path=ipc_path,
)
def _notify_mpv_completion(metadata: Dict[str, Any], execution_result: Dict[str, Any]) -> bool:
callback_sent = _notify_mpv_callback(metadata, execution_result)
notify = metadata.get("mpv_notify") if isinstance(metadata, dict) else None
if not isinstance(notify, dict):
return False
return callback_sent
success = bool(execution_result.get("success"))
error_text = str(execution_result.get("error") or "").strip()
@@ -167,13 +210,16 @@ def _notify_mpv_completion(metadata: Dict[str, Any], execution_result: Dict[str,
message = error_text
if not message:
return False
return callback_sent
try:
duration_ms = int(notify.get("duration_ms") or 3500)
except Exception:
duration_ms = 3500
return _notify_mpv_osd(message, duration_ms=duration_ms)
ipc_path = str(notify.get("ipc_path") or "").strip() or None
notified = _notify_mpv_osd(message, duration_ms=duration_ms, ipc_path=ipc_path)
return bool(callback_sent or notified)
class _OldWorkerStages:
@@ -2246,6 +2292,7 @@ Come to love it when others take what you share, as there is no greater joy
queued_inputs: List[Dict[str, Any]] = []
queued_inputs_lock = threading.Lock()
repl_queue_stop = threading.Event()
injected_payload: Optional[Dict[str, Any]] = None
def _drain_repl_queue() -> None:
try:
@@ -2258,7 +2305,7 @@ Come to love it when others take what you share, as there is no greater joy
queued_inputs.extend(pending)
def _inject_repl_command(payload: Dict[str, Any]) -> bool:
nonlocal session
nonlocal session, injected_payload
command_text = str(payload.get("command") or "").strip()
source_text = str(payload.get("source") or "external").strip() or "external"
if not command_text or session is None:
@@ -2272,10 +2319,12 @@ Come to love it when others take what you share, as there is no greater joy
injected = False
def _apply() -> None:
nonlocal injected
nonlocal injected, injected_payload
try:
buffer = getattr(session, "default_buffer", None)
if buffer is not None:
with queued_inputs_lock:
injected_payload = payload
buffer.document = Document(text=command_text, cursor_position=len(command_text))
try:
buffer.validate_and_handle()
@@ -2283,9 +2332,13 @@ Come to love it when others take what you share, as there is no greater joy
return
except Exception:
pass
with queued_inputs_lock:
injected_payload = payload
app.exit(result=command_text)
injected = True
except Exception:
with queued_inputs_lock:
injected_payload = None
injected = False
try:
@@ -2335,6 +2388,11 @@ Come to love it when others take what you share, as there is no greater joy
user_input = ""
else:
user_input = session.prompt(prompt_text).strip()
if user_input:
with queued_inputs_lock:
if injected_payload is not None:
queued_payload = injected_payload
injected_payload = None
except (EOFError, KeyboardInterrupt):
print("He who is victorious through deceit is defeated by the truth.")
repl_queue_stop.set()
@@ -2387,6 +2445,11 @@ Come to love it when others take what you share, as there is no greater joy
"command_text": user_input,
}
print(syntax_error.message, file=sys.stderr)
if queued_metadata:
try:
_notify_mpv_completion(queued_metadata, execution_result)
except Exception:
pass
continue
except Exception:
pass
@@ -2401,6 +2464,11 @@ Come to love it when others take what you share, as there is no greater joy
"command_text": user_input,
}
print(f"Syntax error: {exc}", file=sys.stderr)
if queued_metadata:
try:
_notify_mpv_completion(queued_metadata, execution_result)
except Exception:
pass
continue
if not tokens: