Skip to content

limit queue size to 10 images for realtime processing #390

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ralph-sj wants to merge 4 commits into
base: main
Choose a base branch
from
Open

limit queue size to 10 images for realtime processing #390

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2270696
limit queue size to 10 images for realtime processing
ralph-sj Jun 16, 2026
8eb10c7
fix formatting
ralph-sj Jun 16, 2026
2c9eba5
bump version
ralph-sj Jun 16, 2026
f2f1687
Merge branch 'main' into process-realtime-without-backlog
ralph-sj Jun 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 7 additions & 6 deletions pyopia/realtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import logging
import pathlib
import queue
from collections import deque
import threading
import time

Expand Down Expand Up @@ -46,7 +47,7 @@ def _enqueue_file_if_new(
return False
seen_files.add(file_key)

file_queue.put(file_path)
file_queue.append(file_path)
return True


Expand Down Expand Up @@ -103,8 +104,9 @@ def _worker_loop(
):
while not stop_event.is_set():
try:
filepath = file_queue.get(timeout=1)
except queue.Empty:
filepath = file_queue.pop()
except IndexError:
time.sleep(0.1)
continue

try:
Expand All @@ -123,7 +125,7 @@ def _worker_loop(
finally:
with state_lock:
runtime_state["current_file"] = "idle"
file_queue.task_done()
time.sleep(0.1)


def run_realtime(pipeline_config: dict, watch_folder: str | None = None):
Expand All @@ -145,7 +147,7 @@ def run_realtime(pipeline_config: dict, watch_folder: str | None = None):

processing_pipeline = pyopia.pipeline.Pipeline(pipeline_config)

file_queue = queue.Queue()
file_queue = deque(maxlen=10)
stop_event = threading.Event()
seen_files: set[str] = set()
seen_lock = threading.Lock()
Expand Down Expand Up @@ -212,7 +214,6 @@ def run_realtime(pipeline_config: dict, watch_folder: str | None = None):
description=(
"[blue]Realtime active"
f" | processed: {processed_count}"
f" | queued: {file_queue.qsize()}"
f" | current: {current_file}"
),
)
Expand Down
Loading