bead.behavioral

Behavioral data analytics and extraction utilities for judgment experiment responses.

Analytics

analytics

Behavioral analytics models for bead.

This module provides data models for per-judgment behavioral metrics and participant-level summaries, linking slopit behavioral data to bead's item-based experimental structure.

JudgmentAnalytics

Bases: BeadBaseModel

Behavioral analytics for a single judgment.

Links slopit behavioral data to a specific item judgment, enabling analysis of participant behavior during individual responses.

Attributes:

Name	Type	Description
item_id	UUID	Reference to the Item being judged.
participant_id	str	Participant identifier (from slopit session).
trial_index	int	Zero-indexed position in the experiment session.
session_id	str	Slopit session identifier.
response_value	JsonValue	The participant's response value.
response_time_ms	int	Response time in milliseconds.
keystroke_metrics	KeystrokeMetrics | None	Keystroke-derived behavioral metrics.
focus_metrics	FocusMetrics | None	Focus and visibility metrics.
timing_metrics	TimingMetrics | None	Timing-related metrics.
paste_event_count	int	Number of paste events during this judgment.
flags	list[AnalysisFlag]	Behavioral analysis flags from slopit analyzers.
max_severity	Severity | None	Maximum severity among all flags.

Examples:

>>> from uuid import uuid4
>>> analytics = JudgmentAnalytics(
...     item_id=uuid4(),
...     participant_id="participant_001",
...     trial_index=0,
...     session_id="session_001",
...     response_value=5,
...     response_time_ms=2500,
... )
>>> analytics.is_flagged
False

has_paste_events: bool property

Check if this judgment had any paste events.

Returns:

Type	Description
bool	True if paste events occurred during this judgment.

is_flagged: bool property

Check if this judgment has any behavioral flags.

Returns:

Type	Description
bool	True if any analysis flags are present.

get_flag_types() -> list[str]

Get list of flag types for this judgment.

Returns:

Type	Description
list[str]	List of flag type identifiers.

ParticipantBehavioralSummary

Bases: BeadBaseModel

Aggregated behavioral metrics for a participant across all judgments.

Provides summary statistics useful for identifying participants who may require exclusion from analysis.

Attributes:

Name	Type	Description
participant_id	str	Participant identifier.
session_id	str	Slopit session identifier.
total_judgments	int	Total number of judgments analyzed.
flagged_judgments	int	Number of judgments with behavioral flags.
mean_response_time_ms	float	Mean response time across all judgments.
mean_iki	float | None	Mean inter-keystroke interval (averaged across judgments).
total_keystrokes	int	Total keystrokes across all judgments.
total_paste_events	int	Total paste events across all judgments.
total_blur_events	int	Total window blur events.
total_blur_duration_ms	float	Total time with window blurred in milliseconds.
flag_counts	dict[str, int]	Count of each flag type.
max_severity	Severity | None	Maximum flag severity across all judgments.

Examples:

>>> summary = ParticipantBehavioralSummary(
...     participant_id="participant_001",
...     session_id="session_001",
...     total_judgments=50,
...     flagged_judgments=3,
...     mean_response_time_ms=2500.0,
... )
>>> summary.flag_rate
0.06

flag_rate: float property

Calculate proportion of judgments that were flagged.

Returns:

Type	Description
float	Proportion of flagged judgments (0.0 to 1.0).

has_paste_events: bool property

Check if participant had any paste events.

Returns:

Type	Description
bool	True if any paste events occurred.

AnalyticsCollection

Bases: BeadBaseModel

Collection of judgment analytics with I/O and filtering support.

Provides methods for persisting analytics to JSONL files, converting to DataFrames, and filtering by behavioral flags.

Attributes:

Name	Type	Description
name	str	Collection name (e.g., study identifier).
analytics	list[JudgmentAnalytics]	List of per-judgment analytics records.

Examples:

>>> from uuid import uuid4
>>> collection = AnalyticsCollection(name="study_001")
>>> analytics = JudgmentAnalytics(
...     item_id=uuid4(),
...     participant_id="p001",
...     trial_index=0,
...     session_id="s001",
...     response_time_ms=2000,
... )
>>> collection.add_analytics(analytics)
>>> len(collection)
1

