371 lines
19 KiB
Python
371 lines
19 KiB
Python
"""Real reader/preferences/GTK flow against a private, completion-aware SSIP server."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
|
|
from native_x11 import wait_for
|
|
|
|
|
|
class PreviewSpeechServer:
|
|
def __init__(self, path, defaultsSupported=True):
|
|
self.listener = socket.socket(socket.AF_UNIX)
|
|
self.listener.bind(str(path))
|
|
self.listener.listen()
|
|
self.lock = threading.RLock()
|
|
self.clients = {}
|
|
self.spoken = []
|
|
self.defaultsSupported = defaultsSupported
|
|
threading.Thread(target=self.accept_clients, daemon=True).start()
|
|
|
|
def accept_clients(self):
|
|
while True:
|
|
try:
|
|
connection, _ = self.listener.accept()
|
|
except OSError:
|
|
return
|
|
with self.lock:
|
|
clientId = len(self.clients) + 1
|
|
client = {"socket": connection, "lock": threading.RLock(), "notifications": False,
|
|
"module": "main", "voice": "", "active": None,
|
|
"settings": {"RATE": 20, "PITCH": -10, "VOLUME": 100}}
|
|
self.clients[clientId] = client
|
|
threading.Thread(target=self.serve_client, args=(clientId, client), daemon=True).start()
|
|
|
|
def reply(self, client, text):
|
|
with client["lock"]:
|
|
try:
|
|
client["socket"].sendall(text.encode())
|
|
except OSError:
|
|
pass
|
|
|
|
def finish(self, clientId, client, record, canceled=False):
|
|
with self.lock:
|
|
if client["active"] is not record:
|
|
return
|
|
client["active"] = None
|
|
record["finished"], record["canceled"] = True, canceled
|
|
if client["notifications"]:
|
|
code = 703 if canceled else 702
|
|
self.reply(client, f"{code}-1\r\n{code}-{clientId}\r\n{code} DONE\r\n")
|
|
|
|
def serve_client(self, clientId, client):
|
|
try:
|
|
with client["socket"].makefile("rb") as stream:
|
|
for raw in stream:
|
|
command = raw.decode().strip()
|
|
response = "200 OK\r\n"
|
|
if command == "HISTORY GET CLIENT_ID":
|
|
response = f"245-{clientId}\r\n245 OK\r\n"
|
|
elif command == "LIST OUTPUT_MODULES":
|
|
response = "200-echo-a\r\n200-echo-b\r\n200-main\r\n200 OK\r\n"
|
|
elif command == "LIST SYNTHESIS_VOICES":
|
|
voice = {"echo-a": "Echo A", "echo-b": "Echo B", "main": "Main"}[client["module"]]
|
|
response = f"200-{voice}\ten\tnone\r\n200 OK\r\n"
|
|
elif command.startswith("GET "):
|
|
field = command.removeprefix("GET ")
|
|
response = (f"200-{client['settings'][field]}\r\n200 OK\r\n"
|
|
if self.defaultsSupported and field in client["settings"] else "500 UNSUPPORTED\r\n")
|
|
elif any(command.startswith("SET SELF " + field + " ") for field in client["settings"]):
|
|
_, _, field, value = command.split()
|
|
client["settings"][field] = int(value)
|
|
elif command.startswith("SET SELF OUTPUT_MODULE "):
|
|
client["module"] = command.removeprefix("SET SELF OUTPUT_MODULE ")
|
|
elif command.startswith("SET SELF SYNTHESIS_VOICE "):
|
|
client["voice"] = command.removeprefix("SET SELF SYNTHESIS_VOICE ")
|
|
elif command.startswith("SET SELF NOTIFICATION ALL "):
|
|
client["notifications"] = command.endswith(" on")
|
|
elif command.startswith("CANCEL "):
|
|
with self.lock:
|
|
targetId = int(command.split()[1])
|
|
target = self.clients[targetId]
|
|
if target["active"] is not None:
|
|
self.finish(targetId, target, target["active"], canceled=True)
|
|
elif command == "SPEAK" or command.startswith("CHAR "):
|
|
parts = []
|
|
if command == "SPEAK":
|
|
for line in stream:
|
|
if line == b".\r\n":
|
|
break
|
|
parts.append(line.decode().rstrip("\r\n"))
|
|
else:
|
|
parts.append(command.removeprefix("CHAR "))
|
|
record = {"text": "\n".join(parts), "module": client["module"],
|
|
"voice": client["voice"], "finished": False, "canceled": False,
|
|
"settings": dict(client["settings"])}
|
|
with self.lock:
|
|
self.spoken.append(record)
|
|
client["active"] = record
|
|
self.reply(client, "230 OK\r\n200 QUEUED\r\n" if command == "SPEAK" else "225 QUEUED\r\n")
|
|
if client["notifications"]:
|
|
self.reply(client, f"701-1\r\n701-{clientId}\r\n701 BEGIN\r\n")
|
|
timer = threading.Timer(0.25, self.finish, args=(clientId, client, record))
|
|
timer.daemon = True
|
|
timer.start()
|
|
continue
|
|
self.reply(client, response)
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
def snapshot(self):
|
|
with self.lock:
|
|
return [dict(record) for record in self.spoken]
|
|
|
|
def close(self):
|
|
self.listener.close()
|
|
for client in self.clients.values():
|
|
client["socket"].close()
|
|
|
|
|
|
def check_radio_buttons(controls, tabs, server, readerLog):
|
|
import pyatspi
|
|
|
|
def radio_cues():
|
|
records = [json.loads(line) for line in readerLog.read_text().splitlines() if line.startswith("{")]
|
|
return [record["cue"] for record in records if record.get("message") == "playing object sound"
|
|
and record.get("cue", "").startswith("radiobutton_")]
|
|
|
|
assert tabs.querySelection().selectChild(0)
|
|
assert controls["Desktop"].queryComponent().grabFocus()
|
|
wait_for(lambda: radio_cues(), "initial radio sound did not play")
|
|
time.sleep(0.35)
|
|
for keyName, target in (("Right", "Laptop"), ("Left", "Desktop"), ("Right", "Laptop")):
|
|
cueOffset = len(radio_cues())
|
|
speechOffset = len(server.snapshot())
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", keyName], check=True)
|
|
wait_for(lambda: controls[target].getState().contains(pyatspi.STATE_CHECKED), target + " was not selected")
|
|
wait_for(lambda: len(radio_cues()) > cueOffset, target + " selection sound did not play")
|
|
# Cover the deferred focus cue and duplicate focus event after checking.
|
|
time.sleep(0.35)
|
|
cues = radio_cues()[cueOffset:]
|
|
assert cues == ["radiobutton_checked"], (keyName, target, cues)
|
|
spoken = [record["text"] for record in server.snapshot()[speechOffset:]]
|
|
assert any(target in text for text in spoken), target + " label was lost"
|
|
assert not any("radio button" in text for text in spoken), (target, spoken)
|
|
records = [json.loads(line) for line in readerLog.read_text().splitlines() if line.startswith("{")]
|
|
errors = [record for record in records if record.get("message") == "play sound"
|
|
and record.get("cue", "").startswith("radiobutton_") and record.get("error") != "context canceled"]
|
|
assert not errors, errors
|
|
|
|
|
|
def check_flat_review(tabs, server, readerLog):
|
|
def records():
|
|
return [json.loads(line) for line in readerLog.read_text().splitlines() if line.startswith("{")]
|
|
|
|
def key(*keys):
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", *keys], check=True)
|
|
|
|
assert tabs.querySelection().selectChild(0)
|
|
key("ctrl+Page_Down", "ctrl+Page_Up")
|
|
wait_for(lambda: any(record["text"] == "General" for record in server.snapshot()) and
|
|
server.snapshot()[-1]["finished"], "General tab was not announced")
|
|
time.sleep(0.4)
|
|
for command in ("KP_Up", "KP_Page_Up", "KP_Right", "KP_Page_Down", "KP_End", "KP_Home"):
|
|
logOffset = len(records())
|
|
speechOffset = len(server.snapshot())
|
|
key(command)
|
|
def review_delivered():
|
|
return any(record.get("message") == "delivering accessible presentation" and
|
|
record.get("cause") == "flat-review" for record in records()[logOffset:])
|
|
wait_for(review_delivered, "flat review did not speak: " + command)
|
|
wait_for(lambda: len(server.snapshot()) > speechOffset and server.snapshot()[-1]["finished"],
|
|
"review speech did not finish: " + command)
|
|
# Include the focus replay after the key grab has been released.
|
|
time.sleep(0.4)
|
|
deliveries = [record for record in records()[logOffset:]
|
|
if record.get("message") == "delivering accessible presentation"]
|
|
assert deliveries and all(record.get("cause") == "flat-review" for record in deliveries), deliveries
|
|
assert not any(record["canceled"] for record in server.snapshot()[speechOffset:])
|
|
assert tabs.querySelection().getSelectedChild(0).name == "General"
|
|
|
|
speechOffset = len(server.snapshot())
|
|
nextTab = tabs[1].name
|
|
key("Right")
|
|
wait_for(lambda: any(record["text"] == nextTab for record in server.snapshot()[speechOffset:]) and
|
|
server.snapshot()[-1]["finished"], "new tab focus was suppressed")
|
|
speechOffset = len(server.snapshot())
|
|
key("Tab")
|
|
wait_for(lambda: len(server.snapshot()) > speechOffset and server.snapshot()[-1]["finished"],
|
|
"Tab into the page did not speak")
|
|
speechOffset = len(server.snapshot())
|
|
key("shift+Tab")
|
|
wait_for(lambda: any(record["text"] == nextTab for record in server.snapshot()[speechOffset:]) and
|
|
server.snapshot()[-1]["finished"], "return to tab header was suppressed")
|
|
|
|
|
|
def check_interruption_summary(controls, tabs, server):
|
|
import pyatspi
|
|
|
|
def key(*keys):
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", *keys], check=True)
|
|
|
|
assert tabs.querySelection().selectChild(2)
|
|
summary = controls["Defined interruption keys"]
|
|
assert summary.queryText().getText(0, -1) == "Tab, Enter"
|
|
assert not summary.getState().contains(pyatspi.STATE_EDITABLE)
|
|
assert controls["Record interruption keys"].queryComponent().grabFocus()
|
|
offset = len(server.snapshot())
|
|
key("Tab")
|
|
wait_for(lambda: any(record["text"] == "Tab, Enter" and record["finished"]
|
|
for record in server.snapshot()[offset:]), "read-only summary contents were not spoken on focus")
|
|
assert any(record["text"] == "Defined interruption keys" for record in server.snapshot()[offset:])
|
|
key("Home")
|
|
for keyName, character, caret in (("Right", "a", 1), ("Left", "T", 0)):
|
|
offset = len(server.snapshot())
|
|
key(keyName)
|
|
wait_for(lambda: summary.queryText().caretOffset == caret, "summary caret did not move for " + keyName)
|
|
wait_for(lambda: any(record["text"] == character and record["finished"]
|
|
for record in server.snapshot()[offset:]), "summary character was not spoken for " + keyName)
|
|
key("shift+Tab")
|
|
wait_for(lambda: controls["Record interruption keys"].getState().contains(pyatspi.STATE_FOCUSED),
|
|
"read-only summary trapped Shift+Tab")
|
|
|
|
|
|
def check_runtime(radioOnly=False, flatReviewOnly=False, interruptionOnly=False, noConfigOnly=False):
|
|
import pyatspi
|
|
directory = Path(tempfile.mkdtemp(prefix="cthulhu-echo-runtime-"))
|
|
binary = os.environ["CTHULHU_BINARY"]
|
|
wmConfig = directory / "i3.config"
|
|
wmConfig.write_text(f"font pango:monospace 10\nipc-socket {directory / 'i3.sock'}\n")
|
|
server = PreviewSpeechServer(directory / "speech.sock", defaultsSupported=os.environ.get("CTHULHU_PREFS_DEFAULTS_UNSUPPORTED") != "1")
|
|
wm = subprocess.Popen(["i3", "-c", str(wmConfig)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
reader = None
|
|
try:
|
|
wait_for(lambda: "0x" in subprocess.check_output(["xprop", "-root", "_NET_SUPPORTING_WM_CHECK"], text=True), "private WM missing")
|
|
configPath = directory / "settings.toml"
|
|
if not noConfigOnly:
|
|
configPath.write_text('''schema-version = 2
|
|
active-profile = "default"
|
|
[profiles.default.speech]
|
|
output-module = "main"
|
|
[profiles.default.braille]
|
|
enabled = false
|
|
[profiles.default.sound]
|
|
enabled = false
|
|
[profiles.default.key-echo]
|
|
enabled = false
|
|
custom-voice = true
|
|
custom-speech-server = true
|
|
speech-server = "echo-a"
|
|
[profiles.default.key-echo.voice]
|
|
rate = 50
|
|
average-pitch = 5
|
|
gain = 5
|
|
[profiles.default.key-echo.voice.family]
|
|
name = "Echo A"
|
|
lang = "en"
|
|
''')
|
|
if interruptionOnly:
|
|
with configPath.open("a") as settingsFile:
|
|
settingsFile.write('\n[profiles.default.speech.interruption]\nmode = "custom"\nkeys = ["Tab", "Return"]\n')
|
|
if radioOnly:
|
|
configPath.write_text(configPath.read_text().replace(
|
|
"[profiles.default.sound]\nenabled = false",
|
|
'[profiles.default.sound]\nenabled = true\nsink = "auto"\nplay-for-state = true\nrole-presentation-strategy = "sound_only"'))
|
|
readerEnv = os.environ.copy()
|
|
if noConfigOnly:
|
|
# Preferences checks must not launch Wine Access in live prefixes.
|
|
readerEnv["CTHULHU_WINE_ACCESS_HELPER"] = "/usr/bin/true"
|
|
if radioOnly or noConfigOnly:
|
|
import gi
|
|
gi.require_version("Gst", "1.0")
|
|
from gi.repository import Gst
|
|
Gst.init(None)
|
|
assert Gst.ElementFactory.find("fakeaudiosink"), "silent GStreamer audio sink is unavailable"
|
|
# Only the child reader sees these ranks. Auto audio can choose
|
|
# the fake sink; every real audio sink is excluded from discovery.
|
|
ranks = []
|
|
for factory in Gst.Registry.get().get_feature_list(Gst.ElementFactory):
|
|
kind = factory.get_metadata("klass") or ""
|
|
if "Audio" in kind and "Sink" in kind:
|
|
rank = "MAX" if factory.get_name() == "fakeaudiosink" else "0"
|
|
ranks.append(factory.get_name() + ":" + rank)
|
|
readerEnv["GST_PLUGIN_FEATURE_RANK"] = ",".join(ranks)
|
|
reader = subprocess.Popen([binary, "--config", str(configPath), "--settings", "--no-tentacles",
|
|
"--speech-socket", str(directory / "speech.sock"),
|
|
"--debug-file", str(directory / "reader.log")],
|
|
env=readerEnv, stdout=subprocess.DEVNULL, stderr=(directory / "stderr.log").open("w"))
|
|
def find_window():
|
|
for application in pyatspi.Registry.getDesktop(0):
|
|
for window in application:
|
|
if window and window.name == "Cthulhu Preferences":
|
|
return window
|
|
window = wait_for(find_window, "reader's preferences did not open")
|
|
def walk(node):
|
|
yield node
|
|
for child in node:
|
|
if child is not None:
|
|
yield from walk(child)
|
|
controls = {node.name: node for node in walk(window) if node.name}
|
|
xid = subprocess.check_output(["xdotool", "search", "--onlyvisible", "--name", "^Cthulhu Preferences$"], text=True).splitlines()[0]
|
|
subprocess.run(["xdotool", "windowactivate", "--sync", xid], check=True)
|
|
tabs = next(node for node in walk(window) if node.getRole() == pyatspi.ROLE_PAGE_TAB_LIST)
|
|
if noConfigOnly:
|
|
from native_speech_defaults import check_no_config
|
|
check_no_config(window, tabs, server, configPath)
|
|
print(f"PASS: missing config, editable speech sliders, server defaults/fallback, inheritance, Apply/save/reopen; evidence: {directory}")
|
|
return
|
|
if interruptionOnly:
|
|
check_interruption_summary(controls, tabs, server)
|
|
print(f"PASS: real reader speaks read-only interruption summary on focus and character arrows; Tab traversal preserved; evidence: {directory}")
|
|
return
|
|
if flatReviewOnly:
|
|
check_flat_review(tabs, server, directory / "reader.log")
|
|
print(f"PASS: notebook flat review has no repeated tab speech; tab changes and focus returns speak; evidence: {directory}")
|
|
return
|
|
if radioOnly:
|
|
check_radio_buttons(controls, tabs, server, directory / "reader.log")
|
|
print(f"PASS: real GTK Desktop/Laptop arrows play only the final selected sound and retain labels; evidence: {directory}")
|
|
return
|
|
assert tabs.querySelection().selectChild(4)
|
|
assert controls["Use custom echo voice"].queryComponent().grabFocus()
|
|
wait_for(lambda: any(record["text"] == "checked" and record["finished"]
|
|
for record in server.snapshot()), "initial checkbox announcement did not finish")
|
|
time.sleep(0.3)
|
|
def key(*keys):
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", *keys], check=True)
|
|
def check_output(keys, expected):
|
|
offset = len(server.snapshot())
|
|
key(*keys)
|
|
def completed():
|
|
records = server.snapshot()[offset:]
|
|
matching = [record for record in records if record["text"] in {text for text, _, _ in expected}]
|
|
return len(matching) >= len(expected) and all(record["finished"] for record in matching)
|
|
wait_for(completed, f"preview did not complete: {keys}")
|
|
# The original delayed menu-item event followed its preview by up
|
|
# to 160 ms. Require quiescence beyond that delay, then inspect all
|
|
# speech requests, including duplicates in the main voice.
|
|
time.sleep(0.25)
|
|
records = server.snapshot()[offset:]
|
|
got = [(record["text"], record["module"], record["voice"]) for record in records]
|
|
assert got == expected, (keys, got, expected)
|
|
assert not any(record["canceled"] for record in records), (keys, records)
|
|
check_output(("Tab",), [("Key echo speech synthesizer", "main", ""), ("echo-a", "echo-a", "Echo A")])
|
|
check_output(("Down",), [("echo-b", "echo-b", "")])
|
|
check_output(("Tab",), [("Key echo person", "main", ""), ("Default", "echo-b", "")])
|
|
check_output(("Down",), [("Echo B (en)", "echo-b", "Echo B")])
|
|
for control in ("rate", "pitch", "volume"):
|
|
value = "50.0" if control == "rate" else "5.0"
|
|
check_output(("Tab",), [("Key echo " + control, "main", ""), (value, "echo-b", "Echo B")])
|
|
for control in ("pitch", "rate", "person", "synthesizer"):
|
|
value = {"pitch": "5.0", "rate": "50.0", "person": "Echo B (en)", "synthesizer": "echo-b"}[control]
|
|
label = "Key echo speech synthesizer" if control == "synthesizer" else "Key echo " + control
|
|
check_output(("shift+Tab",), [(label, "main", ""), (value, "echo-b", "Echo B")])
|
|
print(f"PASS: real reader + GTK + SSIP focus/selection previews, voice completion, no duplicate menu-item speech; evidence: {directory}")
|
|
finally:
|
|
if reader is not None:
|
|
reader.terminate()
|
|
try:
|
|
reader.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
reader.kill()
|
|
reader.wait(timeout=5)
|
|
wm.terminate()
|
|
wm.wait(timeout=5)
|
|
server.close()
|
|
(directory / "speech.json").write_text(json.dumps(server.snapshot(), indent=2))
|