Extract the valid listen event definition enforcement into a dedicated module

scripts/plots.py

@@ -16,7 +16,7 @@
 from impact.cli import inject_impact_config
 from impact.config import ImpactConfig
 from impact.logger import configure_logging
-from impact.pipelines import DataPipeline, IMPACTEvaluationPipeline
+from impact.pipelines import DataPipeline, ImpactEvaluationPipeline
 from impact.utils import seed_everything
 
 logger = structlog.stdlib.get_logger(__name__)
@@ -154,20 +154,10 @@ def cli() -> None:
 @click.option(
     "--lang",
     type=click.Choice(["pl", "en"], case_sensitive=False),
-    default="pl",
+    default="en",
     show_default=True,
     help="Language for axis labels and annotations.",
 )
-@click.option(
-    "--n-partitions",
-    type=int,
-    default=1,
-    show_default=True,
-    help="""
-        Number of partitions for sequential processing. Use >1 for large datasets that
-        exceed available RAM (data is partitioned by user, i.e. uid column).
-    """,
-)
 @click.option(
     "--raw-data-dir",
     type=click.Path(file_okay=False, path_type=Path),
@@ -189,7 +179,7 @@ def cli() -> None:
 @click.option(
     "--processed-data-dir",
     type=click.Path(file_okay=False, path_type=Path),
-    required=False,
+    default=None,
     help="""
         Base directory for processed data (the final, canonical datasets for modeling).
         Defaults to <project_root>/data/processed.
@@ -269,9 +259,8 @@ def cli() -> None:
 def plot_alpha_curve(
     impact_config: ImpactConfig,
     sasrec_checkpoint_path: Path,
-    log_level: str,
-    lang: str,
-    n_partitions: int,
+    log_level: str = "WARNING",
+    lang: str = "en",
     raw_data_dir: Path | None = None,
     interim_data_dir: Path | None = None,
     processed_data_dir: Path | None = None,
@@ -320,11 +309,11 @@ def plot_alpha_curve(
         config=impact_config,
     )
 
-    logger.info("Fetching raw events data...")
-    events_lf = data_pipeline.get_raw_events()
+    logger.info("Fetching listening events data...")
+    listens_lf = data_pipeline.get_listen_events()
 
     logger.info("Fetching train/val/test splits...")
-    splits, item_index = data_pipeline.get_splits(n_partitions=n_partitions)
+    splits, item_index = data_pipeline.get_splits()
 
     logger.info("Tensorizing splits...")
     tensorized_splits = data_pipeline.tensorize(splits)
@@ -346,7 +335,7 @@ def plot_alpha_curve(
         )
 
         logger.info(f"Running IMPACT evaluation for {alpha=}...")
-        eval_pipeline = IMPACTEvaluationPipeline(eval_config, item_index, events_lf)
+        eval_pipeline = ImpactEvaluationPipeline(eval_config, item_index, listens_lf)
         metrics: dict[str, float] = eval_pipeline.evaluate(
             split_partition=tensorized_splits.val,
             sasrec_state_dict_path=sasrec_checkpoint_path,
@@ -521,7 +510,7 @@ def _create_subplot(
 @click.option(
     "--lang",
     type=click.Choice(["pl", "en"], case_sensitive=False),
-    default="pl",
+    default="en",
     show_default=True,
     help="Language for axis labels and annotations.",
 )
@@ -539,9 +528,9 @@ def _create_subplot(
     "--format",
     "-f",
     "formats",
-    type=click.Choice(["svg", "pdf", "png"], case_sensitive=False),
+    type=click.Choice(("svg", "pdf", "png"), case_sensitive=False),
     multiple=True,
-    default=["svg", "pdf"],
+    default=("svg", "pdf"),
     show_default=True,
     help="Output format(s). Can be specified multiple times: -f svg -f pdf.",
 )
@@ -554,11 +543,11 @@ def _create_subplot(
 )
 def plot_sasrec_training(
     log_file_path: Path,
-    log_level: str,
-    lang: str,
-    out_dir: Path | None,
-    formats: list[Literal["svg", "pdf", "png"]],
-    png_scale: float,
+    log_level: str = "INFO",
+    lang: str = "en",
+    out_dir: Path | None = None,
+    formats: tuple[Literal["svg", "pdf", "png"], ...] = ("svg", "pdf"),
+    png_scale: float = 3.0,
 ) -> None:
     """Plot SASRec training curves from a structured NDJSON log file.
 

scripts/tables.py

@@ -37,14 +37,14 @@ def cli() -> None:
 @click.option(
     "--lang",
     type=click.Choice(["pl", "en"], case_sensitive=False),
-    default="pl",
+    default="en",
     help="Language for the table headers and textual content (pl or en).",
 )
 @click.option(
     "--out-dir",
     "-o",
     type=click.Path(file_okay=False, path_type=Path),
-    required=False,
+    default=None,
     help="""
         Output directory for the LaTeX code and corresponding PipelineConfig YAML dumps.
         Defaults to project root / 'reports' / 'tables' / 'data-funnel-stats'.
@@ -53,7 +53,7 @@ def cli() -> None:
 @click.option(
     "--raw-data-dir",
     type=click.Path(file_okay=False, path_type=Path),
-    required=False,
+    default=None,
     help="""
         Root directory for raw data (the original, immutable data dump).
         Defaults to project root / 'data' / 'raw'.
@@ -62,7 +62,7 @@ def cli() -> None:
 @click.option(
     "--interim-data-dir",
     type=click.Path(file_okay=False, path_type=Path),
-    required=False,
+    default=None,
     help="""
         Root directory for interim data (intermediate data that has been transformed).
         Defaults to project root / 'data' / 'interim'.
@@ -71,7 +71,7 @@ def cli() -> None:
 @click.option(
     "--processed-data-dir",
     type=click.Path(file_okay=False, path_type=Path),
-    required=False,
+    default=None,
     help="""
         Root directory for processed data (the final, canonical datasets for modeling).
         Defaults to project root / 'data' / 'processed'.
@@ -80,8 +80,8 @@ def cli() -> None:
 @inject_impact_config
 def generate_data_funnel_stats_table(
     impact_config: ImpactConfig,
-    log_level: str,
-    lang: str = "pl",
+    log_level: str = "INFO",
+    lang: str = "en",
     out_dir: Path | None = None,
     raw_data_dir: Path | None = None,
     interim_data_dir: Path | None = None,
@@ -134,20 +134,17 @@ def generate_data_funnel_stats_table(
 
     class _VariantConfig(NamedTuple):
         step: int
-        n_partitions: int
         dataset_name: str  # Name of the dataset for the "reporting" (in the thesis)
         marker: str  # Marker for the footnotes
 
     variants: dict[DatasetSize, _VariantConfig] = {
         DatasetSize.SIZE_50M: _VariantConfig(
             step=5,
-            n_partitions=1,
             dataset_name="Yambda-50M",
             marker="a",
         ),
         DatasetSize.SIZE_5B: _VariantConfig(
             step=20,
-            n_partitions=32,
             dataset_name="Yambda-5B",
             marker="b",
         ),
@@ -166,12 +163,13 @@ class _VariantConfig(NamedTuple):
         pipeline.config = variant_config  # Switch pipeline config
 
         events_lf = pipeline.get_raw_events()
-        sessions_lf = pipeline.get_sessions(n_partitions=variant.n_partitions)
-        sequences_lf = pipeline.get_sequences(n_partitions=variant.n_partitions)
+        sessions_lf = pipeline.get_sessions()
+        sequences_lf = pipeline.get_sequences()
 
         tnote = f"\\tnote{{\\hyperlink{{note:{variant.marker}}}{{{variant.marker}}}}}"
         rows.append({
             "dataset": f"{variant.dataset_name}{tnote}",
+            # TODO: report also the number of valid listening events
             "raw": _format_num(events_lf.select(pl.len()).collect().item()),
             "sessionized": _format_num(sessions_lf.select(pl.len()).collect().item()),
             "sequenced": _format_num(sequences_lf.select(pl.len()).collect().item()),

src/impact/cli.py

@@ -57,15 +57,6 @@ def inject_impact_config(f: Callable[..., Any]) -> Callable[..., Any]:
                 variants contain only a specific subset.
             """,
         ),
-        click.option(
-            "--session-inactivity-gap-seconds",
-            type=int,
-            default=1200,
-            show_default=True,
-            help="""
-                Number of seconds of inactivity before a session is considered finished.
-            """,
-        ),
         click.option(
             "--session-tracks-per-session",
             type=int,
@@ -77,17 +68,6 @@ def inject_impact_config(f: Callable[..., Any]) -> Callable[..., Any]:
                 K tracks are truncated.
             """,
         ),
-        click.option(
-            "--session-min-listen-seconds",
-            type=int,
-            default=30,
-            show_default=True,
-            help="""
-                Minimum playback duration (in seconds) to consider an event as a valid
-                listen. Defaults to 30 (industry standard used by authors of other
-                publications).
-            """,
-        ),
         click.option(
             "--sequence-context-sessions",
             type=int,

src/impact/config.py

@@ -39,22 +39,13 @@ def __post_init__(self) -> None:
 class SessionConfig:
     """Configuration for the sessionization stage (Algorithm 1 in the thesis)."""
 
-    inactivity_gap_seconds: int = 1200
-    """Number of seconds of inactivity before a session is considered finished."""
-
     tracks_per_session: int = 10
     """Number of tracks that constitute a session. Denoted as K in the thesis.
 
     Sessions with less than K tracks are discarded. Sessions with more than K tracks
     are truncated.
     """
 
-    min_listen_seconds: int = 30
-    """Minimum playback duration (in seconds) to consider an event as a valid listen.
-
-    Defaults to 30 (industry standard used by authors of other publications).
-    """
-
 
 @dataclass(frozen=True)
 class SequenceConfig:
@@ -241,9 +232,7 @@ def sessionization_id(self) -> str:
         """Cache key for the sessionization stage (Algorithm 1 in the thesis)."""
         parts = [
             "sessions",
-            f"gap{self.session.inactivity_gap_seconds}",
             f"k{self.session.tracks_per_session}",
-            f"min{self.session.min_listen_seconds}",
         ]
         return "_".join(parts)
 

src/impact/features/__init__.py

@@ -1,10 +1,12 @@
+from .preprocessing import filter_listen_events
 from .sequence_extraction import extract_sequences
 from .sessionization import sessionize
 from .temporal_split import split
 from .tensorization import tensorize
 
 __all__ = [
     "extract_sequences",
+    "filter_listen_events",
     "sessionize",
     "split",
     "tensorize",

src/impact/features/preprocessing.py

@@ -0,0 +1,54 @@
+"""Raw event preprocessing applied before any feature-engineering step."""
+
+import polars as pl
+import structlog
+
+logger = structlog.stdlib.get_logger(__name__)
+
+MIN_LISTEN_SECONDS: int = 30
+"""Minimum playback duration (in seconds) to consider an event as a valid listen.
+Below this threshold events are treated as noise (accidental plays, skips).
+Follows the industry standard established in the literature.
+"""
+
+
+def filter_listen_events(
+    events_lf: pl.LazyFrame,
+    min_listen_seconds: int = MIN_LISTEN_SECONDS,
+) -> pl.LazyFrame:
+    """Filter raw events of all types down to valid listening events only.
+
+    This is the only place that decides what constitutes a valid listen, so the
+    definition stays consistent throughout the entire codebase.
+
+    Args:
+        events_lf: Raw LazyFrame containing all event types.
+        min_listen_seconds: Minimum playback duration (in seconds) to consider an event
+            as a valid listen. Below this threshold events are treated as noise
+            (accidental plays, skips).
+
+    Returns:
+        A LazyFrame containing only `event_type == "listen"` rows where the user
+        actually played enough of the track, with an `actual_played_seconds` column
+        added.
+    """
+    logger.debug(
+        "Filtering raw events to valid listen events",
+        min_listen_seconds=min_listen_seconds,
+    )
+    return (
+        events_lf
+
+        # Filter out non-listen events
+        .filter(pl.col("event_type") == "listen")
+
+        # Calculate actual played time in seconds
+        .with_columns(
+            (pl.col("track_length_seconds") * pl.col("played_ratio_pct") / 100).alias(
+                "actual_played_seconds"
+            )
+        )
+
+        # Filter valid listening events
+        .filter(pl.col("actual_played_seconds") >= min_listen_seconds)
+    )  # fmt: skip