__len__() -> int

Return number of analytics records.

Returns:

Type	Description
int	Number of analytics records in the collection.

add_analytics(analytics: JudgmentAnalytics) -> None

Add a single analytics record to the collection.

Parameters:

Name	Type	Description	Default
analytics	JudgmentAnalytics	Analytics record to add.	required

add_many(analytics_list: list[JudgmentAnalytics]) -> None

Add multiple analytics records to the collection.

Parameters:

Name	Type	Description	Default
analytics_list	list[JudgmentAnalytics]	List of analytics records to add.	required

get_by_participant(participant_id: str) -> list[JudgmentAnalytics]

Get all analytics for a specific participant.

Parameters:

Name	Type	Description	Default
participant_id	str	Participant identifier to filter by.	required

Returns:

Type	Description
list[JudgmentAnalytics]	Analytics records for the participant.

get_by_item(item_id: UUID) -> list[JudgmentAnalytics]

Get all analytics for a specific item.

Parameters:

Name	Type	Description	Default
item_id	UUID	Item UUID to filter by.	required

Returns:

Type	Description
list[JudgmentAnalytics]	Analytics records for the item.

filter_flagged(min_severity: Severity | None = None, exclude_flagged: bool = False) -> AnalyticsCollection

Filter analytics by flag status.

Parameters:

Name	Type	Description	Default
min_severity	Severity | None	If provided, only include analytics with flags at or above this severity. Severity order: info < low < medium < high.	None
exclude_flagged	bool	If True, exclude flagged records. If False, include only flagged records. Default is False (return flagged records).	False

Returns:

Type	Description
AnalyticsCollection	New collection with filtered analytics.

get_participant_ids() -> list[str]

Get unique participant IDs in the collection.

Returns:

Type	Description
list[str]	List of unique participant IDs.

get_participant_summaries() -> list[ParticipantBehavioralSummary]

Generate behavioral summaries for all participants.

Returns:

Type	Description
list[ParticipantBehavioralSummary]	Summary for each participant in the collection.

to_jsonl(path: Path | str) -> None

Write analytics to JSONL file.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to output file.	required

from_jsonl(path: Path | str, name: str = 'loaded_analytics') -> AnalyticsCollection classmethod

Load analytics from JSONL file.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to JSONL file.	required
name	str	Name for the collection.	'loaded_analytics'

Returns:

Type	Description
AnalyticsCollection	Collection with loaded analytics.

to_dataframe(backend: Literal['pandas', 'polars'] = 'pandas', include_metrics: bool = True, include_flags: bool = True) -> DataFrame

Convert to pandas or polars DataFrame.

Parameters:

Name	Type	Description	Default
backend	Literal['pandas', 'polars']	DataFrame backend to use (default: "pandas").	'pandas'
include_metrics	bool	If True, flatten behavioral metrics into columns.	True
include_flags	bool	If True, include flag-related columns.	True

Returns:

Type	Description
DataFrame	pandas or polars DataFrame with analytics data.

Extraction

extraction

Behavioral data extraction from slopit sessions.

This module provides functions for extracting per-judgment behavioral analytics from slopit session data, using slopit's IO loaders and analysis pipeline.

extract_from_trial(trial: SlopitTrial, session: SlopitSession, item_id_key: str = 'item_id') -> JudgmentAnalytics | None

Extract behavioral analytics from a single slopit trial.

Parameters:

Name	Type	Description	Default
trial	SlopitTrial	Slopit trial data.	required
session	SlopitSession	Parent session for participant context.	required
item_id_key	str	Key in platform_data containing the item UUID.	'item_id'

Returns:

Type	Description
JudgmentAnalytics | None	Analytics record, or None if item_id not found in trial.

extract_from_session(session: SlopitSession, item_id_key: str = 'item_id') -> list[JudgmentAnalytics]

Extract behavioral analytics from all trials in a slopit session.

Parameters:

Name	Type	Description	Default
session	SlopitSession	Slopit session containing trial data.	required
item_id_key	str	Key in platform_data containing the item UUID.	'item_id'

Returns:

Type	Description
list[JudgmentAnalytics]	Analytics records for trials with valid item_id.

