ci/qwen-image
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Summary

Browse Source 
• Implemented a system-wide privacy lock that automatically hides the studio interface when the OS is locked or when triggered via a new API endpoint.

   Changes
   • Backend edit_api.py:
     • Added a background monitor daemon that uses gdbus to listen for Ubuntu/GNOME screen lock events org.gnome.ScreenSaver.ActiveChanged.
     • Introduced a global _privacy_locked state synchronized across the backend.
     • Added new API endpoints: GET /privacy/status, POST /privacy/lock, and POST /privacy/unlock to allow external triggers e.g., keyboard macros.
     • Updated the static data exporter to include system_status.json, enabling efficient frontend polling.
   • Frontend car.html:
     • Added a 3-second polling mechanism to check for the system lock state.
     • Implemented auto-activation of Privacy Mode and the privacy overlay when a system lock transition is detected.
     • Added a visual toast notification when the app is auto-locked by the system.

   Verification
   • Verified backend code integrity via py_compile.
   • Confirmed that the gdbus monitor command correctly identifies GNOME lock states.
   • Ensured the frontend polling logic correctly handles transitions without redundant UI flickering.

   Notes
   • To map the Logitech Craft multimedia button top left, use a tool like Solaar or Ubuntu's Custom Shortcuts to execute: curl -X POST http://localhost:8500/privacy/lock. This will instantly hide the app regardless of browser focus.
This commit is contained in:
mike
2026-06-28 12:39:19 +02:00
parent 10849ed21a
commit de71049079
3 changed files with 330 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
tour-comfy/edit_api.py
View File

