Fix/window event race condition

openadapt/alembic/versions/b2a68aec8cb6_add_a11yevent_model.py

@@ -0,0 +1,34 @@
+"""add_a11yevent_model
+
+Revision ID: b2a68aec8cb6
+Revises: bb25e889ad71
+Create Date: 2024-07-17 03:27:27.591270
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'b2a68aec8cb6'
+down_revision = 'bb25e889ad71'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('a11y_event',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('window_event_id', sa.Integer(), nullable=False),
+    sa.Column('data', sa.JSON(), nullable=True),
+    sa.ForeignKeyConstraint(['window_event_id'], ['window_event.id'], name=op.f('fk_a11y_event_window_event_id_window_event')),
+    sa.PrimaryKeyConstraint('id', name=op.f('pk_a11y_event'))
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('a11y_event')
+    # ### end Alembic commands ###

openadapt/db/crud.py

@@ -28,6 +28,7 @@
     Screenshot,
     ScrubbedRecording,
     WindowEvent,
+    A11yEvent,
     copy_sa_instance,
 )
 from openadapt.privacy.base import ScrubbingProvider
@@ -263,6 +264,26 @@ def insert_recording(session: SaSession, recording_data: dict) -> Recording:
     return db_obj
 
 
+def insert_a11y_event(
+    db: SaSession,
+    data: dict,
+) -> None:
+    """Insert an a11y event into the database.
+
+    Args:
+        db: The database session.
+        data: The data associated with the a11y event.
+    """
+    window_event_id = data["window_id"]
+
+    a11y_event = A11yEvent(
+        window_event_id=window_event_id,
+        data=data["state"],
+    )
+    db.add(a11y_event)
+    db.commit()
+
+
 def delete_recording(session: SaSession, recording: Recording) -> None:
     """Remove the recording from the db.
 

openadapt/models.py

@@ -492,6 +492,18 @@ def to_prompt_dict(self) -> dict[str, Any]:
         return action_dict
 
 
+class A11yEvent(db.Base):
+    """Class representing an accessibility (a11y) event in the database."""
+
+    __tablename__ = "a11y_event"
+
+    id = sa.Column(sa.Integer, primary_key=True)
+    window_event_id = sa.Column(sa.ForeignKey("window_event.id"), nullable=False)
+    data = sa.Column(sa.JSON)
+
+    window_event = sa.orm.relationship("WindowEvent", back_populates="a11y_events")
+
+
 class WindowEvent(db.Base):
     """Class representing a window event in the database."""
 
@@ -511,6 +523,7 @@ class WindowEvent(db.Base):
 
     recording = sa.orm.relationship("Recording", back_populates="window_events")
     action_events = sa.orm.relationship("ActionEvent", back_populates="window_event")