extract_from_file(path: Path | str, item_id_key: str = 'item_id') -> list[JudgmentAnalytics]

Extract behavioral analytics from a slopit session file.

Uses slopit's load_session() to automatically detect format.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to session file (JSON or JATOS format).	required
item_id_key	str	Key in platform_data containing the item UUID.	'item_id'

Returns:

Type	Description
list[JudgmentAnalytics]	Analytics records from the session.

Examples:

>>> analytics = extract_from_file("data/session_001.json")
>>> len(analytics)
50

extract_from_directory(path: Path | str, pattern: str = '*', item_id_key: str = 'item_id', name: str | None = None) -> AnalyticsCollection

Extract behavioral analytics from all session files in a directory.

Uses slopit's load_sessions() to load all files.

Parameters:

Name	Type	Description	Default
path	Path | str	Directory containing session files.	required
pattern	str	Glob pattern for file matching (default: "*").	'*'
item_id_key	str	Key in platform_data containing the item UUID.	'item_id'
name	str | None	Name for the collection. Defaults to directory name.	None

Returns:

Type	Description
AnalyticsCollection	Collection of analytics from all sessions.

Examples:

>>> collection = extract_from_directory("data/jatos_export/")
>>> print(f"Extracted {len(collection)} analytics records")

analyze_sessions(sessions: list[SlopitSession], analyzers: list[Analyzer] | None = None) -> list[SlopitSession]

Run slopit behavioral analyzers on sessions.

Uses slopit's AnalysisPipeline to process sessions with the specified analyzers.

Parameters:

Name	Type	Description	Default
sessions	list[SlopitSession]	Sessions to analyze.	required
analyzers	list[Analyzer] | None	Analyzers to run. If None, uses default set: KeystrokeAnalyzer, FocusAnalyzer, PasteAnalyzer, TimingAnalyzer.	None

Returns:

Type	Description
list[SlopitSession]	Sessions with analysis flags added.

Examples:

>>> from slopit import load_sessions
>>> sessions = load_sessions("data/")
>>> analyzed = analyze_sessions(sessions)
>>> # Sessions now have analysis flags populated

extract_with_analysis(path: Path | str, pattern: str = '*', item_id_key: str = 'item_id', analyzers: list[Analyzer] | None = None, name: str | None = None) -> AnalyticsCollection

Load sessions, run analysis, and extract analytics in one step.

Convenience function that combines loading, analysis, and extraction.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to session file or directory.	required
pattern	str	Glob pattern for directory (default: "*").	'*'
item_id_key	str	Key in platform_data containing the item UUID.	'item_id'
analyzers	list[Analyzer] | None	Analyzers to run. If None, uses default set.	None
name	str | None	Name for the collection.	None

Returns:

Type	Description
AnalyticsCollection	Collection with analyzed behavioral data.

Examples:

>>> collection = extract_with_analysis("data/jatos_export/")
>>> summaries = collection.get_participant_summaries()
>>> for s in summaries:
...     if s.flag_rate > 0.1:
...         print(f"Participant {s.participant_id}: {s.flag_rate:.1%} flagged")

Merging

merging

Utilities for merging behavioral analytics with judgment data.

This module provides functions for joining behavioral analytics with judgment DataFrames for analysis. All functions support both pandas and polars DataFrames, preserving the input type.

merge_behavioral_analytics(judgments_df: DataFrame, analytics: AnalyticsCollection, item_id_column: str = 'item_id', participant_id_column: str = 'participant_id', include_metrics: bool = True, include_flags: bool = True, how: str = 'left') -> DataFrame

Merge behavioral analytics into a judgments DataFrame.

Preserves input DataFrame type (pandas in -> pandas out, polars in -> polars out).

Parameters:

Name	Type	Description	Default
judgments_df	DataFrame	DataFrame containing judgment data.	required
analytics	AnalyticsCollection	Collection of behavioral analytics.	required
item_id_column	str	Column in judgments_df containing item IDs (default: "item_id").	'item_id'
participant_id_column	str	Column in judgments_df containing participant IDs.	'participant_id'
include_metrics	bool	If True, include flattened behavioral metrics columns.	True
include_flags	bool	If True, include flag-related columns.	True
how	str	Merge type: "left", "inner", "outer" (default: "left").	'left'

