From 4e933b0ba6854c0439a02ebe4166aa1c95b15145 Mon Sep 17 00:00:00 2001 From: Marcel Schmutz Date: Mon, 8 Jun 2026 16:20:54 +0200 Subject: [PATCH 1/5] Unit test and filter chain implementation --- .github/workflows/docs.yml | 39 ++ .github/workflows/unit-tests.yml | 39 ++ docs/_config.yml | 3 + docs/index.md | 12 + pyproject.toml | 9 +- requirements-dev.txt | 6 + requirements.txt | 2 +- src/py3r/pose/cli/__init__.py | 32 +- src/py3r/pose/cli/limit_rate.py | 29 +- src/py3r/pose/cli/prediction_job.py | 100 ++--- src/py3r/pose/cli/render_job.py | 26 +- .../pose/core/filtering/abc/pose_filter.py | 4 +- src/py3r/pose/core/filtering/arena_filter.py | 39 +- .../pose/core/filtering/confidence_filter.py | 6 +- .../core/filtering/instance_type_filter.py | 4 +- src/py3r/pose/core/filtering/median_filter.py | 18 +- .../pose/core/filtering/sequential_filter.py | 12 +- tests/__init__.py | 0 tests/test_prediction_job.py | 366 ++++++++++++++++++ 19 files changed, 598 insertions(+), 148 deletions(-) create mode 100644 .github/workflows/docs.yml create mode 100644 .github/workflows/unit-tests.yml create mode 100644 docs/_config.yml create mode 100644 docs/index.md create mode 100644 requirements-dev.txt create mode 100644 tests/__init__.py create mode 100644 tests/test_prediction_job.py diff --git a/.github/workflows/docs.yml b/.github/workflows/docs.yml new file mode 100644 index 0000000..b4a0571 --- /dev/null +++ b/.github/workflows/docs.yml @@ -0,0 +1,39 @@ +name: Docs on Release + +on: + release: + types: [published] + +permissions: + contents: write + +jobs: + deploy-docs: + # Only deploy docs for non-prerelease releases + if: ${{ !github.event.release.prerelease }} + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install docs dependencies + run: | + python -m pip install -U pip + # Install only the docs extras — no need for the full runtime on the CI runner + pip install 'mkdocs-material >= 9.5' 'mkdocstrings[python] >= 0.25' 'mkdocs-autorefs >= 1.0' + # Install the package itself (needed so mkdocstrings can import and introspect it), + # but skip heavy runtime deps that require native libs or Windows-only packages. + pip install -e . --no-deps + + - name: Build and deploy docs + run: | + git config user.name "github-actions" + git config user.email "github-actions@users.noreply.github.com" + mkdocs gh-deploy --force diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml new file mode 100644 index 0000000..0252f67 --- /dev/null +++ b/.github/workflows/unit-tests.yml @@ -0,0 +1,39 @@ +name: Unit Tests + +on: + push: + pull_request: + +jobs: + test: + name: Python ${{ matrix.python-version }} / ${{ matrix.os }} + runs-on: ${{ matrix.os }} + + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest] + python-version: ["3.9", "3.10", "3.11", "3.12"] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + # opencv-python needs libGL on headless Ubuntu runners. + - name: Install system dependencies (Linux) + if: runner.os == 'Linux' + run: sudo apt-get update -q && sudo apt-get install -y libgl1 + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e . + pip install -r requirements-dev.txt + + - name: Run unit tests + run: python -m pytest tests/unit/ -q --tb=short + diff --git a/docs/_config.yml b/docs/_config.yml new file mode 100644 index 0000000..5730c3f --- /dev/null +++ b/docs/_config.yml @@ -0,0 +1,3 @@ +title: Py3R Pose Tracking +description: Pose estimation and analysis library for the ETH Zurich 3R Hub +theme: minima diff --git a/docs/index.md b/docs/index.md new file mode 100644 index 0000000..ca82e31 --- /dev/null +++ b/docs/index.md @@ -0,0 +1,12 @@ +--- +title: Home +nav_order: 1 +--- + +# Py3R Pose Tracking + +Pose estimation library for the ETH Zurich 3R Hub. + +## Pages + + diff --git a/pyproject.toml b/pyproject.toml index 3b00361..c0f91a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,17 +4,17 @@ build-backend = "setuptools.build_meta" [project] name = "py3r_pose" -version = "0.7.0" +version = "1.0.0" authors = [ { name="Marcel Schmutz", email="mschmut@ethz.ch" }, ] description = "Library for pose estimation and pose analysis in the ETH Zurich 3R Hub" readme = "README.md" -requires-python = ">=3.8" +requires-python = ">=3.9" license = "MIT" license-files = ["LICEN[CS]E*"] dependencies = [ - "py3r_media @ git+https://github.com/ETHZ-INS/Py3R-Media.git@v0.5.0", + "py3r_media @ git+https://github.com/ETHZ-INS/Py3R-Media.git@v0.7.0", "numpy >= 2", "opencv-python >= 4", "reactivex >= 4.0.0", @@ -22,6 +22,9 @@ dependencies = [ ] [project.optional-dependencies] +dev = [ + "pytest >= 8.0", +] yolo = [ "ultralytics >= 8.0.0", "torch>=2", diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..6c296f1 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,6 @@ +py3r_media @ git+https://github.com/ETHZ-INS/Py3R-Media.git@master +numpy>=2 +opencv-python>=4 +reactivex>=4.0.0 +pyyaml>=6.0.0 +pytest>=8.0 diff --git a/requirements.txt b/requirements.txt index 59e293a..4f3ade7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,4 @@ py3r_media @ git+https://github.com/ETHZ-INS/Py3R-Media.git@master numpy>=2 opencv-python>=4 reactivex>=4.0.0 -pyyaml>=6.0.0 \ No newline at end of file +pyyaml>=6.0.0 diff --git a/src/py3r/pose/cli/__init__.py b/src/py3r/pose/cli/__init__.py index 81ce9ea..1496b97 100644 --- a/src/py3r/pose/cli/__init__.py +++ b/src/py3r/pose/cli/__init__.py @@ -205,9 +205,15 @@ def run_track(args): job.set_pose_renderer(pose_renderer) job.set_tracker(tracker) - job.set_live_preview(args.live_preview) + if args.live_preview: + from py3r.pose.cli.image_display import OpenCVImageDisplay + job.set_live_preview_display(OpenCVImageDisplay("Live Preview")) job.set_quiet(args.quiet) else: + from py3r.pose.yolo.model.staged_yolo_pose_model import StagedYoloPoseModel + from py3r.pose.core.serialization.dynamic_csv_writer import DynamicPoseCSVWriter + from py3r.media.video.ffmpeg_video_file_writer import FFmpegVideoFileWriter + models = [build_model(model_identifier) for model_identifier in args.model] if len(models) == 1: @@ -217,7 +223,8 @@ def run_track(args): pose_renderer = build_pose_renderer(args, model.instance_types) - job = PredictionJob(model) + staged_model = StagedYoloPoseModel(model, max_batch=args.batch_size, input_channels=1 if args.grayscale else 3) + job = PredictionJob(staged_model) job.set_source(source) if args.original_speed: @@ -226,8 +233,13 @@ def run_track(args): job.set_start_frame(start_frame) job.set_end_frame(end_frame) - job.set_output_file(output_file) - job.set_visualization_file(vis_file) + if output_file is not None: + output_file.parent.mkdir(parents=True, exist_ok=True) + job.set_pose_writer(DynamicPoseCSVWriter(output_file)) + + if vis_file is not None: + vis_file.parent.mkdir(parents=True, exist_ok=True) + job.set_video_writer(FFmpegVideoFileWriter(vis_file, size=source.get_size(), fps=source.get_fps(), quality="medium", grayscale=False)) job.set_pose_renderer(pose_renderer) job.set_batch_size(args.batch_size) @@ -235,7 +247,13 @@ def run_track(args): #job.set_label_filter(label_filter) job.set_tracker(tracker) - job.set_live_preview(args.live_preview) + if args.filter: + from py3r.pose.cli.filter_chain_builder import build_sequential_filter + job.set_pose_filter(build_sequential_filter(args.filter)) + + if args.live_preview: + from py3r.pose.cli.image_display import OpenCVImageDisplay + job.set_live_preview_display(OpenCVImageDisplay("Live Preview")) job.set_no_progress(args.quiet) if len(input_files) > 1 and not args.quiet: @@ -346,6 +364,8 @@ def _track_arguments(parser): parser.add_argument("--render", action="store_true", help="Use existing pose data to create visualization / live preview") + parser.add_argument("--filter", type=str, default=None, help="Filter chain string (e.g. 'smooth | confidence_threshold[0.5]')") + return parser @@ -433,8 +453,6 @@ def main(): run_concat([Path(input_file) for input_file in args.input_files], args.output_file, args.reset_frame_index) elif args.command == "merge": run_merge([Path(input_file) for input_file in args.input_files], args.output_file) - elif args.command == "stereo_track": - run_stereo_track diff --git a/src/py3r/pose/cli/limit_rate.py b/src/py3r/pose/cli/limit_rate.py index 8f4f171..1218603 100644 --- a/src/py3r/pose/cli/limit_rate.py +++ b/src/py3r/pose/cli/limit_rate.py @@ -34,7 +34,9 @@ def _subscribe( ) -> DisposableBase: lock = threading.RLock() disposed = False - next_due = time.monotonic() + # Initialised lazily on the first item so that startup delays + # (scheduler overhead, skipped frames, etc.) don't build up debt. + next_due = None def dispose_once() -> None: nonlocal disposed @@ -44,19 +46,18 @@ def dispose_once() -> None: def on_next(item: T) -> None: nonlocal next_due - while True: - with lock: - if disposed: - return - - now = time.monotonic() - wait = next_due - now - if wait <= 0: - # Reserve the next slot before calling downstream. - next_due = max(next_due + period, now + period) - break - - # Sleep outside the lock so dispose() is not blocked. + with lock: + if disposed: + return + # Lazy init: anchor the clock to the first item's arrival + # so any pre-pipeline delay doesn't build up catch-up debt. + if next_due is None: + next_due = time.monotonic() - period + next_due += period + due = next_due # capture for use outside the lock + + wait = due - time.monotonic() + if wait > 0: time.sleep(wait) with lock: diff --git a/src/py3r/pose/cli/prediction_job.py b/src/py3r/pose/cli/prediction_job.py index 7cdffc6..fb7ab06 100644 --- a/src/py3r/pose/cli/prediction_job.py +++ b/src/py3r/pose/cli/prediction_job.py @@ -1,18 +1,13 @@ import time from concurrent.futures import Future -from pathlib import Path from typing import List, Optional -import cv2 from py3r.media.streaming.observables.reader_observable import reader_observable from py3r.media.streaming.operators import observe_on_bounded, finally_future, adaptive_pace -from py3r.media.streaming.operators.opencv_imshow import opencv_imshow from py3r.media.streaming.operators.write_to import write_to from py3r.media.video import VideoSource -from py3r.media.video.ffmpeg_video_file_writer import FFmpegVideoFileWriter from py3r.pose.core.filtering.abc.pose_filter import IPoseFilter -from py3r.pose.core.model.pose_model import PoseModel -from py3r.pose.core.serialization.dynamic_csv_writer import DynamicPoseCSVWriter +from py3r.pose.core.model.abc.pose_model import IPoseModel from py3r.pose.core.streaming.filter_poses import filter_poses from py3r.pose.core.streaming.predict_poses import predict_poses from py3r.pose.core.streaming.render_poses import render_poses @@ -20,21 +15,20 @@ from py3r.pose.core.types import VideoFramePoses from py3r.pose.core.visualization.pose_renderer import PoseRenderer -from py3r.pose.yolo.model.staged_yolo_pose_model import StagedYoloPoseModel import reactivex as rx from reactivex import operators as ops, Subject from reactivex.disposable import CompositeDisposable from reactivex.scheduler import EventLoopScheduler +from py3r.pose.cli.image_display import IImageDisplay, display_image from py3r.pose.cli.limit_rate import limit_rate from py3r.pose.cli.operators import ensure_3_channel -from py3r.pose.cli.preview_pace import preview_pace from py3r.pose.cli.progress_bar_observer import ProgressBarObserver class PredictionJob: - def __init__(self, model: PoseModel): + def __init__(self, model: IPoseModel): self.pose_model = model self.batch_size = 4 @@ -49,10 +43,10 @@ def __init__(self, model: PoseModel): self.pose_renderer: Optional[PoseRenderer] = None self.tracker: Optional[FixedInstancesTracker] = None - self.output_file: Optional[Path] = None - self.live_preview: bool = False - self.visualization_file: Optional[Path] = None + self.pose_writer = None + self.video_writer = None + self.live_preview_display: Optional[IImageDisplay] = None self.visualization_length_frames: Optional[int] = None self.no_progress: bool = False @@ -83,14 +77,14 @@ def set_tracker(self, tracker: FixedInstancesTracker): def set_pose_renderer(self, pose_renderer: PoseRenderer): self.pose_renderer = pose_renderer - def set_output_file(self, output_file: Path): - self.output_file = output_file + def set_pose_writer(self, writer): + self.pose_writer = writer - def set_live_preview(self, live_preview: bool): - self.live_preview = live_preview + def set_live_preview_display(self, display: IImageDisplay): + self.live_preview_display = display - def set_visualization_file(self, visualization_file: Path): - self.visualization_file = visualization_file + def set_video_writer(self, writer): + self.video_writer = writer def set_visualization_length_frames(self, visualization_length_frames: Optional[int]): self.visualization_length_frames = visualization_length_frames @@ -108,16 +102,16 @@ def run(self): drains: List[Future] = [] main_scheduler = EventLoopScheduler() - camera_scheduler = EventLoopScheduler() pose_estimation_scheduler = EventLoopScheduler() + pose_results_scheduler = EventLoopScheduler() schedulers.add(main_scheduler) - schedulers.add(camera_scheduler) schedulers.add(pose_estimation_scheduler) + schedulers.add(pose_results_scheduler) stop = Subject() - frames = reader_observable(self.source, read_timeout_seconds=30.0).pipe( + frames = reader_observable(self.source, read_timeout_seconds=10.0).pipe( ops.take_while(lambda x: x is not None), ops.skip(self.start_frame) ) @@ -140,18 +134,16 @@ def run(self): ops.publish() ) - frame_images = frames.pipe(ops.map(lambda x: x.img)) - #frame_images = frame_images.pipe(ops.map(lambda x: cv2.cvtColor(x, cv2.COLOR_BGR2RGB))) + frame_images = frames.pipe(ops.map(lambda x: x.img), ops.share()) if grayscale: - color_frame_images = frame_images.pipe(ensure_3_channel) + color_frame_images = frame_images.pipe(ensure_3_channel, ops.share()) else: color_frame_images = frame_images - pose_model = StagedYoloPoseModel(self.pose_model, max_batch=self.batch_size, input_channels=1 if grayscale else 3) poses = frame_images.pipe( observe_on_bounded(pose_estimation_scheduler, maxsize=30), - predict_poses(pose_model, batch_size=self.batch_size), - observe_on_bounded(main_scheduler, maxsize=30), + predict_poses(self.pose_model, batch_size=self.batch_size), + observe_on_bounded(pose_results_scheduler, maxsize=30), ) if self.pose_filter is not None: @@ -162,24 +154,20 @@ def run(self): poses = poses.pipe(ops.share()) - if self.output_file is not None: - self.output_file.parent.mkdir(parents=True, exist_ok=True) - + if self.pose_writer is not None: pose_results_done = Future() drains.append(pose_results_done) - pose_results_scheduler = EventLoopScheduler() - schedulers.add(pose_results_scheduler) - - pose_results_writer = DynamicPoseCSVWriter(self.output_file) + pose_results_writer_scheduler = EventLoopScheduler() + schedulers.add(pose_results_writer_scheduler) pose_results_sub = poses.pipe( ops.zip(frames), ops.map(lambda p: VideoFramePoses.from_pair(p)), - observe_on_bounded(pose_results_scheduler, maxsize=30), + observe_on_bounded(pose_results_writer_scheduler, maxsize=30), finally_future(pose_results_done), - write_to(pose_results_writer) - ).subscribe() + write_to(self.pose_writer) + ).subscribe(on_error=lambda e: print(f"Error writing pose results: {e}")) subscriptions.add(pose_results_sub) frames_poses = rx.zip(color_frame_images, poses) @@ -197,53 +185,41 @@ def run(self): visualizations = visualizations.pipe(ops.share()) - if self.visualization_file is not None: + if self.video_writer is not None: assert visualizations is not None, "Visualizations are required to save a video" - self.visualization_file.parent.mkdir(parents=True, exist_ok=True) - video_writer_done = Future() drains.append(video_writer_done) video_writer_scheduler = EventLoopScheduler() schedulers.add(video_writer_scheduler) - frame_size = self.source.get_size() - assert frame_size is not None, "Frame size is not known" - fps = self.source.get_fps() - assert fps is not None, "FPS is not known" - - video_writer = FFmpegVideoFileWriter(self.visualization_file, size=frame_size, fps=fps, quality="medium", grayscale=False) - video_writer_sub = visualizations.pipe( observe_on_bounded(video_writer_scheduler, maxsize=30), finally_future(video_writer_done), - write_to(video_writer) - ).subscribe() + write_to(self.video_writer) + ).subscribe(on_error=lambda e: print(f"Error writing visualization video: {e}")) subscriptions.add(video_writer_sub) - if self.live_preview: + if self.live_preview_display is not None: live_preview_input = visualizations if visualizations is not None else color_frame_images - pacing_scheduler = EventLoopScheduler() - schedulers.add(pacing_scheduler) - display_scheduler = EventLoopScheduler() schedulers.add(display_scheduler) live_preview_done = Future() drains.append(live_preview_done) - def window_still_open(_): - # treat <= 0 (and -1 on some platforms) as closed - return cv2.getWindowProperty("Live Preview", cv2.WND_PROP_VISIBLE) > 0.5 - live_preview_sub = live_preview_input.pipe( - preview_pace(1/30, scheduler=display_scheduler), - opencv_imshow("Live Preview", scheduler=display_scheduler), - ops.take_while(window_still_open), + adaptive_pace(window_size=self.batch_size * 4, initial_interval=1/30), + ops.sample(1/30, scheduler=display_scheduler), + display_image(self.live_preview_display, scheduler=display_scheduler), + # TODO: I guess since display_image opens the display async, there is a small chance that the display is not open yet + # when the first frame arrives at ops.take_while, which would cause this path to end immediately. + # Don't know how to prevent that + ops.take_while(lambda _: self.live_preview_display.is_open()), finally_future(live_preview_done) - ).subscribe() + ).subscribe(on_error=lambda e: print(f"Error in live preview: {e}")) subscriptions.add(live_preview_sub) progress_done: Optional[Future] = None @@ -256,7 +232,7 @@ def window_still_open(_): progress_subscription = poses.pipe( ops.buffer_with_time(progress_update_interval_seconds), finally_future(progress_done) - ).subscribe(progress_bar_observer) + ).subscribe(progress_bar_observer, on_error=lambda e: print(f"Error in progress bar: {e}")) subscriptions.add(progress_subscription) try: diff --git a/src/py3r/pose/cli/render_job.py b/src/py3r/pose/cli/render_job.py index cd676f3..b616362 100644 --- a/src/py3r/pose/cli/render_job.py +++ b/src/py3r/pose/cli/render_job.py @@ -3,10 +3,8 @@ from pathlib import Path from typing import List, Optional -import cv2 from py3r.media.streaming.observables.reader_observable import reader_observable -from py3r.media.streaming.operators import observe_on_bounded, finally_future, subscribe_on_future -from py3r.media.streaming.operators.opencv_imshow import opencv_imshow +from py3r.media.streaming.operators import observe_on_bounded, finally_future from py3r.media.streaming.operators.write_to import write_to from py3r.media.video import VideoSource from py3r.media.video.ffmpeg_video_file_writer import FFmpegVideoFileWriter @@ -22,6 +20,7 @@ from reactivex.disposable import CompositeDisposable from reactivex.scheduler import EventLoopScheduler +from py3r.pose.cli.image_display import IImageDisplay, display_image from py3r.pose.cli.limit_rate import limit_rate from py3r.pose.cli.operators import ensure_3_channel from py3r.pose.cli.preview_pace import preview_pace @@ -42,7 +41,7 @@ def __init__(self): self.pose_renderer: Optional[PoseRenderer] = None self.tracker: Optional[FixedInstancesTracker] = None - self.live_preview: bool = False + self.live_preview_display: Optional[IImageDisplay] = None self.visualization_file: Optional[Path] = None self.quiet: bool = False @@ -73,8 +72,8 @@ def set_tracker(self, tracker: FixedInstancesTracker): def set_pose_renderer(self, pose_renderer: PoseRenderer): self.pose_renderer = pose_renderer - def set_live_preview(self, live_preview: bool): - self.live_preview = live_preview + def set_live_preview_display(self, display: IImageDisplay): + self.live_preview_display = display def set_visualization_file(self, visualization_file: Path): self.visualization_file = visualization_file @@ -180,28 +179,21 @@ def run(self): ).subscribe(on_error=print) subscriptions.add(video_writer_sub) - if self.live_preview: + if self.live_preview_display is not None: live_preview_input = visualizations if visualizations is not None else color_frame_images - pacing_scheduler = EventLoopScheduler() - schedulers.add(pacing_scheduler) - display_scheduler = EventLoopScheduler() schedulers.add(display_scheduler) live_preview_done = Future() drains.append(live_preview_done) - def window_still_open(_): - # treat <= 0 (and -1 on some platforms) as closed - return cv2.getWindowProperty("Live Preview", cv2.WND_PROP_VISIBLE) > 0.5 - live_preview_sub = live_preview_input.pipe( preview_pace(1/30, scheduler=display_scheduler), - opencv_imshow("Live Preview", scheduler=display_scheduler), - ops.take_while(window_still_open), + display_image(self.live_preview_display, scheduler=display_scheduler), + ops.take_while(lambda _: self.live_preview_display.is_open()), finally_future(live_preview_done) - ).subscribe(on_error=print) + ).subscribe(on_error=lambda e: print(f"Error in live preview: {e}")) subscriptions.add(live_preview_sub) progress_done: Optional[Future] = None diff --git a/src/py3r/pose/core/filtering/abc/pose_filter.py b/src/py3r/pose/core/filtering/abc/pose_filter.py index c44de05..d518eeb 100644 --- a/src/py3r/pose/core/filtering/abc/pose_filter.py +++ b/src/py3r/pose/core/filtering/abc/pose_filter.py @@ -4,5 +4,5 @@ class IPoseFilter(Protocol): - def filter(self, pose_results: List[PoseInstance]) -> List[PoseInstance]: ... - def filter_all(self, pose_results_list: List[List[PoseInstance]]) -> List[List[PoseInstance]]: ... + def filter(self, instances: List[PoseInstance], context: List[PoseInstance] = []) -> List[PoseInstance]: ... + def filter_all(self, instance_lists: List[List[PoseInstance]], context_lists: List[List[PoseInstance]] = []) -> List[List[PoseInstance]]: ... diff --git a/src/py3r/pose/core/filtering/arena_filter.py b/src/py3r/pose/core/filtering/arena_filter.py index 26357c9..c1579b3 100644 --- a/src/py3r/pose/core/filtering/arena_filter.py +++ b/src/py3r/pose/core/filtering/arena_filter.py @@ -1,17 +1,22 @@ -from typing import List, Callable +from typing import List, Optional -from py3r.pose.core.types import PoseInstanceType from py3r.pose.core.types.instance import PoseInstance class ArenaPoseFilter: - def __init__(self, arena_type_filter: Callable[[PoseInstanceType], bool], min_intersection: float = 0.1): - # Filter out instances that don't overlap with at least one arena instance - # Used to filter out instances outside the arena, which are probably false positives - # arena_type_name: Name of the arena type - # min_intersection: Minimum intersection required (as fraction of instance box area) + """ + Filters out subject instances that don't overlap with at least one arena instance. - self.arena_type_filter = arena_type_filter + arena_type: type name used to identify arena instances within the context list + min_intersection: minimum required overlap as a fraction of the subject instance's box area + + When used via InstanceScopedFilter with no explicit context selector, the full frame + is passed as context automatically, so arena instances will be present without any + extra configuration in the filter chain syntax. + """ + + def __init__(self, arena_type, min_intersection: float = 0.1): + self.arena_types: List[str] = [arena_type] if isinstance(arena_type, str) else list(arena_type) self.min_intersection = min_intersection def _bounding_box_overlap(self, instance_box, arena_box): @@ -33,19 +38,11 @@ def _bounding_box_overlap(self, instance_box, arena_box): intersection /= instance_area - # Intersection (as fraction of instance box area) is greater than min_overlap - if intersection > self.min_intersection: - return True - - return False + return intersection > self.min_intersection - def filter(self, instances: List[PoseInstance]) -> List[PoseInstance]: - arena_instances = [ - instance for instance in instances - if self.arena_type_filter(instance.type) - ] + def filter(self, instances: List[PoseInstance], context: Optional[List[PoseInstance]] = None) -> List[PoseInstance]: + arena_instances = [i for i in (context or []) if i.type.name in self.arena_types] - # Filter out instances that don't overlap with at least one arena instance filtered_instances = [] for instance in instances: for arena_instance in arena_instances: @@ -55,5 +52,7 @@ def filter(self, instances: List[PoseInstance]) -> List[PoseInstance]: return filtered_instances - def filter_all(self, instance_lists: List[List[PoseInstance]]) -> List[List[PoseInstance]]: + def filter_all(self, instance_lists: List[List[PoseInstance]], context_lists: Optional[List[List[PoseInstance]]] = None) -> List[List[PoseInstance]]: + if context_lists: + return [self.filter(instances, context) for instances, context in zip(instance_lists, context_lists)] return [self.filter(instances) for instances in instance_lists] diff --git a/src/py3r/pose/core/filtering/confidence_filter.py b/src/py3r/pose/core/filtering/confidence_filter.py index 37aa467..5e57884 100644 --- a/src/py3r/pose/core/filtering/confidence_filter.py +++ b/src/py3r/pose/core/filtering/confidence_filter.py @@ -13,12 +13,12 @@ def _filter_points(self, instance: PoseInstance) -> PoseInstance: instance = PoseInstance(instance.id, instance.type, instance.box, instance.points, instance.conf) instance.points = [ # TODO: conf is currently allowed to be None, meaning human annotated or full confidence, that's a bit weird - point if point is not None and point.conf >= self.point_confidence_threshold and not(point.x < 0.01 and point.y < 0.01) else PosePoint(0.0, 0.0, 0.0) + point if point is not None and point.conf >= self.point_confidence_threshold and not(point.x < 0.01 and point.y < 0.01) else None for point in instance.points ] return instance - def filter(self, instances: List[PoseInstance]) -> List[PoseInstance]: + def filter(self, instances: List[PoseInstance], context: List[PoseInstance] = []) -> List[PoseInstance]: instances = [self._filter_points(instance) for instance in instances] # Filter out instances with low confidence and no points instances = [ @@ -27,5 +27,5 @@ def filter(self, instances: List[PoseInstance]) -> List[PoseInstance]: ] return instances - def filter_all(self, instance_lists: List[List[PoseInstance]]) -> List[List[PoseInstance]]: + def filter_all(self, instance_lists: List[List[PoseInstance]], context_lists: List[List[PoseInstance]] = []) -> List[List[PoseInstance]]: return [self.filter(instances) for instances in instance_lists] diff --git a/src/py3r/pose/core/filtering/instance_type_filter.py b/src/py3r/pose/core/filtering/instance_type_filter.py index 252dc99..37b9945 100644 --- a/src/py3r/pose/core/filtering/instance_type_filter.py +++ b/src/py3r/pose/core/filtering/instance_type_filter.py @@ -10,11 +10,11 @@ def __init__(self, instance_types: List[Union[str, PoseInstanceType]], whitelist self.instance_types = [t if isinstance(t, str) else t.name for t in instance_types] self.whitelist = whitelist - def filter(self, instances: List[PoseInstance]) -> List[PoseInstance]: + def filter(self, instances: List[PoseInstance], context: List[PoseInstance] = []) -> List[PoseInstance]: return [ inst for inst in instances if (inst.type.name in self.instance_types) == self.whitelist ] - def filter_all(self, instance_lists: List[List[PoseInstance]]) -> List[List[PoseInstance]]: + def filter_all(self, instance_lists: List[List[PoseInstance]], context_lists: List[List[PoseInstance]] = []) -> List[List[PoseInstance]]: return [self.filter(instances) for instances in instance_lists] diff --git a/src/py3r/pose/core/filtering/median_filter.py b/src/py3r/pose/core/filtering/median_filter.py index c989cb8..de036f6 100644 --- a/src/py3r/pose/core/filtering/median_filter.py +++ b/src/py3r/pose/core/filtering/median_filter.py @@ -1,16 +1,14 @@ from copy import deepcopy -from typing import List, Callable +from typing import List import numpy as np -from py3r.pose.core.types import PoseInstanceType from py3r.pose.core.types.instance import PoseInstance from py3r.pose.core.types.point import PosePoint class MedianPoseFilter: - def __init__(self, instance_type_filter: Callable[[PoseInstanceType], bool] = None, replace_missing: bool = True): - self.instance_type_filter = instance_type_filter + def __init__(self, replace_missing: bool = True): self.replace_missing = replace_missing @staticmethod @@ -42,16 +40,14 @@ def _calculate_median_instance(instances: List[PoseInstance]) -> PoseInstance: median_instance = PoseInstance(instances[0].id, instances[0].type, median_box, median_points, median_conf) return median_instance - def filter(self, instances: List[PoseInstance]) -> List[PoseInstance]: - raise NotImplementedError + def filter(self, instances: List[PoseInstance], context: List[PoseInstance] = []) -> List[PoseInstance]: + raise NotImplementedError("MedianPoseFilter is a post-processing filter; use filter_all() instead") - def filter_all(self, instance_lists: List[List[PoseInstance]]) -> List[List[PoseInstance]]: + def filter_all(self, instance_lists: List[List[PoseInstance]], context_lists: List[List[PoseInstance]] = []) -> List[List[PoseInstance]]: unique_instances_set = set() unique_instances = [] for frame in instance_lists: for instance in frame: - if self.instance_type_filter is not None and not self.instance_type_filter(instance.type): - continue if (instance.type.name, instance.id) in unique_instances_set: continue unique_instances_set.add((instance.type.name, instance.id)) @@ -67,11 +63,11 @@ def filter_all(self, instance_lists: List[List[PoseInstance]]) -> List[List[Pose instance for frame in instance_lists for instance in frame - if instance.type.name == instance_type and instance.id == instance_id + if instance.type.name == instance_type.name and instance.id == instance_id ]) for instances, filtered_instances in zip(instance_lists, filtered_instance_lists): if self.replace_missing or any(instance.type.name == instance_type.name and instance.id == instance_id for instance in instances): filtered_instances.append(deepcopy(median_instance)) - return instance_lists + return filtered_instance_lists \ No newline at end of file diff --git a/src/py3r/pose/core/filtering/sequential_filter.py b/src/py3r/pose/core/filtering/sequential_filter.py index e2f67d7..2dd0c9c 100644 --- a/src/py3r/pose/core/filtering/sequential_filter.py +++ b/src/py3r/pose/core/filtering/sequential_filter.py @@ -8,12 +8,12 @@ class SequentialPoseFilter: def __init__(self, filters: List[IPoseFilter]): self.filters = filters - def filter(self, instances: List[PoseInstance]) -> List[PoseInstance]: - for label_filter in self.filters: - instances = label_filter.filter(instances) + def filter(self, instances: List[PoseInstance], context: List[PoseInstance] = []) -> List[PoseInstance]: + for f in self.filters: + instances = f.filter(instances) return instances - def filter_all(self, instance_lists: List[List[PoseInstance]]) -> List[List[PoseInstance]]: - for label_filter in self.filters: - instance_lists = label_filter.filter_all(instance_lists) + def filter_all(self, instance_lists: List[List[PoseInstance]], context_lists: List[List[PoseInstance]] = []) -> List[List[PoseInstance]]: + for f in self.filters: + instance_lists = f.filter_all(instance_lists) return instance_lists diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_prediction_job.py b/tests/test_prediction_job.py new file mode 100644 index 0000000..a03e99d --- /dev/null +++ b/tests/test_prediction_job.py @@ -0,0 +1,366 @@ +""" +Unit tests for PredictionJob. + +Mocks used: + - MockVideoSource – VideoSource returning a fixed number of blank frames + - CollectingPoseWriter – accumulates written VideoFramePoses for assertions + - MockImageDisplay – IImageDisplay that closes itself after N is_open() calls +""" +from __future__ import annotations + +import threading +import time +from typing import List, Optional + +import numpy as np + +from py3r.media.types import VideoFrame +from py3r.pose.cli.prediction_job import PredictionJob +from py3r.pose.core.model.mock_pose_model import MockPoseModel +from py3r.pose.core.types import PoseInstance, PoseInstanceType, PosePoint, VideoFramePoses + + +# ── Mock helpers ────────────────────────────────────────────────────────────── + +class MockVideoSource: + """VideoSource that yields a fixed number of blank frames and then EOF.""" + + def __init__( + self, + num_frames: int, + width: int = 64, + height: int = 64, + channels: int = 3, + fps: float = 30.0, + ): + self._num_frames = num_frames + self._width = width + self._height = height + self._channels = channels + self._fps = fps + self._frame_index = 0 + self._open = False + + # lifecycle + def open(self) -> None: + self._open = True + self._frame_index = 0 + + def close(self) -> None: + self._open = False + + def is_open(self) -> bool: + return self._open + + # capability probes + def has_timing(self) -> bool: return True + def has_size(self) -> bool: return True + def has_fps(self) -> bool: return True + def has_num_frames(self) -> bool: return True + def is_seekable(self) -> bool: return False + + # metadata + def get_size(self) -> tuple[int, int]: return (self._width, self._height) + def get_fps(self) -> float: return self._fps + def get_num_channels(self) -> int: return self._channels + def get_num_frames(self) -> int: return self._num_frames + def seek(self, frame_index: int) -> None: raise NotImplementedError + + def read(self, timeout: Optional[float] = None) -> Optional[VideoFrame]: + if self._frame_index >= self._num_frames: + return None + shape = ( + (self._height, self._width) + if self._channels == 1 + else (self._height, self._width, self._channels) + ) + frame = VideoFrame( + img=np.zeros(shape, dtype=np.uint8), + frame_index=self._frame_index, + timestamp=self._frame_index / self._fps, + ) + self._frame_index += 1 + return frame + + +class CollectingPoseWriter: + """Pose writer that accumulates VideoFramePoses for inspection after the job.""" + + def __init__(self): + self.frames: List[VideoFramePoses] = [] + self.opened = False + self.closed = False + + def open(self) -> None: + self.opened = True + + def write(self, frame: VideoFramePoses) -> None: + self.frames.append(frame) + + def close(self) -> None: + self.closed = True + + +class MockImageDisplay: + """ + IImageDisplay that closes itself after *close_after* calls to is_open(). + + is_open() is called once per frame that passes through take_while, so + close_after effectively limits the number of frames shown. + display() is scheduled asynchronously on the display_scheduler, so + len(self.frames) may lag slightly behind is_open_calls; use is_open_calls + for reliable counting. + """ + + def __init__(self, close_after: int = 999_999): + self._close_after = close_after + self.is_open_calls: int = 0 + self.frames: List[np.ndarray] = [] + self.torn_down = False + + def setup(self) -> None: + pass + + def display(self, img: np.ndarray) -> None: + self.frames.append(img.copy()) + + def is_open(self) -> bool: + self.is_open_calls += 1 + return self.is_open_calls <= self._close_after + + def teardown(self) -> None: + self.torn_down = True + + +class SlowMockPoseModel(MockPoseModel): + """MockPoseModel with a per-frame sleep to simulate GPU latency. + + This creates real back-pressure in the pipeline: the pose_estimation_scheduler + falls behind the frame source, filling the bounded queues and exposing any + scheduler-sharing deadlocks that a fast mock would never trigger. + """ + + def __init__(self, instances, delay_per_frame: float = 0.02): + super().__init__(instances) + self.delay_per_frame = delay_per_frame + + def predict_batch(self, batch): + time.sleep(self.delay_per_frame * len(batch)) + return super().predict_batch(batch) + + +# ── Shared helpers ──────────────────────────────────────────────────────────── + +def _instance(name: str = "mouse") -> PoseInstance: + t = PoseInstanceType(name, ["nose", "tail"]) + return PoseInstance( + id="0", + type=t, + box=(10.0, 10.0, 50.0, 50.0), + points=[PosePoint(20.0, 20.0, 0.9), PosePoint(40.0, 40.0, 0.8)], + conf=0.95, + ) + + +def _job(source: MockVideoSource, model: MockPoseModel) -> PredictionJob: + job = PredictionJob(model) + job.set_source(source) + job.set_no_progress(True) + return job + + +# ── Tests: pose writing ─────────────────────────────────────────────────────── + +class TestPoseWriting: + + def test_all_frames_are_written(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=10), MockPoseModel([])) + job.set_pose_writer(writer) + job.run() + + assert len(writer.frames) == 10 + + def test_writer_is_opened_and_closed(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=3), MockPoseModel([])) + job.set_pose_writer(writer) + job.run() + + assert writer.opened + assert writer.closed + + def test_frame_indices_are_sequential_from_zero(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=5), MockPoseModel([])) + job.set_pose_writer(writer) + job.run() + + assert [f.frame_index for f in writer.frames] == list(range(5)) + + def test_frame_size_matches_source_dimensions(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=2, width=128, height=96), MockPoseModel([])) + job.set_pose_writer(writer) + job.run() + + assert all(f.size == (128, 96) for f in writer.frames) + + def test_instances_from_model_appear_in_every_frame(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=4), MockPoseModel([_instance("mouse")])) + job.set_pose_writer(writer) + job.run() + + assert all(len(f.instances) == 1 for f in writer.frames) + assert all(f.instances[0].type.name == "mouse" for f in writer.frames) + + def test_multiple_instances_per_frame(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=3), MockPoseModel([_instance("mouse"), _instance("arena")])) + job.set_pose_writer(writer) + job.run() + + assert all(len(f.instances) == 2 for f in writer.frames) + + def test_empty_model_writes_empty_instance_lists(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=4), MockPoseModel([])) + job.set_pose_writer(writer) + job.run() + + assert all(f.instances == [] for f in writer.frames) + + +# ── Tests: frame range ──────────────────────────────────────────────────────── + +class TestFrameRange: + + def test_start_frame_skips_leading_frames(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=10), MockPoseModel([])) + job.set_pose_writer(writer) + job.set_start_frame(5) + job.run() + + assert len(writer.frames) == 5 + assert writer.frames[0].frame_index == 5 + + def test_end_frame_limits_frames_processed(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=10), MockPoseModel([])) + job.set_pose_writer(writer) + job.set_end_frame(5) + job.run() + + assert len(writer.frames) == 5 + + def test_start_and_end_frame_selects_a_slice(self): + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=10), MockPoseModel([])) + job.set_pose_writer(writer) + job.set_start_frame(3) + job.set_end_frame(7) + job.run() + + assert len(writer.frames) == 4 + assert writer.frames[0].frame_index == 3 + assert writer.frames[-1].frame_index == 6 + + +# ── Tests: live preview ─────────────────────────────────────────────────────── + +class TestLivePreview: + + def test_job_terminates_when_display_closes(self): + """ + Source has 100 frames but the display signals closed after 5 frames. + run() must return (not hang), exercising the full adaptive_pace + + sample + display_image + take_while pipeline. + """ + display = MockImageDisplay(close_after=5) + job = _job(MockVideoSource(num_frames=100), MockPoseModel([])) + job.set_live_preview_display(display) + job.run() + + assert display.torn_down + + def test_display_teardown_called_after_normal_completion(self): + display = MockImageDisplay() + job = _job(MockVideoSource(num_frames=5), MockPoseModel([])) + job.set_live_preview_display(display) + job.run() + + assert display.torn_down + + def test_display_receives_at_least_one_frame(self): + """Smoke test that display_image actually delivers frames to the display.""" + display = MockImageDisplay() + job = _job(MockVideoSource(num_frames=10), MockPoseModel([])) + job.set_live_preview_display(display) + job.run() + + assert len(display.frames) > 0 + + def test_preview_and_writer_run_concurrently(self): + """Both outputs can be active simultaneously without deadlock.""" + writer = CollectingPoseWriter() + display = MockImageDisplay() + job = _job(MockVideoSource(num_frames=10), MockPoseModel([_instance()])) + job.set_pose_writer(writer) + job.set_live_preview_display(display) + job.run() + + assert len(writer.frames) == 10 + assert len(display.frames) > 0 + + +# ── Tests: no-output mode ───────────────────────────────────────────────────── + +class TestNoOutputs: + + def test_runs_without_error_when_no_outputs_configured(self): + """With no writer and no display the job has no drains and returns + immediately after connecting — verifying it doesn't hang or raise.""" + job = _job(MockVideoSource(num_frames=5), MockPoseModel([_instance()])) + job.run() + + +# ── Tests: deadlock detection ───────────────────────────────────────────────── + +class TestDeadlockDetection: + # 100 frames × 20 ms/frame ≈ 2 s with the correct scheduler. + # A deadlock never finishes. 10 s gives a 5× safety margin. + TIMEOUT = 10.0 + + def _run_with_timeout(self, job) -> bool: + """Run job.run() in a daemon thread; return True if it finished in time.""" + thread = threading.Thread(target=job.run, daemon=True) + thread.start() + thread.join(timeout=self.TIMEOUT) + return not thread.is_alive() + + def test_completes_under_back_pressure(self): + """ + Uses SlowMockPoseModel (20 ms/frame) with 100 frames so the bounded + queues saturate — the condition that causes a deadlock when two pipeline + stages share the same EventLoopScheduler. + + This test FAILS (timeout) with the buggy scheduler assignment and + PASSES with the correct one, making the regression detectable. + """ + writer = CollectingPoseWriter() + job = _job(MockVideoSource(num_frames=200), SlowMockPoseModel([], delay_per_frame=0.02)) + job.set_pose_writer(writer) + + finished = self._run_with_timeout(job) + + assert finished, ( + f"job.run() did not complete within {self.TIMEOUT}s — " + "likely deadlock caused by two pipeline stages sharing the same scheduler" + ) + assert len(writer.frames) == 200 + + + + From cfb80278db7df3f25bf9a37ba64e26d4ee74334c Mon Sep 17 00:00:00 2001 From: Marcel Schmutz Date: Mon, 8 Jun 2026 16:23:13 +0200 Subject: [PATCH 2/5] fixup! Unit test and filter chain implementation --- .github/workflows/unit-tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml index 0252f67..7259649 100644 --- a/.github/workflows/unit-tests.yml +++ b/.github/workflows/unit-tests.yml @@ -35,5 +35,5 @@ jobs: pip install -r requirements-dev.txt - name: Run unit tests - run: python -m pytest tests/unit/ -q --tb=short + run: python -m pytest tests/ -q --tb=short From 85549ad28cbc2d4d5bfeff8fb2d4267d32c9bc96 Mon Sep 17 00:00:00 2001 From: Marcel Schmutz Date: Mon, 8 Jun 2026 16:24:46 +0200 Subject: [PATCH 3/5] fixup! Unit test and filter chain implementation --- src/py3r/pose/cli/filter_chain_builder.py | 74 ++++++ src/py3r/pose/cli/filter_chain_parser.py | 301 ++++++++++++++++++++++ src/py3r/pose/cli/filter_registry.py | 162 ++++++++++++ src/py3r/pose/cli/image_display.py | 123 +++++++++ 4 files changed, 660 insertions(+) create mode 100644 src/py3r/pose/cli/filter_chain_builder.py create mode 100644 src/py3r/pose/cli/filter_chain_parser.py create mode 100644 src/py3r/pose/cli/filter_registry.py create mode 100644 src/py3r/pose/cli/image_display.py diff --git a/src/py3r/pose/cli/filter_chain_builder.py b/src/py3r/pose/cli/filter_chain_builder.py new file mode 100644 index 0000000..6bc870f --- /dev/null +++ b/src/py3r/pose/cli/filter_chain_builder.py @@ -0,0 +1,74 @@ +import difflib +from dataclasses import dataclass +from typing import List, Optional + +from py3r.pose.core.filtering.abc.pose_filter import IPoseFilter +from py3r.pose.core.filtering.instance_scoped_filter import InstanceScopedFilter +from py3r.pose.core.filtering.sequential_filter import SequentialPoseFilter +from py3r.pose.cli.filter_chain_parser import parse_filter_chain +from py3r.pose.cli.filter_registry import FILTER_REGISTRY + + +@dataclass +class FilterStep: + """ + A fully-built filter step ready for use in a single-stream pipeline. + + filter — an InstanceScopedFilter wrapping the inner filter class. + + Note: multi-stream routing (input source streams via , output + stream naming via @name) is not yet implemented. When added, FilterStep + will gain input_streams and output_stream fields alongside a stream + graph evaluator to replace the current linear SequentialPoseFilter usage. + """ + filter: InstanceScopedFilter + + +def build_filter_chain(chain_str: str) -> List[FilterStep]: + """ + Parse and build a filter chain from a filter chain string. + + Raises FilterSyntaxError on parse errors, ValueError on unknown filter + names or invalid params. + """ + specs = parse_filter_chain(chain_str) + steps: List[FilterStep] = [] + + for spec in specs: + if spec.name not in FILTER_REGISTRY: + close = difflib.get_close_matches(spec.name, FILTER_REGISTRY.keys(), n=1) + hint = f", did you mean '{close[0]}'?" if close else "" + raise ValueError(f"Unknown filter '{spec.name}'{hint}") + + descriptor = FILTER_REGISTRY[spec.name] + resolved = descriptor.resolve_params(spec.params) + inner = descriptor.build(resolved) + + subject_types: Optional[List[str]] = None + if spec.subject_selector is not None: + names = spec.subject_selector.instance_type_names + subject_types = names if names else None + + context_types: Optional[List[str]] = ( + spec.context_selector.instance_type_names + if spec.context_selector is not None + else None + ) + + steps.append(FilterStep(InstanceScopedFilter(inner, subject_types, context_types))) + + return steps + + +def filters_from_chain(steps: List[FilterStep]) -> List[InstanceScopedFilter]: + """Return the ordered list of filters from a built chain.""" + return [step.filter for step in steps] + + +def build_sequential_filter(chain_str: str) -> IPoseFilter: + """ + Build a single composed pose filter object for the current single-stream implementation. + + The returned object can be passed directly to PredictionJob / RenderJob. + """ + return SequentialPoseFilter(filters_from_chain(build_filter_chain(chain_str))) \ No newline at end of file diff --git a/src/py3r/pose/cli/filter_chain_parser.py b/src/py3r/pose/cli/filter_chain_parser.py new file mode 100644 index 0000000..8508e1b --- /dev/null +++ b/src/py3r/pose/cli/filter_chain_parser.py @@ -0,0 +1,301 @@ +""" +Filter chain syntax parser. + +Current grammar (single-stream): + + filter_chain := filter_step (PIPE filter_step)* + filter_step := NAME subject_sel? context_sel? LPAREN param_list? RPAREN + subject_sel := LANGLE selector_list RANGLE + context_sel := LBRACKET selector_list RBRACKET + selector_list := selector (COMMA selector)* + selector := 'i:' type_name # instance type (explicit prefix) + | type_name # instance type (shorthand) + param_list := param (COMMA param)* + param := NAME EQUALS value # named param + | value # positional param + value := LBRACKET NAME (COMMA NAME)* RBRACKET # type name list + | NUMBER + | NAME + | BOOL + +Examples: + confidence(0.5, 0.3) | arena[epm_arena](0.1) + instance_type([mouse, epm_arena]) | median() + confidence(instance_threshold=0.7) + +Planned additions (multi-stream graph model, not yet implemented): + - Stream selectors inside <> and []: 's:stream_name' + e.g. arena[s:arena_cam](0.1) + - Output stream naming: '@name' suffix on a step + e.g. confidence(0.5) @filtered + Semantics: declares input source stream(s); @name declares the + output stream name (default: replaces input). Steps with no stream + qualifier read from the implicit current stream (output of the previous + step, or 'main' for the first step). +""" + +import re +from dataclasses import dataclass, field +from typing import Any, List, Optional, Tuple, Union + + +# --------------------------------------------------------------------------- +# Token types +# --------------------------------------------------------------------------- + +class _T: + NAME = "NAME" + NUMBER = "NUMBER" + BOOL = "BOOL" + LANGLE = "LANGLE" # < + RANGLE = "RANGLE" # > + LBRACKET = "LBRACKET" # [ + RBRACKET = "RBRACKET" # ] + LPAREN = "LPAREN" # ( + RPAREN = "RPAREN" # ) + COLON = "COLON" # : + COMMA = "COMMA" # , + PIPE = "PIPE" # | + EQUALS = "EQUALS" # = + EOF = "EOF" + + +@dataclass +class Token: + type: str + value: Any + pos: int + + +# --------------------------------------------------------------------------- +# Error +# --------------------------------------------------------------------------- + +class FilterSyntaxError(ValueError): + def __init__(self, message: str, pos: int, source: str = ""): + if source: + arrow = " " * pos + "^" + super().__init__(f"{message}\n {source}\n {arrow}") + else: + super().__init__(f"{message} (position {pos})") + self.pos = pos + + +# --------------------------------------------------------------------------- +# Tokeniser +# --------------------------------------------------------------------------- + +_TOKEN_PATTERNS = [ + (re.compile(r"\s+"), None), + (re.compile(r"\|"), _T.PIPE), + (re.compile(r"<"), _T.LANGLE), + (re.compile(r">"), _T.RANGLE), + (re.compile(r"\["), _T.LBRACKET), + (re.compile(r"]"), _T.RBRACKET), + (re.compile(r"\("), _T.LPAREN), + (re.compile(r"\)"), _T.RPAREN), + (re.compile(r":"), _T.COLON), + (re.compile(r","), _T.COMMA), + (re.compile(r"="), _T.EQUALS), + (re.compile(r"[+-]?\d+\.\d+"), _T.NUMBER), # float before int + (re.compile(r"[+-]?\d+"), _T.NUMBER), + (re.compile(r"[a-zA-Z_][a-zA-Z0-9_]*"), _T.NAME), +] + + +def _tokenize(text: str) -> List[Token]: + tokens: List[Token] = [] + pos = 0 + while pos < len(text): + for pattern, tok_type in _TOKEN_PATTERNS: + m = pattern.match(text, pos) + if m: + if tok_type is not None: + raw = m.group(0) + if tok_type == _T.NAME and raw.lower() in ("true", "false"): + tokens.append(Token(_T.BOOL, raw.lower() == "true", pos)) + elif tok_type == _T.NUMBER: + val: Union[int, float] = float(raw) if "." in raw else int(raw) + tokens.append(Token(_T.NUMBER, val, pos)) + else: + tokens.append(Token(tok_type, raw, pos)) + pos = m.end() + break + else: + raise FilterSyntaxError(f"Unexpected character '{text[pos]}'", pos, text) + tokens.append(Token(_T.EOF, None, pos)) + return tokens + + +# --------------------------------------------------------------------------- +# AST dataclasses +# --------------------------------------------------------------------------- + +@dataclass +class SelectorSpec: + """Parsed content of a <...> or [...] selector block.""" + instance_type_names: List[str] = field(default_factory=list) + + +# A raw param is (name_or_None, value) — name is None for positional params. +RawParam = Tuple[Optional[str], Any] + + +@dataclass +class FilterStepSpec: + name: str + subject_selector: Optional[SelectorSpec] # from <...>, or None + context_selector: Optional[SelectorSpec] # from [...], or None + params: List[RawParam] + pos: int # source position for error messages + + +# --------------------------------------------------------------------------- +# Parser +# --------------------------------------------------------------------------- + +class _Parser: + def __init__(self, tokens: List[Token], source: str): + self._tokens = tokens + self._pos = 0 + self._source = source + + # -- token helpers ------------------------------------------------------- + + def _peek(self) -> Token: + return self._tokens[self._pos] + + def _consume(self, expected_type: Optional[str] = None) -> Token: + tok = self._tokens[self._pos] + if expected_type and tok.type != expected_type: + raise FilterSyntaxError( + f"Expected {expected_type} but got '{tok.value}'", + tok.pos, self._source, + ) + self._pos += 1 + return tok + + def _match(self, *types: str) -> bool: + return self._peek().type in types + + def _peek_ahead(self, offset: int) -> Token: + idx = self._pos + offset + if idx < len(self._tokens): + return self._tokens[idx] + return self._tokens[-1] # EOF + + # -- grammar rules ------------------------------------------------------- + + def parse(self) -> List[FilterStepSpec]: + steps = [self._parse_step()] + while self._match(_T.PIPE): + self._consume(_T.PIPE) + steps.append(self._parse_step()) + self._consume(_T.EOF) + return steps + + def _parse_step(self) -> FilterStepSpec: + pos = self._peek().pos + name_tok = self._consume(_T.NAME) + + subject_sel: Optional[SelectorSpec] = None + if self._match(_T.LANGLE): + subject_sel = self._parse_selector(_T.LANGLE, _T.RANGLE) + + context_sel: Optional[SelectorSpec] = None + if self._match(_T.LBRACKET): + context_sel = self._parse_selector(_T.LBRACKET, _T.RBRACKET) + + self._consume(_T.LPAREN) + params: List[RawParam] = [] + if not self._match(_T.RPAREN): + params = self._parse_param_list() + self._consume(_T.RPAREN) + + return FilterStepSpec(name_tok.value, subject_sel, context_sel, params, pos) + + def _parse_selector(self, open_type: str, close_type: str) -> SelectorSpec: + self._consume(open_type) + spec = SelectorSpec() + if not self._match(close_type): + self._parse_selector_item(spec) + while self._match(_T.COMMA): + self._consume(_T.COMMA) + self._parse_selector_item(spec) + self._consume(close_type) + return spec + + def _parse_selector_item(self, spec: SelectorSpec) -> None: + name_tok = self._consume(_T.NAME) + if self._match(_T.COLON): + self._consume(_T.COLON) + value_tok = self._consume(_T.NAME) + if name_tok.value == "s": + raise FilterSyntaxError( + f"Stream qualifiers (s:{value_tok.value}) are not yet supported. " + "Planned for future multi-stream support.", + name_tok.pos, self._source, + ) + elif name_tok.value == "i": + spec.instance_type_names.append(value_tok.value) + else: + raise FilterSyntaxError( + f"Unknown selector prefix '{name_tok.value}': use 'i:' for instance types", + name_tok.pos, self._source, + ) + else: + # bare name = instance type shorthand + spec.instance_type_names.append(name_tok.value) + + def _parse_param_list(self) -> List[RawParam]: + params = [self._parse_param()] + while self._match(_T.COMMA): + self._consume(_T.COMMA) + params.append(self._parse_param()) + return params + + def _parse_param(self) -> RawParam: + # named param: NAME EQUALS value + if self._match(_T.NAME) and self._peek_ahead(1).type == _T.EQUALS: + name_tok = self._consume(_T.NAME) + self._consume(_T.EQUALS) + return (name_tok.value, self._parse_value()) + return (None, self._parse_value()) + + def _parse_value(self) -> Any: + if self._match(_T.LBRACKET): + # list of type names: [mouse, epm_arena] + self._consume(_T.LBRACKET) + items: List[str] = [] + if not self._match(_T.RBRACKET): + items.append(self._consume(_T.NAME).value) + while self._match(_T.COMMA): + self._consume(_T.COMMA) + items.append(self._consume(_T.NAME).value) + self._consume(_T.RBRACKET) + return items + if self._match(_T.NUMBER): + return self._consume(_T.NUMBER).value + if self._match(_T.BOOL): + return self._consume(_T.BOOL).value + if self._match(_T.NAME): + return self._consume(_T.NAME).value + tok = self._peek() + raise FilterSyntaxError( + f"Expected a value (number, name, true/false, or [list]) but got '{tok.value}'", + tok.pos, self._source, + ) + + +# --------------------------------------------------------------------------- +# Public entry point +# --------------------------------------------------------------------------- + +def parse_filter_chain(text: str) -> List[FilterStepSpec]: + """ + Parse a filter chain string into a list of FilterStepSpec objects. + Raises FilterSyntaxError with position information on invalid input. + """ + tokens = _tokenize(text) + return _Parser(tokens, text).parse() + diff --git a/src/py3r/pose/cli/filter_registry.py b/src/py3r/pose/cli/filter_registry.py new file mode 100644 index 0000000..35ffeb4 --- /dev/null +++ b/src/py3r/pose/cli/filter_registry.py @@ -0,0 +1,162 @@ +from dataclasses import dataclass +from typing import Any, Dict, List + +from py3r.pose.core.filtering.abc.pose_filter import IPoseFilter + + +@dataclass +class FilterParamSpec: + name: str + type: type # str | float | int | bool | list + default: Any = None + required: bool = False + + +class FilterDescriptor: + """ + Describes a named filter type: its accepted parameters and how to instantiate + the inner filter class. Registered in FILTER_REGISTRY by filter_name. + """ + + def __init__(self, filter_name: str, params: List[FilterParamSpec]): + self.filter_name = filter_name + self.params = params + + def resolve_params(self, raw_params: list) -> dict: + """ + Resolve a list of (name_or_None, value) tuples into a fully-named dict, + applying positional mapping and defaults. Raises ValueError on unknown, + duplicate, missing-required, or wrongly-typed params. + """ + resolved: Dict[str, Any] = {} + positional_idx = 0 + + for param_name, value in raw_params: + if param_name is not None: + spec = next((p for p in self.params if p.name == param_name), None) + if spec is None: + known = ", ".join(p.name for p in self.params) + raise ValueError( + f"Filter '{self.filter_name}': unknown param '{param_name}'. " + f"Known params: {known}" + ) + if param_name in resolved: + raise ValueError(f"Filter '{self.filter_name}': duplicate param '{param_name}'") + resolved[param_name] = self._coerce(value, spec) + else: + # advance past already-filled named params + while positional_idx < len(self.params) and self.params[positional_idx].name in resolved: + positional_idx += 1 + if positional_idx >= len(self.params): + raise ValueError(f"Filter '{self.filter_name}': too many positional params") + spec = self.params[positional_idx] + resolved[spec.name] = self._coerce(value, spec) + positional_idx += 1 + + # apply defaults / check required + for spec in self.params: + if spec.name not in resolved: + if spec.required: + raise ValueError(f"Filter '{self.filter_name}': required param '{spec.name}' is missing") + resolved[spec.name] = spec.default + + return resolved + + def _coerce(self, value: Any, spec: FilterParamSpec) -> Any: + if spec.type is list: + if isinstance(value, str): + return [value] + if not isinstance(value, list): + raise ValueError(f"Filter '{self.filter_name}': param '{spec.name}' must be a list or a single string") + return value + if spec.type is bool: + if not isinstance(value, bool): + raise ValueError(f"Filter '{self.filter_name}': param '{spec.name}' must be true or false") + return value + if spec.type is float: + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) + raise ValueError(f"Filter '{self.filter_name}': param '{spec.name}' must be a number") + if spec.type is int: + if isinstance(value, int) and not isinstance(value, bool): + return value + raise ValueError(f"Filter '{self.filter_name}': param '{spec.name}' must be an integer") + if spec.type is str: + if not isinstance(value, str): + raise ValueError(f"Filter '{self.filter_name}': param '{spec.name}' must be a string") + return value + return value + + def build(self, resolved_params: dict) -> IPoseFilter: + raise NotImplementedError + + +# --------------------------------------------------------------------------- +# Registry +# --------------------------------------------------------------------------- + +FILTER_REGISTRY: Dict[str, FilterDescriptor] = {} + + +def register_filter(descriptor: FilterDescriptor) -> None: + FILTER_REGISTRY[descriptor.filter_name] = descriptor + + +# --------------------------------------------------------------------------- +# Built-in descriptors +# --------------------------------------------------------------------------- + +class _ConfidenceDescriptor(FilterDescriptor): + def __init__(self): + super().__init__("confidence", [ + FilterParamSpec("instance_threshold", float, default=0.5), + FilterParamSpec("point_threshold", float, default=0.5), + ]) + + def build(self, p: dict) -> IPoseFilter: + from py3r.pose.core.filtering.confidence_filter import ConfidencePoseFilter + return ConfidencePoseFilter( + instance_confidence_threshold=p["instance_threshold"], + point_confidence_threshold=p["point_threshold"], + ) + + +class _ArenaDescriptor(FilterDescriptor): + def __init__(self): + super().__init__("arena", [ + FilterParamSpec("arena_type", list, required=True), + FilterParamSpec("min_intersection", float, default=0.1), + ]) + + def build(self, p: dict) -> IPoseFilter: + from py3r.pose.core.filtering.arena_filter import ArenaPoseFilter + return ArenaPoseFilter(arena_type=p["arena_type"], min_intersection=p["min_intersection"]) + + +class _InstanceTypeDescriptor(FilterDescriptor): + def __init__(self): + super().__init__("instance_type", [ + FilterParamSpec("types", list, required=True), + FilterParamSpec("whitelist", bool, default=True), + ]) + + def build(self, p: dict) -> IPoseFilter: + from py3r.pose.core.filtering.instance_type_filter import InstanceTypePoseFilter + return InstanceTypePoseFilter(instance_types=p["types"], whitelist=p["whitelist"]) + + +class _MedianDescriptor(FilterDescriptor): + def __init__(self): + super().__init__("median", [ + FilterParamSpec("replace_missing", bool, default=True), + ]) + + def build(self, p: dict) -> IPoseFilter: + from py3r.pose.core.filtering.median_filter import MedianPoseFilter + return MedianPoseFilter(replace_missing=p["replace_missing"]) + + +register_filter(_ConfidenceDescriptor()) +register_filter(_ArenaDescriptor()) +register_filter(_InstanceTypeDescriptor()) +register_filter(_MedianDescriptor()) diff --git a/src/py3r/pose/cli/image_display.py b/src/py3r/pose/cli/image_display.py new file mode 100644 index 0000000..597bfdd --- /dev/null +++ b/src/py3r/pose/cli/image_display.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +from typing import Callable, Protocol + +import cv2 +import numpy as np +import reactivex as rx +from reactivex import Observable +from reactivex.abc import SchedulerBase, ObserverBase +from reactivex.disposable import Disposable, SerialDisposable + + +class IImageDisplay(Protocol): + """Protocol for image display backends used by the display_image operator.""" + + def setup(self) -> None: + """Called once when the display is first set up (e.g. create a window).""" + ... + + def display(self, img: np.ndarray) -> None: + """Called for each frame to display it.""" + ... + + def is_open(self) -> bool: + """Return True while the display is still active (e.g. window not closed).""" + ... + + def teardown(self) -> None: + """Called once when the display is torn down (e.g. destroy the window).""" + ... + + +class OpenCVImageDisplay: + """cv2-backed image display for production use.""" + + def __init__(self, window_name: str, flags: int = cv2.WINDOW_AUTOSIZE, wait_ms: int = 1): + self.window_name = window_name + self.flags = flags + self.wait_ms = wait_ms + + def setup(self) -> None: + cv2.namedWindow(self.window_name, self.flags) + + def display(self, img: np.ndarray) -> None: + cv2.imshow(self.window_name, img) + cv2.waitKey(self.wait_ms) + + def is_open(self) -> bool: + # treat <= 0 (and -1 on some platforms) as closed + return cv2.getWindowProperty(self.window_name, cv2.WND_PROP_VISIBLE) > 0.5 + + def teardown(self) -> None: + try: + cv2.destroyWindow(self.window_name) + except Exception: + pass + + +def display_image( + display: IImageDisplay, + *, + scheduler: SchedulerBase, +) -> Callable[[Observable[np.ndarray]], Observable[np.ndarray]]: + """ + Rx operator that passes frames through while displaying each one as a side effect. + + setup/display/teardown are all dispatched onto *scheduler*, so the display object + is only ever touched from that one thread — no locking required. + + For tests, pass CurrentThreadScheduler() so that display calls execute + synchronously on the calling thread without spawning a background thread. + """ + + def _operator(source: Observable[np.ndarray]) -> Observable[np.ndarray]: + def _subscribe( + observer: ObserverBase[np.ndarray], + _subscribe_scheduler: SchedulerBase | None = None, + ): + disposed = False + upstream = SerialDisposable() + + def do_setup(*_): + if not disposed: + display.setup() + + scheduler.schedule(do_setup) + + def on_next(img: np.ndarray) -> None: + def do_display(*_): + if not disposed: + display.display(img) + + scheduler.schedule(do_display) + observer.on_next(img) + + def on_error(err: Exception) -> None: + observer.on_error(err) + + def on_completed() -> None: + observer.on_completed() + + upstream.disposable = source.subscribe(on_next, on_error, on_completed) + + def dispose() -> None: + nonlocal disposed + if disposed: + return + disposed = True + upstream.dispose() + + def do_teardown(*_): + try: + display.teardown() + except Exception: + pass + + scheduler.schedule(do_teardown) + + return Disposable(dispose) + + return rx.create(_subscribe) + + return _operator From d3e073cf4221ab60e8a0c1f6b9dcc4acc51cd040 Mon Sep 17 00:00:00 2001 From: Marcel Schmutz Date: Mon, 8 Jun 2026 16:28:20 +0200 Subject: [PATCH 4/5] fixup! Unit test and filter chain implementation --- src/py3r/pose/core/types/__init__.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/src/py3r/pose/core/types/__init__.py b/src/py3r/pose/core/types/__init__.py index 4c3417f..c34339a 100644 --- a/src/py3r/pose/core/types/__init__.py +++ b/src/py3r/pose/core/types/__init__.py @@ -1,5 +1,5 @@ from dataclasses import dataclass, field, replace -from typing import List, Tuple, runtime_checkable, Protocol, Self +from typing import List, Tuple, runtime_checkable, Protocol from py3r.media.types import HasImageMeta, HasFrameMeta @@ -20,23 +20,23 @@ class ImagePoses(HasImageMeta): size: Tuple[int, int] = field(default_factory=lambda: (0, 0)) @classmethod - def from_parts(cls, poses: List[PoseInstance], meta: HasImageMeta) -> Self: + def from_parts(cls, poses: List[PoseInstance], meta: HasImageMeta) -> "ImagePoses": return cls( instances=poses, size=meta.size, ) @classmethod - def from_pair(cls, pair: Tuple[List[PoseInstance], HasImageMeta]) -> Self: + def from_pair(cls, pair: Tuple[List[PoseInstance], HasImageMeta]) -> "ImagePoses": return cls.from_parts(*pair) - def with_meta(self, meta: HasImageMeta) -> Self: + def with_meta(self, meta: HasImageMeta) -> "ImagePoses": return replace( self, size=meta.size, ) - def with_poses(self, poses: List[PoseInstance]) -> Self: + def with_poses(self, poses: List[PoseInstance]) -> "ImagePoses": return replace( self, instances=poses, @@ -50,7 +50,7 @@ class VideoFramePoses(HasFrameMeta): timestamp: float = 0.0 @classmethod - def from_parts(cls, poses: List[PoseInstance], meta: HasFrameMeta) -> Self: + def from_parts(cls, poses: List[PoseInstance], meta: HasFrameMeta) -> "VideoFramePoses": return cls( instances=poses, size=meta.size, @@ -59,10 +59,10 @@ def from_parts(cls, poses: List[PoseInstance], meta: HasFrameMeta) -> Self: ) @classmethod - def from_pair(cls, pair: Tuple[List[PoseInstance], HasFrameMeta]) -> Self: + def from_pair(cls, pair: Tuple[List[PoseInstance], HasFrameMeta]) -> "VideoFramePoses": return cls.from_parts(*pair) - def with_meta(self, meta: HasFrameMeta) -> Self: + def with_meta(self, meta: HasFrameMeta) -> "VideoFramePoses": return replace( self, size=meta.size, @@ -70,7 +70,7 @@ def with_meta(self, meta: HasFrameMeta) -> Self: timestamp=meta.timestamp, ) - def with_poses(self, poses: List[PoseInstance]) -> Self: + def with_poses(self, poses: List[PoseInstance]) -> "VideoFramePoses": return replace( self, instances=poses, From 58dfe2b25fb5e5f75fde554133ff33f5dd086216 Mon Sep 17 00:00:00 2001 From: Marcel Schmutz Date: Mon, 8 Jun 2026 16:31:01 +0200 Subject: [PATCH 5/5] fixup! Unit test and filter chain implementation --- .github/workflows/unit-tests.yml | 2 +- pyproject.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml index 7259649..99543be 100644 --- a/.github/workflows/unit-tests.yml +++ b/.github/workflows/unit-tests.yml @@ -13,7 +13,7 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest, windows-latest] - python-version: ["3.9", "3.10", "3.11", "3.12"] + python-version: ["3.10", "3.11", "3.12"] steps: - uses: actions/checkout@v4 diff --git a/pyproject.toml b/pyproject.toml index c0f91a9..aa38c41 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ authors = [ ] description = "Library for pose estimation and pose analysis in the ETH Zurich 3R Hub" readme = "README.md" -requires-python = ">=3.9" +requires-python = ">=3.10" license = "MIT" license-files = ["LICEN[CS]E*"] dependencies = [