From 4054fc66fa6be8279ee1eccde244b004bd53676e Mon Sep 17 00:00:00 2001 From: Guillermo Dylan Carvajal Aza Date: Tue, 5 May 2026 13:11:23 +0200 Subject: [PATCH 1/2] fix: Robust jerkiness methods --- src/pywib/__init__.py | 5 +-- src/pywib/core/__init__.py | 3 +- src/pywib/core/movement/movement.py | 36 +++++++++++++----- src/pywib/utils/movement.py | 8 ++-- src/pywib/utils/utils.py | 12 ++++-- src/pywib/utils/visualization.py | 59 ++++++++++++++++++++--------- test/test_movement.py | 27 +++++++++++++ 7 files changed, 111 insertions(+), 39 deletions(-) diff --git a/src/pywib/__init__.py b/src/pywib/__init__.py index fac4457..9fb1753 100644 --- a/src/pywib/__init__.py +++ b/src/pywib/__init__.py @@ -28,7 +28,6 @@ # Utility functions "validate_dataframe", "validate_dataframe_keyboard", - "extract_trace", "visualize_trace", "compute_space_time_diff", "extract_traces_by_session", @@ -45,7 +44,7 @@ "acceleration_metrics", "jerkiness_metrics", "deviations", - "auc" + "auc", # Mouse functions "number_of_clicks", @@ -55,7 +54,7 @@ "typing_speed", "typing_speed_metrics", "backspace_usage", - "typing_durations" + "typing_durations", # Timing "pauses_metrics", diff --git a/src/pywib/core/__init__.py b/src/pywib/core/__init__.py index 95d3296..8da0d98 100644 --- a/src/pywib/core/__init__.py +++ b/src/pywib/core/__init__.py @@ -23,10 +23,9 @@ "jerkiness_metrics", "click_slip", "number_of_clicks", - "deviations" + "deviations", "typing_speed", "typing_speed_metrics", "backspace_usage", "typing_durations", - "typing_speed" ] \ No newline at end of file diff --git a/src/pywib/core/movement/movement.py b/src/pywib/core/movement/movement.py index 74aa06b..28cba33 100644 --- a/src/pywib/core/movement/movement.py +++ b/src/pywib/core/movement/movement.py @@ -164,15 +164,33 @@ def jerkiness_metrics(df: pd.DataFrame, traces: dict[str, list[pd.DataFrame]] = """ validate_any_not_none(df, traces) - - if(df is not None): - if((ColumnNames.JERKINESS not in df.columns) or (traces is None)): - validate_dataframe(df) - if(ColumnNames.ACCELERATION not in df.columns): - if(ColumnNames.VELOCITY not in df.columns): - traces = velocity(df, per_traces=True) - traces = acceleration(df, traces, per_traces=True) - traces = jerkiness(df, traces, per_traces=True) + + traces_missing_jerkiness = False + if traces is not None: + traces_missing_jerkiness = any( + ColumnNames.JERKINESS not in trace.columns + for session_traces in traces.values() + for trace in session_traces + ) + + if traces_missing_jerkiness and df is None: + raise ValueError( + f"Provided traces do not contain '{ColumnNames.JERKINESS}'. " + f"Please provide a DataFrame so {ColumnNames.JERKINESS} can be computed, " + f"or pass traces already containing '{ColumnNames.JERKINESS}'." + ) + + if (df is not None) and ( + (ColumnNames.JERKINESS not in df.columns) or + (traces is None) or + traces_missing_jerkiness + ): + validate_dataframe(df) + if ColumnNames.ACCELERATION not in df.columns: + if ColumnNames.VELOCITY not in df.columns: + traces = velocity(df, per_traces=True) + traces = acceleration(df, traces, per_traces=True) + traces = jerkiness(df, traces, per_traces=True) return compute_metrics_from_traces( df=df, diff --git a/src/pywib/utils/movement.py b/src/pywib/utils/movement.py index eeb63dc..f298534 100644 --- a/src/pywib/utils/movement.py +++ b/src/pywib/utils/movement.py @@ -77,8 +77,8 @@ def acceleration_df(df: pd.DataFrame) -> pd.DataFrame: if(ColumnNames.VELOCITY not in df.columns): df = velocity_df(df) - df['acceleration'] = df['velocity'].diff().fillna(0) / df[ColumnNames.DT] - df['acceleration'] = df['acceleration'].fillna(0) + df[ColumnNames.ACCELERATION] = df[ColumnNames.VELOCITY].diff().fillna(0) / df[ColumnNames.DT] + df[ColumnNames.ACCELERATION] = df[ColumnNames.ACCELERATION].fillna(0) return df @@ -114,8 +114,8 @@ def jerkiness_df(df: pd.DataFrame) -> pd.DataFrame: if(ColumnNames.ACCELERATION not in df.columns): df = acceleration_df(df) - df['jerkiness'] = df['acceleration'].diff().fillna(0) / df[ColumnNames.DT] - df['jerkiness'] = df['jerkiness'].fillna(0) + df[ColumnNames.JERKINESS] = df[ColumnNames.ACCELERATION].diff().fillna(0) / df[ColumnNames.DT] + df[ColumnNames.JERKINESS] = df[ColumnNames.JERKINESS].fillna(0) return df diff --git a/src/pywib/utils/utils.py b/src/pywib/utils/utils.py index 39f4c7e..1104d1f 100644 --- a/src/pywib/utils/utils.py +++ b/src/pywib/utils/utils.py @@ -24,9 +24,9 @@ def compute_space_time_diff(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df.sort_values(by=[ColumnNames.TIME_STAMP], inplace=True) df[ColumnNames.TIME_STAMP] = pd.to_numeric(df[ColumnNames.TIME_STAMP], errors='coerce') - df['dt'] = df.groupby([ColumnNames.SESSION_ID])[ColumnNames.TIME_STAMP].diff().fillna(0) - df['dx'] = df.groupby([ColumnNames.SESSION_ID])[ColumnNames.X].diff().fillna(0) - df['dy'] = df.groupby([ColumnNames.SESSION_ID])[ColumnNames.Y].diff().fillna(0) + df[ColumnNames.DT] = df.groupby([ColumnNames.SESSION_ID])[ColumnNames.TIME_STAMP].diff().fillna(0) + df[ColumnNames.DX] = df.groupby([ColumnNames.SESSION_ID])[ColumnNames.X].diff().fillna(0) + df[ColumnNames.DY] = df.groupby([ColumnNames.SESSION_ID])[ColumnNames.Y].diff().fillna(0) return df def compute_metrics_from_traces( @@ -78,6 +78,12 @@ def compute_metrics_from_traces( metrics = {} for session_id, session_traces in traces.items(): if(len(session_traces) > 0): + for trace_index, trace in enumerate(session_traces): + if column_name not in trace.columns: + raise ValueError( + f"Missing required column '{column_name}' in " + f"session '{session_id}', trace index {trace_index}." + ) values = pd.concat([trace[column_name] for trace in session_traces]) if preprocess_fn: values = preprocess_fn(values) diff --git a/src/pywib/utils/visualization.py b/src/pywib/utils/visualization.py index 0764cf7..f0ea0b0 100644 --- a/src/pywib/utils/visualization.py +++ b/src/pywib/utils/visualization.py @@ -7,16 +7,26 @@ from pywib.constants import ColumnNames from pywib.utils.validation import validate_dataframe_keyboard -def visualize_trace(df, stroke_indices, stroke_id): +def visualize_trace(df, stroke_indices, stroke_id, plot_name: str = None, plot: bool = True, show_info: bool = False, show_optimal_line: bool = False): """ - Generates a plot visualizing the trace of a stroke. + Generates (and optionally saves) a plot visualizing the trace of a stroke. Parameters: df (pd.DataFrame): DataFrame containing the stroke data with 'x', 'y and 'timeStamp' columns. - stroke_indices (list): List of indices corresponding to the stroke in the DataFrame. Can be obtained using df.index. + stroke_indices (list): List of indices corresponding to the stroke in the DataFrame. Can be obtained using df.index stroke_id (str): Identifier for the stroke to be displayed in the title. - Returns: - None: Displays a plot of the stroke trace. + plot_name (str, optional): If provided, saves the plot to this file path. + plot (bool): Whether to display the plot. + """ + """ + Generates (and optionally saves) a plot visualizing the trace of a stroke. + + Parameters: + df (pd.DataFrame): DataFrame containing the stroke data with 'x', 'y and 'timeStamp' columns. + stroke_indices (list): List of indices corresponding to the stroke in the DataFrame. Can be obtained using df.index + stroke_id (str): Identifier for the stroke to be displayed in the title. + plot_name (str, optional): If provided, saves the plot to this file path. + plot (bool): Whether to display the plot. """ stroke_data = df.loc[stroke_indices] plt.figure(figsize=(10, 8)) @@ -24,19 +34,32 @@ def visualize_trace(df, stroke_indices, stroke_id): x_start, y_start = stroke_data[ColumnNames.X].iloc[0], stroke_data[ColumnNames.Y].iloc[0] x_end, y_end = stroke_data[ColumnNames.X].iloc[-1], stroke_data[ColumnNames.Y].iloc[-1] - plt.plot([x_start, x_end], [y_start, y_end], 'r--', linewidth=2, label='Optimal trace') - - plt.plot(x_start, y_start, 'go', markersize=8, label='Inicio') - plt.plot(x_end, y_end, 'ro', markersize=8, label='Fin') - - duration = stroke_data['timeStamp'].iloc[-1] - stroke_data['timeStamp'].iloc[0] - plt.xlabel('X (píxeles)') - plt.ylabel('Y (píxeles)') - plt.title(f'Trace {stroke_id} - Duration: {duration:.0f}ms - Points: {len(stroke_data)}') - plt.legend() - plt.grid(True, alpha=0.3) - plt.gca().invert_yaxis() - plt.show() + + if(show_optimal_line): + plt.plot([x_start, x_end], [y_start, y_end], 'r--', linewidth=2, label='Optimal trace') + + plt.plot(x_start, y_start, 'go', markersize=8, label='Start') + plt.plot(x_end, y_end, 'ro', markersize=8, label='End') + + + if(show_info): + duration = stroke_data[ColumnNames.TIME_STAMP].iloc[-1] - stroke_data[ColumnNames.TIME_STAMP].iloc[0] + plt.xlabel('X (px)') + plt.ylabel('Y (px)') + plt.title(f'Trace {stroke_id} - Duration: {duration:.0f}ms - Points: {len(stroke_data)}') + plt.legend() + plt.grid(True, alpha=0.3) + plt.gca().invert_yaxis() + else: + plt.axis('off') + + if plot_name: + plt.savefig(plot_name, bbox_inches='tight', dpi=300) + + if plot: + plt.show() + else: + plt.close() def video_from_trace(df, user_id, outfile: str, width=640, height=480, fps=30, colored=False): """ diff --git a/test/test_movement.py b/test/test_movement.py index 0286a1d..9d22abc 100644 --- a/test/test_movement.py +++ b/test/test_movement.py @@ -178,6 +178,33 @@ def test_jerkiness_from_df(self): self.assertGreaterEqual(session['mean'], 0) self.assertGreaterEqual(session['max'], session['min']) + def test_jerkiness_metrics_recomputes_when_traces_missing_column(self): + partial_traces = velocity(self.test_data.copy(), per_traces=True) + metrics = jerkiness_metrics(self.test_data.copy(), traces=partial_traces) + + self.assertIsInstance(metrics, dict) + self.assertTrue(len(metrics) > 0) + for _, session in metrics.items(): + self.assertIn('mean', session) + self.assertIn('max', session) + self.assertIn('min', session) + + def test_jerkiness_metrics_raises_with_incomplete_traces_and_no_df(self): + partial_traces = velocity(self.test_data.copy(), per_traces=True) + with self.assertRaises(ValueError): + jerkiness_metrics(None, traces=partial_traces) + + def test_jerkiness_metrics_with_precomputed_jerkiness_traces(self): + full_traces = jerkiness(self.test_data.copy(), per_traces=True) + metrics = jerkiness_metrics(None, traces=full_traces) + + self.assertIsInstance(metrics, dict) + self.assertTrue(len(metrics) > 0) + for _, session in metrics.items(): + self.assertIn('mean', session) + self.assertIn('max', session) + self.assertIn('min', session) + if __name__ == '__main__': From 8765f4a689cdc242c288400fd68ec828879a0464 Mon Sep 17 00:00:00 2001 From: Guillermo Dylan Carvajal Aza Date: Wed, 6 May 2026 12:52:59 +0200 Subject: [PATCH 2/2] fix: defensive metrics update + tests --- docs/source/mouse.rst | 5 ++- src/pywib/core/movement/movement.py | 55 ++++++++++++++++------------- src/pywib/utils/visualization.py | 10 ------ test/test_movement.py | 41 ++++++++++++++++++++- test/test_public_exports.py | 50 ++++++++++++++++++++++++++ 5 files changed, 122 insertions(+), 39 deletions(-) create mode 100644 test/test_public_exports.py diff --git a/docs/source/mouse.rst b/docs/source/mouse.rst index 43492d1..9d4b2f3 100644 --- a/docs/source/mouse.rst +++ b/docs/source/mouse.rst @@ -7,8 +7,7 @@ Clicks ------ From a dataset of mouse interaction points, there are several click-related metrics that can be computed to analyze user behavior and interaction patterns. -Two common click metrics are: +Common click metrics are: 1. **Number of Clicks**: This metric counts the total number of clicks recorded during a session. See the function implementation in :py:func:`~pywib.number_of_clicks`. -2. **Click Slip**: This metric measures the deviation of click positions from the intended targets, providing insights into user accuracy. See the function implementation in :py:func:`~pywib.click_slip`. -.. 3. **Click Duration**: This metric measures the time duration of each click action, providing insights into user interaction speed. See the function implementation in :py:func:`~pywib.click_duration`. \ No newline at end of file +2. **Click Slip**: This metric measures the deviation of click positions from the intended targets, providing insights into user accuracy. See the function implementation in :py:func:`~pywib.click_slip`. \ No newline at end of file diff --git a/src/pywib/core/movement/movement.py b/src/pywib/core/movement/movement.py index 28cba33..6a615a0 100644 --- a/src/pywib/core/movement/movement.py +++ b/src/pywib/core/movement/movement.py @@ -7,6 +7,15 @@ from pywib.utils.movement import velocity_traces_parallel from pywib.utils.validation import validate_any_not_none +def _traces_missing_column(traces: dict[str, list[pd.DataFrame]] | None, column_name: str) -> bool: + if traces is None: + return False + return any( + column_name not in trace.columns + for session_traces in traces.values() + for trace in session_traces + ) + def velocity(df: pd.DataFrame = None, traces: dict[str, list[pd.DataFrame]] = None, per_traces: bool = True, parallel:bool = False, n_jobs: int = 2) -> dict[str, list[pd.DataFrame]]: """ Function to calculate velocity for either a single DataFrame or a traces dictionary. @@ -44,7 +53,7 @@ def velocity(df: pd.DataFrame = None, traces: dict[str, list[pd.DataFrame]] = No return velocity_traces(traces) -def velocity_metrics(df: pd.DataFrame, traces: dict[str, list[pd.DataFrame]] = None) -> dict: +def velocity_metrics(df: pd.DataFrame = None, traces: dict[str, list[pd.DataFrame]] = None) -> dict: """ Calculate velocity metrics for the given DataFrame or traces. This function computes the mean, max, and min velocity for each session. @@ -59,11 +68,19 @@ def velocity_metrics(df: pd.DataFrame, traces: dict[str, list[pd.DataFrame]] = N validate_any_not_none(df, traces) + if (df is not None) and ( + (ColumnNames.VELOCITY not in df.columns) or + (traces is None) or + _traces_missing_column(traces, ColumnNames.VELOCITY) + ): + validate_dataframe(df) + traces = velocity(df, per_traces=True) + return compute_metrics_from_traces( df=df, traces=traces, column_name=ColumnNames.VELOCITY, - compute_traces_fn=velocity, + compute_traces_fn=lambda _: traces, preprocess_fn=lambda s: s[s > 0] # Exclude zero velocities ) @@ -88,7 +105,7 @@ def acceleration(df: pd.DataFrame = None, traces: dict[str, list[pd.DataFrame]] # Compute acceleration for each trace return acceleration_traces(traces) -def acceleration_metrics(df: pd.DataFrame, traces: dict[str, list[pd.DataFrame]] = None) -> dict: +def acceleration_metrics(df: pd.DataFrame = None, traces: dict[str, list[pd.DataFrame]] = None) -> dict: """ Calculate acceleration metrics for the given DataFrame or traces. This function computes the mean, max, and min acceleration for each session. @@ -103,12 +120,15 @@ def acceleration_metrics(df: pd.DataFrame, traces: dict[str, list[pd.DataFrame]] validate_any_not_none(df, traces) - if(df is not None): - if (ColumnNames.ACCELERATION not in df.columns) or (traces is None): - validate_dataframe(df) - if ColumnNames.VELOCITY not in df.columns: - traces = velocity(df, per_traces=True) - traces = acceleration(df, traces, per_traces=True) + if (df is not None) and ( + (ColumnNames.ACCELERATION not in df.columns) or + (traces is None) or + _traces_missing_column(traces, ColumnNames.ACCELERATION) + ): + validate_dataframe(df) + if ColumnNames.VELOCITY not in df.columns: + traces = velocity(df, per_traces=True) + traces = acceleration(df, traces, per_traces=True) return compute_metrics_from_traces( df=df, @@ -165,25 +185,10 @@ def jerkiness_metrics(df: pd.DataFrame, traces: dict[str, list[pd.DataFrame]] = validate_any_not_none(df, traces) - traces_missing_jerkiness = False - if traces is not None: - traces_missing_jerkiness = any( - ColumnNames.JERKINESS not in trace.columns - for session_traces in traces.values() - for trace in session_traces - ) - - if traces_missing_jerkiness and df is None: - raise ValueError( - f"Provided traces do not contain '{ColumnNames.JERKINESS}'. " - f"Please provide a DataFrame so {ColumnNames.JERKINESS} can be computed, " - f"or pass traces already containing '{ColumnNames.JERKINESS}'." - ) - if (df is not None) and ( (ColumnNames.JERKINESS not in df.columns) or (traces is None) or - traces_missing_jerkiness + _traces_missing_column(traces, ColumnNames.JERKINESS) ): validate_dataframe(df) if ColumnNames.ACCELERATION not in df.columns: diff --git a/src/pywib/utils/visualization.py b/src/pywib/utils/visualization.py index f0ea0b0..9fa1fdc 100644 --- a/src/pywib/utils/visualization.py +++ b/src/pywib/utils/visualization.py @@ -11,16 +11,6 @@ def visualize_trace(df, stroke_indices, stroke_id, plot_name: str = None, plot: """ Generates (and optionally saves) a plot visualizing the trace of a stroke. - Parameters: - df (pd.DataFrame): DataFrame containing the stroke data with 'x', 'y and 'timeStamp' columns. - stroke_indices (list): List of indices corresponding to the stroke in the DataFrame. Can be obtained using df.index - stroke_id (str): Identifier for the stroke to be displayed in the title. - plot_name (str, optional): If provided, saves the plot to this file path. - plot (bool): Whether to display the plot. - """ - """ - Generates (and optionally saves) a plot visualizing the trace of a stroke. - Parameters: df (pd.DataFrame): DataFrame containing the stroke data with 'x', 'y and 'timeStamp' columns. stroke_indices (list): List of indices corresponding to the stroke in the DataFrame. Can be obtained using df.index diff --git a/test/test_movement.py b/test/test_movement.py index 9d22abc..5de1c4f 100644 --- a/test/test_movement.py +++ b/test/test_movement.py @@ -8,7 +8,7 @@ import_pyModule() from pywib import (velocity, acceleration, compute_space_time_diff, velocity_metrics, acceleration_metrics, - jerkiness, jerkiness_metrics) + jerkiness, jerkiness_metrics, extract_traces_by_session) # Cambiar a True solo al probar en desarrollo DEBUG = True @@ -86,6 +86,24 @@ def test_velocity_metrics_from_velocity_traces(self): assert_between_zero_inf(self, session_id, 'max') assert_between_zero_inf(self, session_id, 'min') + def test_velocity_metrics_recomputes_when_traces_missing_column(self): + """Recompute velocity traces from df when provided traces lack velocity column.""" + partial_traces = extract_traces_by_session(self.test_data.copy()) + metrics = velocity_metrics(self.test_data.copy(), traces=partial_traces) + + self.assertIsInstance(metrics, dict) + self.assertTrue(len(metrics) > 0) + for _, session in metrics.items(): + self.assertIn('mean', session) + self.assertIn('max', session) + self.assertIn('min', session) + + def test_velocity_metrics_raises_with_incomplete_traces_and_no_df(self): + """Raise ValueError when traces lack velocity and no dataframe is available.""" + partial_traces = extract_traces_by_session(self.test_data.copy()) + with self.assertRaises(ValueError): + velocity_metrics(None, traces=partial_traces) + def test_acceleration(self): """Test acceleration calculation""" df = compute_space_time_diff(self.test_data.copy()) @@ -143,6 +161,24 @@ def test_acceleration_metrics_from_aceleration_df(self): self.assertGreaterEqual(session['mean'], 0) self.assertGreaterEqual(session['max'], session['min']) + def test_acceleration_metrics_recomputes_when_traces_missing_column(self): + """Recompute acceleration traces from df when provided traces lack acceleration.""" + partial_traces = velocity(self.test_data.copy(), per_traces=True) + metrics = acceleration_metrics(self.test_data.copy(), traces=partial_traces) + + self.assertIsInstance(metrics, dict) + self.assertTrue(len(metrics) > 0) + for _, session in metrics.items(): + self.assertIn('mean', session) + self.assertIn('max', session) + self.assertIn('min', session) + + def test_acceleration_metrics_raises_with_incomplete_traces_and_no_df(self): + """Raise ValueError when traces lack acceleration and no dataframe is available.""" + partial_traces = velocity(self.test_data.copy(), per_traces=True) + with self.assertRaises(ValueError): + acceleration_metrics(None, traces=partial_traces) + def test_jerkiness(self): """Test jerkiness calculation""" jk_df = jerkiness(self.test_data.copy(), per_traces=True) @@ -179,6 +215,7 @@ def test_jerkiness_from_df(self): self.assertGreaterEqual(session['max'], session['min']) def test_jerkiness_metrics_recomputes_when_traces_missing_column(self): + """Recompute jerkiness traces from df when provided traces lack jerkiness.""" partial_traces = velocity(self.test_data.copy(), per_traces=True) metrics = jerkiness_metrics(self.test_data.copy(), traces=partial_traces) @@ -190,11 +227,13 @@ def test_jerkiness_metrics_recomputes_when_traces_missing_column(self): self.assertIn('min', session) def test_jerkiness_metrics_raises_with_incomplete_traces_and_no_df(self): + """Raise ValueError when traces lack jerkiness and no dataframe is available.""" partial_traces = velocity(self.test_data.copy(), per_traces=True) with self.assertRaises(ValueError): jerkiness_metrics(None, traces=partial_traces) def test_jerkiness_metrics_with_precomputed_jerkiness_traces(self): + """Use precomputed jerkiness traces directly without recomputation.""" full_traces = jerkiness(self.test_data.copy(), per_traces=True) metrics = jerkiness_metrics(None, traces=full_traces) diff --git a/test/test_public_exports.py b/test/test_public_exports.py new file mode 100644 index 0000000..e6d76f2 --- /dev/null +++ b/test/test_public_exports.py @@ -0,0 +1,50 @@ +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(__file__)) +from utils import import_pyModule + +import_pyModule() + +import pywib +import pywib.core as core +import pywib.utils as utils + + +class TestPublicExports(unittest.TestCase): + def test_root_all_has_unique_entries(self): + self.assertEqual(len(pywib.__all__), len(set(pywib.__all__))) + + def test_root_all_symbols_are_resolvable(self): + for name in pywib.__all__: + self.assertTrue( + hasattr(pywib, name), + f"pywib.__all__ includes '{name}' but it is not importable from pywib", + ) + + def test_core_all_symbols_are_resolvable(self): + for name in core.__all__: + self.assertTrue( + hasattr(core, name), + f"pywib.core.__all__ includes '{name}' but it is not importable from pywib.core", + ) + + def test_utils_all_symbols_are_resolvable(self): + for name in utils.__all__: + self.assertTrue( + hasattr(utils, name), + f"pywib.utils.__all__ includes '{name}' but it is not importable from pywib.utils", + ) + + def test_core_exports_are_available_from_root(self): + missing = [name for name in core.__all__ if name not in pywib.__all__] + self.assertEqual( + missing, + [], + f"pywib.core exports missing from pywib root __all__: {missing}", + ) + + +if __name__ == "__main__": + unittest.main()