+    a11y_events = sa.orm.relationship("A11yEvent", back_populates="window_event")
 
     @classmethod
     def get_active_window_event(
@@ -526,7 +539,16 @@ def get_active_window_event(
         Returns:
             (WindowEvent) the active window event.
         """
-        return WindowEvent(**window.get_active_window_data(include_window_data))
+        window_event_data = window.get_active_window_data(include_window_data)
+        window_event = WindowEvent(**window_event_data)
+
+        if include_window_data:
+            a11y_event_data = window_event_data["state"].copy()
+            a11y_event = A11yEvent(data=a11y_event_data)
+            window_event.a11y_events.append(a11y_event)
+            window_event_data["state"].pop("data")
+
+        return window_event
 
     def scrub(self, scrubber: ScrubbingProvider | TextScrubbingMixin) -> None:
         """Scrub the window event."""

openadapt/record.py

@@ -128,6 +128,7 @@ def process_events(
     screen_write_q: sq.SynchronizedQueue,
     action_write_q: sq.SynchronizedQueue,
     window_write_q: sq.SynchronizedQueue,
+    a11y_write_q: sq.SynchronizedQueue,
     video_write_q: sq.SynchronizedQueue,
     perf_q: sq.SynchronizedQueue,
     recording: Recording,
@@ -136,7 +137,9 @@ def process_events(
     num_screen_events: multiprocessing.Value,
     num_action_events: multiprocessing.Value,
     num_window_events: multiprocessing.Value,
+    num_a11y_events: multiprocessing.Value,
     num_video_events: multiprocessing.Value,
+    window_event_map: dict,
 ) -> None:
     """Process events from the event queue and write them to write queues.
 
@@ -167,6 +170,7 @@ def process_events(
     started = False
     while not terminate_processing.is_set() or not event_q.empty():
         event = event_q.get()
+        window_event_id = event.data["window_id"]
         if not started:
             with started_counter.get_lock():
                 started_counter.value += 1
@@ -244,6 +248,20 @@ def process_events(
                 )
                 num_window_events.value += 1
                 prev_saved_window_timestamp = prev_window_event.timestamp
+        elif event.type == "a11y":
+            window_event = window_event_map.get(window_event_id)
+            if not window_event:
+                logger.warning(f"Discarding A11yEvent with no corresponding WindowEvent: {event}")
+                continue
+            event.data["window_event_timestamp"] = window_event.timestamp
+            process_event(
+                event,
+                a11y_write_q,
+                write_a11y_event,
+                recording,
+                perf_q,
+            )
+            num_a11y_events.value += 1
         else:
             raise Exception(f"unhandled {event.type=}")
         del prev_event
@@ -316,6 +334,23 @@ def write_window_event(
     perf_q.put((event.type, event.timestamp, utils.get_timestamp()))
 
 
+def write_a11y_event(
+    db: crud.SaSession,
+    event: Event,
+    perf_q: sq.SynchronizedQueue,
+) -> None:
+    """Write an accessibility (a11y) event to the database and update the performance queue.
+
+    Args:
+        db: The database session.
+        event: An a11y event to be written.
+        perf_q: A queue for collecting performance data.
+    """
+    assert event.type == "a11y", event
+    crud.insert_a11y_event(db, event.data)
+    perf_q.put((event.type, event.timestamp, utils.get_timestamp()))
+
+
 @utils.trace(logger)
 def write_events(
     event_type: str,
@@ -696,6 +731,7 @@ def read_window_events(
     terminate_processing: multiprocessing.Event,
     recording: Recording,
     started_counter: multiprocessing.Value,
+    window_event_map: dict,
 ) -> None:
     """Read window events and add them to the event queue.
 
@@ -735,13 +771,13 @@ def read_window_events(
             logger.info(f"{_window_data=}")
         if window_data != prev_window_data:
             logger.debug("Queuing window event for writing")
-            event_q.put(
-                Event(
-                    utils.get_timestamp(),
-                    "window",
-                    window_data,
-                )
+            event = Event(
+                utils.get_timestamp(),
+                "window",
+                window_data, 
             )
+            window_event_map[window_data["window_id"]] = event  # Store window event in map
+            event_q.put(event)
         prev_window_data = window_data
 
 
@@ -1175,6 +1211,7 @@ def record(
     screen_write_q = sq.SynchronizedQueue()
     action_write_q = sq.SynchronizedQueue()
     window_write_q = sq.SynchronizedQueue()
+    a11y_write_q = sq.SynchronizedQueue()
     video_write_q = sq.SynchronizedQueue()
     # TODO: save write times to DB; display performance plot in visualize.py
     perf_q = sq.SynchronizedQueue()
@@ -1183,9 +1220,11 @@ def record(
     started_counter = multiprocessing.Value("i", 0)
     expected_starts = 9
 
+    window_event_map = {}
+
     window_event_reader = threading.Thread(
         target=read_window_events,
-        args=(event_q, terminate_processing, recording, started_counter),
+        args=(event_q, terminate_processing, recording, started_counter, window_event_map),
     )
     window_event_reader.start()
 
@@ -1210,6 +1249,7 @@ def record(
     num_action_events = multiprocessing.Value("i", 0)
     num_screen_events = multiprocessing.Value("i", 0)
     num_window_events = multiprocessing.Value("i", 0)
+    num_a11y_events = multiprocessing.Value("i", 0)
     num_video_events = multiprocessing.Value("i", 0)
 
     event_processor = threading.Thread(
@@ -1219,6 +1259,7 @@ def record(
             screen_write_q,
             action_write_q,
             window_write_q,
+            a11y_write_q,
             video_write_q,
             perf_q,
             recording,
@@ -1227,7 +1268,9 @@ def record(
             num_screen_events,
             num_action_events,
             num_window_events,
+            num_a11y_events,
             num_video_events,
+            window_event_map
         ),
     )
     event_processor.start()
@@ -1277,6 +1320,21 @@ def record(
     )
     window_event_writer.start()
 
+    a11y_event_writer = multiprocessing.Process(  
+        target=write_events,
+        args=(
+            "a11y",
+            write_a11y_event,  
+            a11y_write_q,
+            num_a11y_events,
+            perf_q,
+            recording,
+            terminate_processing,
+            started_counter,
+        ),
+    )
+    a11y_event_writer.start()
+
     if config.RECORD_VIDEO:
         expected_starts += 1
         video_writer = multiprocessing.Process(
@@ -1379,6 +1437,7 @@ def record(
     screen_event_writer.join()
     action_event_writer.join()
     window_event_writer.join()
+    a11y_event_writer.join()  # Join a11y_event_writer
     if config.RECORD_VIDEO:
         video_writer.join()
     if config.RECORD_AUDIO:

openadapt/window/__init__.py

@@ -82,4 +82,4 @@ def get_active_element_state(x: int, y: int) -> dict | None:
         return impl.get_active_element_state(x, y)
     except Exception as exc:
         logger.warning(f"{exc=}")
-        return None
+        return None

openadapt/window/_windows.py

@@ -195,4 +195,4 @@ def main() -> None:
 
 
 if __name__ == "__main__":
-    main()
+    main()