Returns:

Type	Description
DataFrame	Merged DataFrame with behavioral analytics columns added.

Examples:

>>> import pandas as pd
>>> judgments = pd.DataFrame({
...     "item_id": ["uuid1", "uuid2"],
...     "participant_id": ["p1", "p1"],
...     "response": [5, 3],
... })
>>> # merged = merge_behavioral_analytics(judgments, analytics_collection)

filter_flagged_judgments(judgments_df: DataFrame, analytics: AnalyticsCollection, item_id_column: str = 'item_id', participant_id_column: str = 'participant_id', min_severity: Severity | None = None, exclude_flagged: bool = True) -> DataFrame

Filter judgments based on behavioral flags.

Preserves input DataFrame type.

Parameters:

Name	Type	Description	Default
judgments_df	DataFrame	DataFrame containing judgment data.	required
analytics	AnalyticsCollection	Collection of behavioral analytics.	required
item_id_column	str	Column containing item IDs.	'item_id'
participant_id_column	str	Column containing participant IDs.	'participant_id'
min_severity	Severity | None	Minimum severity level for filtering. If None, any flag counts.	None
exclude_flagged	bool	If True, exclude flagged judgments (default). If False, keep only flagged judgments.	True

Returns:

Type	Description
DataFrame	Filtered DataFrame.

Examples:

>>> # Keep only unflagged judgments
>>> clean_df = filter_flagged_judgments(judgments, analytics, exclude_flagged=True)
>>> # Keep only high-severity flagged judgments for review
>>> flagged_df = filter_flagged_judgments(
...     judgments, analytics, min_severity="high", exclude_flagged=False
... )

create_analysis_dataframe_with_behavior(judgments_df: DataFrame, participants: ParticipantCollection, analytics: AnalyticsCollection, id_mappings: IDMappingCollection | None = None, external_id_column: str | None = None, participant_id_column: str = 'participant_id', item_id_column: str = 'item_id', metadata_columns: list[str] | None = None, include_metrics: bool = True, include_flags: bool = True) -> DataFrame

Create analysis-ready DataFrame with metadata and behavioral analytics.

Combines both participant and behavioral merging in one step. Preserves input DataFrame type.

Parameters:

Name	Type	Description	Default
judgments_df	DataFrame	Raw judgment data.	required
participants	ParticipantCollection	Participant collection with metadata.	required
analytics	AnalyticsCollection	Behavioral analytics collection.	required
id_mappings	IDMappingCollection | None	ID mappings (required if external_id_column is provided).	None
external_id_column	str | None	Column with external IDs to resolve.	None
participant_id_column	str	Column with participant IDs (after resolution).	'participant_id'
item_id_column	str	Column with item IDs.	'item_id'
metadata_columns	list[str] | None	Participant metadata columns to include.	None
include_metrics	bool	If True, include behavioral metrics columns.	True
include_flags	bool	If True, include flag columns.	True

Returns:

Type	Description
DataFrame	Analysis-ready DataFrame with both metadata and behavioral data.

Examples:

>>> analysis_df = create_analysis_dataframe_with_behavior(
...     judgments,
...     participants,
...     analytics,
...     id_mappings=mappings,
...     external_id_column="PROLIFIC_PID",
... )

get_exclusion_list(analytics: AnalyticsCollection, min_flag_rate: float = 0.1, min_severity: Severity | None = None) -> list[str]

Get list of participant IDs that should be excluded based on flags.

Identifies participants with flag rates above the threshold.

Parameters:

Name	Type	Description	Default
analytics	AnalyticsCollection	Behavioral analytics collection.	required
min_flag_rate	float	Minimum proportion of flagged judgments for exclusion (default: 0.1).	0.1
min_severity	Severity | None	Only count flags at or above this severity.	None

Returns:

Type	Description
list[str]	Participant IDs recommended for exclusion.

Examples:

>>> exclude = get_exclusion_list(analytics, min_flag_rate=0.2)
>>> clean_df = judgments_df[~judgments_df["participant_id"].isin(exclude)]