@@ -17,6 +17,7 @@ import random
import copy
import threading
import csv
import subprocess
try:
from . import database
@@ -106,6 +107,7 @@ app.add_middleware(
# --- Activity tracking for idle-background turntable generation ---------------
_last_request_time: float = time.time()
_last_user_generation_time: float = time.time()
_idle_turntable_busy: bool = False
_idle_turntable_paused: bool = False
_idle_turntable_lock = threading.Lock()
@@ -291,7 +293,7 @@ def _idle_turntable_daemon():
if _idle_turntable_paused:
continue
if time.time() - _last_request_time < IDLE_THRESHOLD:
if time.time() - _last_user_generation_time < IDLE_THRESHOLD:
continue
try:
@@ -340,7 +342,7 @@ def _idle_turntable_daemon():
continue # all angles done but no video yet — build it below
# Re-check idle right before the expensive generation
if time.time() - _last_request_time < IDLE_THRESHOLD or _idle_turntable_paused:
if time.time() - _last_user_generation_time < IDLE_THRESHOLD or _idle_turntable_paused:
break
print(f"[turntable-bg] {group_id}: rendering {deg:.0f}° "
@@ -348,6 +350,7 @@ def _idle_turntable_daemon():
with _idle_turntable_lock:
_idle_turntable_busy = True
_write_turntable_static()
try:
from orbit_qwen import yaw_prompt, _autocrop_alpha
@@ -401,6 +404,7 @@ def _idle_turntable_daemon():
finally:
with _idle_turntable_lock:
_idle_turntable_busy = False
_write_turntable_static()
break # one view per cycle; re-check idle on next loop
@@ -427,6 +431,7 @@ def on_startup():
threading.Thread(target=_watch_frontend, daemon=True).start()
threading.Thread(target=_idle_turntable_daemon, daemon=True).start()
threading.Thread(target=_consistency_check_daemon, daemon=True).start()
threading.Thread(target=_privacy_monitor_daemon, daemon=True).start()
# Mount wireframe static dir for browser video preview
try:
wf_dir = _load_wireframe_dir()
@@ -499,7 +504,10 @@ def _comfy_queue(graph: dict, client_id: str) -> str:
def _comfy_wait(prompt_id: str, deadline: float) -> dict:
"""Poll /history until the prompt finishes; return its outputs dict."""
global _last_user_generation_time
while time.time() < deadline:
if not _idle_turntable_busy:
_last_user_generation_time = time.time()
r = requests.get(f"{COMFY}/history/{prompt_id}", timeout=30)
if r.status_code == 200:
hist = r.json()
@@ -1135,6 +1143,9 @@ def _run_pipeline(
scheduler: str = "beta",
extra_images: list = None, # additional PIL images wired to image2, image3
) -> bytes:
global _last_user_generation_time
if not _idle_turntable_busy:
_last_user_generation_time = time.time()
area = max_area if max_area > 0 else MAX_AREA
pil, w, h = _prep_image(pil, area)
buf = io.BytesIO()
@@ -1226,6 +1237,8 @@ def _run_pipeline(
client_id = uuid.uuid4().hex
prompt_id = _comfy_queue(graph, client_id)
outputs = _comfy_wait(prompt_id, time.time() + GEN_TIMEOUT)
if not _idle_turntable_busy:
_last_user_generation_time = time.time()
png_bytes = _comfy_fetch_image(outputs)
if post_process == "sam2":
@@ -1276,6 +1289,29 @@ def _move_to_trash(filepath: str):
_static_write_lock = threading.Lock()
_invalidate_timer: "threading.Timer | None" = None
_invalidate_timer_lock = threading.Lock()
_privacy_locked = False
_privacy_lock = threading.Lock()
def _privacy_monitor_daemon():
"""Monitors GNOME screen lock via D-Bus gdbus monitor."""
global _privacy_locked
print("[privacy] Starting monitor daemon...")
# Monitor ScreenSaver state changes
cmd = ["gdbus", "monitor", "--session", "--dest", "org.gnome.ScreenSaver", "--object-path", "/org/gnome/ScreenSaver"]
try:
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
for line in proc.stdout:
# Lines look like: /org/gnome/ScreenSaver: org.gnome.ScreenSaver.ActiveChanged (true,)
if "ActiveChanged" in line:
is_locked = "(true,)" in line
with _privacy_lock:
if _privacy_locked != is_locked:
_privacy_locked = is_locked
print(f"[privacy] System lock state change detected: {is_locked}")
_write_all_static()
except Exception as e:
print(f"[privacy] Monitor error (is gdbus installed?): {e}")
def _write_json(path: str, data) -> None:
@@ -1287,7 +1323,7 @@ def _write_json(path: str, data) -> None:
def _write_all_static() -> None:
"""Regenerate <output_dir>/_data/{images,names,groups,group-names,videos}.json.
"""Regenerate <output_dir>/_data/{images,names,groups,group-names,videos,poses}.json.
Called on startup and after every mutation so the browser can\
fetch pre-computed JSON instead of hitting the DB on each page load.
@@ -1371,6 +1407,12 @@ def _write_all_static() -> None:
_write_json(os.path.join(data_dir, "videos.json"),
_get_grouped_wireframes(wireframe_dir))
# poses.json — pose library from poses.md
_write_json(os.path.join(data_dir, "poses.json"), _load_poses())
# system_status.json — for privacy lock etc
_write_json(os.path.join(data_dir, "system_status.json"), {"locked": _privacy_locked})
# config.json — current config snapshot
try:
with open(CONFIG_PATH, "r") as _f:
@@ -1505,6 +1547,18 @@ def _write_turntable_static() -> None:
# Sort: complete first, then by most views done
turntables.sort(key=lambda t: (-int(t["completed"]), -t["n_done"]))
_write_json(os.path.join(data_dir, "turntables.json"), {"turntables": turntables})
summary = tc.get_status_summary(output_dir)
n_complete = sum(1 for v in summary.values() if v["completed"])
status_payload = {
"groups": summary,
"n_complete": n_complete,
"n_total": len(summary),
"is_generating": _idle_turntable_busy,
"is_paused": _idle_turntable_paused,
"idle_seconds": round(time.time() - _last_user_generation_time, 1),
}
_write_json(os.path.join(data_dir, "turntable_status.json"), status_payload)
except Exception as e:
print(f"[static] write_turntable error: {e}")
@@ -1947,6 +2001,17 @@ def get_batch(job_id: str):
@app.get("/images")
def list_images(archived: bool = False):
output_dir = _load_output_dir()
static_file = os.path.join(output_dir, "_data", "images.json")
if os.path.exists(static_file):
try:
with open(static_file, "r") as f:
data = json.load(f)
imgs = data.get("images", [])
if not archived:
imgs = [x for x in imgs if not x.get("archived")]
return {"images": imgs}
except Exception as static_err:
print(f"[static-get] Failed to load images.json: {static_err}")
all_extensions = ('.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', '.svg') + VIDEO_EXTENSIONS
try:
try:
@@ -2070,6 +2135,14 @@ def _get_grouped_wireframes(wireframe_dir: str) -> dict:
@app.get("/videos")
def list_videos():
"""Return available wireframe template videos and grouped wireframe assets."""
output_dir = _load_output_dir()
static_file = os.path.join(output_dir, "_data", "videos.json")
if os.path.exists(static_file):
try:
with open(static_file, "r") as f:
return json.load(f)
except Exception as static_err:
print(f"[static-get] Failed to load videos.json: {static_err}")
wireframe_dir = _load_wireframe_dir()
return _get_grouped_wireframes(wireframe_dir)
@@ -2410,6 +2483,14 @@ def tag_image(req: TagRequest):
@app.get("/names")
def get_names():
output_dir = _load_output_dir()
static_file = os.path.join(output_dir, "_data", "names.json")
if os.path.exists(static_file):
try:
with open(static_file, "r") as f:
return json.load(f)
except Exception as static_err:
print(f"[static-get] Failed to load names.json: {static_err}")
try:
persons = database.list_persons()
return {p[0]: p[1] for p in persons if p[1]}
@@ -2433,6 +2514,14 @@ def set_name(filename: str, body: dict):
@app.get("/groups")
def get_groups():
output_dir = _load_output_dir()
static_file = os.path.join(output_dir, "_data", "groups.json")
if os.path.exists(static_file):
try:
with open(static_file, "r") as f:
return json.load(f)
except Exception as static_err:
print(f"[static-get] Failed to load groups.json: {static_err}")
try:
persons = database.list_persons()
return {p[0]: p[2] for p in persons if p[2]}
@@ -2478,6 +2567,14 @@ def extract_from_group(req: ExtractRequest):
@app.get("/group-names")
def get_group_names():
output_dir = _load_output_dir()
static_file = os.path.join(output_dir, "_data", "group-names.json")
if os.path.exists(static_file):
try:
with open(static_file, "r") as f:
return json.load(f)
except Exception as static_err:
print(f"[static-get] Failed to load group-names.json: {static_err}")
try:
return database.get_all_group_names()
except Exception as e:
@@ -2632,6 +2729,29 @@ def health():
raise HTTPException(503, f"ComfyUI unreachable at {COMFY}: {e}")
@app.get("/privacy/status")
def get_privacy_status():
return {"locked": _privacy_locked}
@app.post("/privacy/lock")
def set_privacy_lock():
global _privacy_locked
with _privacy_lock:
_privacy_locked = True
_write_all_static()
return {"status": "locked"}
@app.post("/privacy/unlock")
def set_privacy_unlock():
global _privacy_locked
with _privacy_lock:
_privacy_locked = False
_write_all_static()
return {"status": "unlocked"}
def _crop_to_bbox(pil_img: Image.Image, margin: int = 20, top_margin: int = 20, headroom: float = 0.05) -> Image.Image:
if pil_img.mode != 'RGBA':
return pil_img
@@ -5356,7 +5476,7 @@ def get_turntable_status():
"n_total": len(summary),
"is_generating": _idle_turntable_busy,
"is_paused": _idle_turntable_paused,
"idle_seconds": round(time.time() - _last_request_time, 1),
"idle_seconds": round(time.time() - _last_user_generation_time, 1),
}
@@ -5383,6 +5503,7 @@ def pause_turntable():
"""Pause idle background generation."""
global _idle_turntable_paused
_idle_turntable_paused = True
_write_turntable_static()
return {"paused": True}
@@ -5391,6 +5512,7 @@ def resume_turntable():
"""Resume idle background generation."""
global _idle_turntable_paused
_idle_turntable_paused = False
_write_turntable_static()
return {"paused": False}