diff --git a/openadapt_capture/capture.py b/openadapt_capture/capture.py index 5bfcf51..30bf146 100644 --- a/openadapt_capture/capture.py +++ b/openadapt_capture/capture.py @@ -45,6 +45,10 @@ from openadapt_capture.browser_events import BrowserEvent +class InvalidCaptureEvent(ValueError): + """A stored input event lacks the data required for deterministic replay.""" + + def _parse_structural_observation(raw: object) -> StructuralObservation | None: """Load optional structural evidence without breaking legacy recordings.""" @@ -56,14 +60,52 @@ def _parse_structural_observation(raw: object) -> StructuralObservation | None: return None -def _convert_action_event(db_event) -> PydanticActionEvent | None: +def _required_event_value(db_event, field: str): + value = getattr(db_event, field, None) + if value is None: + raise InvalidCaptureEvent( + f"stored {getattr(db_event, 'name', 'unknown')!r} event is missing {field!r}" + ) + return value + + +def _required_nonempty_string(db_event, field: str) -> str: + value = _required_event_value(db_event, field) + if not isinstance(value, str) or not value.strip(): + raise InvalidCaptureEvent( + f"stored {getattr(db_event, 'name', 'unknown')!r} event has invalid {field!r}" + ) + return value + + +def _require_key_identity(db_event) -> None: + fields = ( + "key_name", + "key_char", + "key_vk", + "canonical_key_name", + "canonical_key_char", + "canonical_key_vk", + ) + if not any(getattr(db_event, field, None) not in (None, "") for field in fields): + raise InvalidCaptureEvent( + f"stored {getattr(db_event, 'name', 'unknown')!r} event has no key identity" + ) + + +def _convert_action_event(db_event) -> PydanticActionEvent: """Convert a SQLAlchemy ActionEvent to a Pydantic event. Args: db_event: SQLAlchemy ActionEvent instance. Returns: - Pydantic event or None if unrecognized. + A validated Pydantic event. + + Raises: + InvalidCaptureEvent: If the stored event is unknown or lacks data that + deterministic replay requires. Missing coordinates are never + replaced with zero because zero is a valid screen position. """ common = { "timestamp": db_event.timestamp, @@ -75,37 +117,40 @@ def _convert_action_event(db_event) -> PydanticActionEvent | None: if db_event.name == "move": return MouseMoveEvent( **common, - x=db_event.mouse_x or 0, - y=db_event.mouse_y or 0, + x=_required_event_value(db_event, "mouse_x"), + y=_required_event_value(db_event, "mouse_y"), ) elif db_event.name == "click": - button = db_event.mouse_button_name or "left" + button = _required_nonempty_string(db_event, "mouse_button_name") if db_event.mouse_pressed is True: return MouseDownEvent( **common, - x=db_event.mouse_x or 0, - y=db_event.mouse_y or 0, + x=_required_event_value(db_event, "mouse_x"), + y=_required_event_value(db_event, "mouse_y"), button=button, ) elif db_event.mouse_pressed is False: return MouseUpEvent( **common, - x=db_event.mouse_x or 0, - y=db_event.mouse_y or 0, + x=_required_event_value(db_event, "mouse_x"), + y=_required_event_value(db_event, "mouse_y"), button=button, ) else: - return None + raise InvalidCaptureEvent( + "stored 'click' event has no pressed/released state" + ) elif db_event.name == "scroll": return MouseScrollEvent( **common, - x=db_event.mouse_x or 0, - y=db_event.mouse_y or 0, - dx=db_event.mouse_dx or 0, - dy=db_event.mouse_dy or 0, + x=_required_event_value(db_event, "mouse_x"), + y=_required_event_value(db_event, "mouse_y"), + dx=_required_event_value(db_event, "mouse_dx"), + dy=_required_event_value(db_event, "mouse_dy"), ) elif db_event.name == "press": + _require_key_identity(db_event) return KeyDownEvent( **common, key_name=db_event.key_name, @@ -116,6 +161,7 @@ def _convert_action_event(db_event) -> PydanticActionEvent | None: canonical_key_vk=db_event.canonical_key_vk, ) elif db_event.name == "release": + _require_key_identity(db_event) return KeyUpEvent( **common, key_name=db_event.key_name, @@ -125,7 +171,7 @@ def _convert_action_event(db_event) -> PydanticActionEvent | None: canonical_key_char=db_event.canonical_key_char, canonical_key_vk=db_event.canonical_key_vk, ) - return None + raise InvalidCaptureEvent(f"unknown stored action event name {db_event.name!r}") def _parse_element_ref(raw: dict | None) -> SemanticElementRef | None: @@ -360,7 +406,14 @@ def keys(self) -> list[str] | None: for child in self.event.children: if isinstance(child, KeyDownEvent): # Get key identifier - key_id = child.key_name or child.key_char or child.key_vk + key_id = ( + child.canonical_key_name + or child.key_name + or child.canonical_key_char + or child.key_char + or child.canonical_key_vk + or child.key_vk + ) if key_id and key_id not in seen: seen.add(key_id) key_names.append(key_id) @@ -540,17 +593,42 @@ def audio_path(self) -> Path | None: def pixel_ratio(self) -> float: """Display pixel ratio (physical/logical), e.g. 2.0 for Retina. - Defaults to 1.0 if not stored in the recording. + Raises ``DisplayMetricsUnavailable`` when no measurement was retained. """ + from math import isfinite + + from openadapt_capture.platform import DisplayMetricsUnavailable + + def validate(value: object, source: str) -> float: + if isinstance(value, bool): + raise DisplayMetricsUnavailable( + f"The capture has an invalid {source} display pixel ratio." + ) + try: + parsed = float(value) + except (TypeError, ValueError) as exc: + raise DisplayMetricsUnavailable( + f"The capture has an invalid {source} display pixel ratio." + ) from exc + if not isfinite(parsed) or parsed <= 0: + raise DisplayMetricsUnavailable( + f"The capture has an invalid {source} display pixel ratio." + ) + return parsed + # Check if the Recording model has a pixel_ratio column ratio = getattr(self._recording, "pixel_ratio", None) if ratio is not None: - return float(ratio) + return validate(ratio, "stored") # Check the config JSON for pixel_ratio config = getattr(self._recording, "config", None) if isinstance(config, dict) and "pixel_ratio" in config: - return float(config["pixel_ratio"]) - return 1.0 + return validate(config["pixel_ratio"], "legacy") + + raise DisplayMetricsUnavailable( + "The capture has no measured display pixel ratio. Supply an exact " + "pixel_ratio in the legacy capture config or record the workflow again." + ) @property def window_capture(self) -> dict | None: @@ -593,9 +671,7 @@ def raw_events(self) -> list[PydanticActionEvent]: for db_event in self._recording.action_events: if getattr(db_event, "disabled", False): continue - pydantic_event = _convert_action_event(db_event) - if pydantic_event is not None: - events.append(pydantic_event) + events.append(_convert_action_event(db_event)) return events def actions(self, include_moves: bool = False) -> Iterator[Action]: diff --git a/openadapt_capture/processing.py b/openadapt_capture/processing.py index 14ba9b1..49daba5 100644 --- a/openadapt_capture/processing.py +++ b/openadapt_capture/processing.py @@ -148,7 +148,16 @@ def remove_invalid_keyboard_events(events: list[ActionEvent]) -> list[ActionEven for event in events: if isinstance(event, (KeyDownEvent, KeyUpEvent)): # Filter out events with no key information - if not any([event.key_name, event.key_char, event.key_vk]): + if not any( + [ + event.key_name, + event.key_char, + event.key_vk, + event.canonical_key_name, + event.canonical_key_char, + event.canonical_key_vk, + ] + ): continue valid_events.append(event) return valid_events @@ -245,9 +254,10 @@ def flush_buffer() -> None: ) else: text = "".join( - event.key_char + event.key_char or event.canonical_key_char for event in down_events - if event.key_char and _modifier_for(event) is None + if (event.key_char or event.canonical_key_char) + and _modifier_for(event) is None ) type_event = KeyTypeEvent( timestamp=first_event.timestamp, @@ -264,11 +274,11 @@ def flush_buffer() -> None: for event in events: if isinstance(event, KeyDownEvent): - key_id = event.key_name or event.key_char or event.key_vk or "" + key_id = _keyboard_key(event) pressed_keys.add(key_id) keyboard_buffer.append(event) elif isinstance(event, KeyUpEvent): - key_id = event.key_name or event.key_char or event.key_vk or "" + key_id = _keyboard_key(event) pressed_keys.discard(key_id) keyboard_buffer.append(event) diff --git a/openadapt_capture/storage.py b/openadapt_capture/storage.py index c7e1524..44efa88 100644 --- a/openadapt_capture/storage.py +++ b/openadapt_capture/storage.py @@ -61,8 +61,8 @@ class Capture(BaseModel): platform: str = Field(description="Platform identifier: 'darwin' | 'win32' | 'linux'") screen_width: int = Field(description="Screen width in physical pixels") screen_height: int = Field(description="Screen height in physical pixels") - pixel_ratio: float = Field( - default=1.0, + pixel_ratio: float | None = Field( + default=None, description="Display pixel ratio (physical/logical), e.g., 2.0 for Retina" ) task_description: str | None = Field(default=None, description="User-provided task description") @@ -116,7 +116,7 @@ class Capture(BaseModel): platform TEXT NOT NULL, screen_width INTEGER NOT NULL, screen_height INTEGER NOT NULL, - pixel_ratio REAL DEFAULT 1.0, + pixel_ratio REAL, task_description TEXT, double_click_interval_seconds REAL, double_click_distance_pixels REAL, @@ -301,7 +301,7 @@ def get_capture(self) -> Capture | None: platform=row["platform"], screen_width=row["screen_width"], screen_height=row["screen_height"], - pixel_ratio=row["pixel_ratio"] if "pixel_ratio" in row.keys() else 1.0, + pixel_ratio=row["pixel_ratio"] if "pixel_ratio" in row.keys() else None, task_description=row["task_description"], double_click_interval_seconds=row["double_click_interval_seconds"], double_click_distance_pixels=row["double_click_distance_pixels"], @@ -537,13 +537,10 @@ def _detect_platform() -> str: def _detect_screen_size() -> tuple[int, int]: - """Detect screen dimensions.""" - try: - from PIL import ImageGrab - screenshot = ImageGrab.grab() - return screenshot.size - except Exception: - return (1920, 1080) # Fallback default + """Measure screen dimensions without substituting plausible geometry.""" + from openadapt_capture import platform as capture_platform + + return capture_platform.get_screen_dimensions() def create_capture( diff --git a/openadapt_capture/storage/sqlite.py b/openadapt_capture/storage/sqlite.py index 67a5c34..302b3dd 100644 --- a/openadapt_capture/storage/sqlite.py +++ b/openadapt_capture/storage/sqlite.py @@ -111,7 +111,7 @@ class SQLiteStorage: platform TEXT NOT NULL, screen_width INTEGER NOT NULL, screen_height INTEGER NOT NULL, - pixel_ratio REAL DEFAULT 1.0, + pixel_ratio REAL, task_description TEXT, double_click_interval_seconds REAL, double_click_distance_pixels REAL, @@ -261,7 +261,7 @@ def load_capture(self) -> "Capture | None": platform=row["platform"], screen_width=row["screen_width"], screen_height=row["screen_height"], - pixel_ratio=row["pixel_ratio"] if "pixel_ratio" in row.keys() else 1.0, + pixel_ratio=row["pixel_ratio"] if "pixel_ratio" in row.keys() else None, task_description=row["task_description"], double_click_interval_seconds=row["double_click_interval_seconds"], double_click_distance_pixels=row["double_click_distance_pixels"], diff --git a/openadapt_capture/storage_impl.py b/openadapt_capture/storage_impl.py index c7e1524..44efa88 100644 --- a/openadapt_capture/storage_impl.py +++ b/openadapt_capture/storage_impl.py @@ -61,8 +61,8 @@ class Capture(BaseModel): platform: str = Field(description="Platform identifier: 'darwin' | 'win32' | 'linux'") screen_width: int = Field(description="Screen width in physical pixels") screen_height: int = Field(description="Screen height in physical pixels") - pixel_ratio: float = Field( - default=1.0, + pixel_ratio: float | None = Field( + default=None, description="Display pixel ratio (physical/logical), e.g., 2.0 for Retina" ) task_description: str | None = Field(default=None, description="User-provided task description") @@ -116,7 +116,7 @@ class Capture(BaseModel): platform TEXT NOT NULL, screen_width INTEGER NOT NULL, screen_height INTEGER NOT NULL, - pixel_ratio REAL DEFAULT 1.0, + pixel_ratio REAL, task_description TEXT, double_click_interval_seconds REAL, double_click_distance_pixels REAL, @@ -301,7 +301,7 @@ def get_capture(self) -> Capture | None: platform=row["platform"], screen_width=row["screen_width"], screen_height=row["screen_height"], - pixel_ratio=row["pixel_ratio"] if "pixel_ratio" in row.keys() else 1.0, + pixel_ratio=row["pixel_ratio"] if "pixel_ratio" in row.keys() else None, task_description=row["task_description"], double_click_interval_seconds=row["double_click_interval_seconds"], double_click_distance_pixels=row["double_click_distance_pixels"], @@ -537,13 +537,10 @@ def _detect_platform() -> str: def _detect_screen_size() -> tuple[int, int]: - """Detect screen dimensions.""" - try: - from PIL import ImageGrab - screenshot = ImageGrab.grab() - return screenshot.size - except Exception: - return (1920, 1080) # Fallback default + """Measure screen dimensions without substituting plausible geometry.""" + from openadapt_capture import platform as capture_platform + + return capture_platform.get_screen_dimensions() def create_capture( diff --git a/tests/test_highlevel.py b/tests/test_highlevel.py index 25322a8..20f7ee5 100644 --- a/tests/test_highlevel.py +++ b/tests/test_highlevel.py @@ -13,8 +13,9 @@ from openadapt_capture import recorder as recorder_module from openadapt_capture import video -from openadapt_capture.capture import Capture +from openadapt_capture.capture import Capture, InvalidCaptureEvent from openadapt_capture.db import create_db, crud +from openadapt_capture.platform import DisplayMetricsUnavailable from openadapt_capture.recorder import Recorder # Sessions/engines created by _create_test_recording, released by the @@ -452,8 +453,8 @@ def test_session_leak_on_no_recording(self, temp_capture_dir): with pytest.raises(FileNotFoundError, match="no recording found"): Capture.load(capture_path) - def test_mouse_pressed_none_skipped(self, temp_capture_dir): - """Test that click events with mouse_pressed=None are skipped.""" + def test_mouse_pressed_none_is_refused(self, temp_capture_dir): + """A corrupt click cannot disappear from the replay event stream.""" capture_path = str(Path(temp_capture_dir) / "capture") recording, db_path, session = _create_test_recording(capture_path) @@ -474,10 +475,8 @@ def test_mouse_pressed_none_skipped(self, temp_capture_dir): }) capture = Capture.load(capture_path) - events = capture.raw_events() - # The click with mouse_pressed=None should be skipped - assert len(events) == 1 - assert events[0].type == "mouse.move" + with pytest.raises(InvalidCaptureEvent, match="pressed/released"): + capture.raw_events() capture.close() def test_disabled_events_filtered(self, temp_capture_dir): @@ -659,7 +658,7 @@ def test_old_recording_without_column_loads(self, temp_capture_dir): assert capture.pixel_ratio == 1.5 # recovered from config JSON capture.close() - # No column and no config -> genuinely unknown -> 1.0. + # No column and no config -> genuinely unknown -> refuse. capture_path2 = str(Path(temp_capture_dir) / "old_no_config") os.makedirs(capture_path2, exist_ok=True) db_path2 = os.path.join(capture_path2, "recording.db") @@ -686,5 +685,6 @@ def test_old_recording_without_column_loads(self, temp_capture_dir): con.close() capture2 = Capture.load(capture_path2) - assert capture2.pixel_ratio == 1.0 + with pytest.raises(DisplayMetricsUnavailable): + _ = capture2.pixel_ratio capture2.close() diff --git a/tests/test_no_fabricated_capture.py b/tests/test_no_fabricated_capture.py index 9de0171..58d4cfe 100644 --- a/tests/test_no_fabricated_capture.py +++ b/tests/test_no_fabricated_capture.py @@ -17,6 +17,7 @@ from openadapt_capture import platform as capture_platform from openadapt_capture import recorder, window +from openadapt_capture.capture import CaptureSession, InvalidCaptureEvent, _convert_action_event from openadapt_capture.platform import DisplayMetricsUnavailable from openadapt_capture.structural import StructuralObservationRequest from openadapt_capture.structural_observer.windows import WindowsUIAStructuralObserver @@ -90,6 +91,138 @@ def _unmeasurable() -> float: # representable in the recording itself. assert recording.pixel_ratio is None + def test_session_refuses_an_unknown_pixel_ratio(self) -> None: + session = CaptureSession( + "/tmp/unused", + None, + SimpleNamespace(pixel_ratio=None, config={}), + ) + + with pytest.raises(DisplayMetricsUnavailable, match="no measured"): + _ = session.pixel_ratio + + @pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf"), True, "bad"]) + def test_session_refuses_an_invalid_pixel_ratio(self, value: object) -> None: + session = CaptureSession( + "/tmp/unused", + None, + SimpleNamespace(pixel_ratio=value, config={}), + ) + + with pytest.raises(DisplayMetricsUnavailable, match="invalid stored"): + _ = session.pixel_ratio + + def test_session_refuses_an_invalid_legacy_pixel_ratio(self) -> None: + session = CaptureSession( + "/tmp/unused", + None, + SimpleNamespace(pixel_ratio=None, config={"pixel_ratio": None}), + ) + + with pytest.raises(DisplayMetricsUnavailable, match="invalid legacy"): + _ = session.pixel_ratio + + def test_storage_create_capture_refuses_an_unmeasured_display( + self, monkeypatch, tmp_path + ) -> None: + from openadapt_capture.storage import create_capture + + def unavailable() -> tuple[int, int]: + raise DisplayMetricsUnavailable("display probe failed") + + monkeypatch.setattr(capture_platform, "get_screen_dimensions", unavailable) + + with pytest.raises(DisplayMetricsUnavailable, match="probe failed"): + create_capture(tmp_path / "capture") + + def test_storage_create_capture_accepts_explicit_display_geometry( + self, monkeypatch, tmp_path + ) -> None: + from openadapt_capture.storage import create_capture + + def unavailable() -> tuple[int, int]: + raise DisplayMetricsUnavailable("must not run") + + monkeypatch.setattr(capture_platform, "get_screen_dimensions", unavailable) + capture, storage = create_capture( + tmp_path / "capture", + screen_width=1280, + screen_height=720, + ) + try: + assert (capture.screen_width, capture.screen_height) == (1280, 720) + finally: + storage.close() + + +class TestStoredInputDataIsNotInvented: + @pytest.mark.parametrize( + ("field", "value"), + [ + ("mouse_x", None), + ("mouse_y", None), + ("mouse_button_name", None), + ("mouse_button_name", ""), + ("mouse_pressed", None), + ], + ) + def test_incomplete_click_is_refused(self, field: str, value: object) -> None: + event = SimpleNamespace( + name="click", + timestamp=1.0, + structural_observation=None, + mouse_x=10, + mouse_y=20, + mouse_button_name="left", + mouse_pressed=True, + ) + setattr(event, field, value) + + with pytest.raises(InvalidCaptureEvent): + _convert_action_event(event) + + def test_unknown_action_event_is_refused(self) -> None: + event = SimpleNamespace( + name="future-action", + timestamp=1.0, + structural_observation=None, + ) + + with pytest.raises(InvalidCaptureEvent, match="unknown stored action"): + _convert_action_event(event) + + def test_key_event_without_an_identity_is_refused(self) -> None: + event = SimpleNamespace( + name="press", + timestamp=1.0, + structural_observation=None, + key_name=None, + key_char=None, + key_vk=None, + canonical_key_name=None, + canonical_key_char=None, + canonical_key_vk=None, + ) + + with pytest.raises(InvalidCaptureEvent, match="no key identity"): + _convert_action_event(event) + + def test_key_event_with_empty_identities_is_refused(self) -> None: + event = SimpleNamespace( + name="press", + timestamp=1.0, + structural_observation=None, + key_name="", + key_char="", + key_vk="", + canonical_key_name="", + canonical_key_char="", + canonical_key_vk="", + ) + + with pytest.raises(InvalidCaptureEvent, match="no key identity"): + _convert_action_event(event) + class TestWindowBackendAvailability: def test_require_impl_raises_when_no_backend_loaded(self, monkeypatch) -> None: diff --git a/tests/test_processing.py b/tests/test_processing.py index 5ef9b88..b3a3756 100644 --- a/tests/test_processing.py +++ b/tests/test_processing.py @@ -1,6 +1,7 @@ """Tests for event processing pipeline.""" +from openadapt_capture.capture import Action from openadapt_capture.events import ( KeyDownEvent, KeyShortcutEvent, @@ -51,6 +52,31 @@ def test_keeps_valid_key_events(self): result = remove_invalid_keyboard_events(events) assert len(result) == 3 + def test_keeps_canonical_only_key_events(self): + event = KeyDownEvent(timestamp=1.0, canonical_key_name="enter") + + assert remove_invalid_keyboard_events([event]) == [event] + + def test_canonical_only_enter_survives_the_action_pipeline(self): + events = [ + KeyDownEvent(timestamp=1.0, canonical_key_name="enter"), + KeyUpEvent(timestamp=1.1, canonical_key_name="enter"), + ] + + processed = process_events(events) + assert len(processed) == 1 + assert Action(processed[0], object()).keys == ["enter"] + + def test_raw_typed_character_preserves_case(self): + events = [ + KeyDownEvent(timestamp=1.0, key_char="A", canonical_key_char="a"), + KeyUpEvent(timestamp=1.1, key_char="A", canonical_key_char="a"), + ] + + processed = process_events(events) + assert len(processed) == 1 + assert processed[0].text == "A" + class TestRemoveRedundantMouseMoveEvents: """Tests for remove_redundant_mouse_move_events.""" diff --git a/tests/test_stats.py b/tests/test_stats.py index d4502c9..65a2cda 100644 --- a/tests/test_stats.py +++ b/tests/test_stats.py @@ -192,6 +192,8 @@ def test_plot_empty_capture(self, tmp_path): capture, storage = create_capture( tmp_path / "empty_capture", task_description="Empty test", + screen_width=1280, + screen_height=720, ) storage.close() @@ -211,6 +213,8 @@ def test_plot_capture_with_events(self, tmp_path): capture, storage = create_capture( tmp_path / "test_capture", task_description="Test", + screen_width=1280, + screen_height=720, ) # Add some events @@ -242,6 +246,8 @@ def test_plot_capture_saves_to_file(self, tmp_path): capture, storage = create_capture( tmp_path / "test_capture", task_description="Test", + screen_width=1280, + screen_height=720, ) storage.write_event(MouseDownEvent( timestamp=1.0, diff --git a/tests/test_storage.py b/tests/test_storage.py index 9262e3f..40340e7 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -292,6 +292,7 @@ def test_capture_defaults(self): ) assert capture.ended_at is None assert capture.task_description is None + assert capture.pixel_ratio is None assert capture.double_click_interval_seconds == 0.5 assert capture.double_click_distance_pixels == 5.0 assert capture.metadata == {}