evaluator

def reset(self) -> None:
    """Reset the evaluator to initial state."""
    logger.info("Resetting the evaluator for a new run...")
    self._run_step = 0

def run_step(self) -> None:
    """Run a single step of the evaluator."""
    if self._run_step == 0:
        logger.info(f"There is a total of {self.setting.num_split} steps. Running step {self._run_step}")
        self._ready_evaluator()

    if self._run_step > self.setting.num_split:
        logger.info("Finished running all steps, call `run_step(reset=True)` to run the evaluation again")
        warn("Running this method again will not have any effect.")
        return
    logger.info("Running step %d", self._run_step)
    self._evaluate_step()
    self._data_release_step()

def run_steps(self, num_steps: int) -> None:
    """Run multiple steps of the evaluator.

    Effectively runs the run_step method num_steps times. Call
    this method to run multiple steps of the evaluator at once.

    Args:
        num_steps: Number of steps to run.

    Raises:
        ValueError: If cannot run the specified number of steps.
    """
    if self._run_step + num_steps > self.setting.num_split:
        raise ValueError(f"Cannot run {num_steps} steps, only {self.setting.num_split - self._run_step} steps left")
    for _ in tqdm(range(num_steps)):
        self.run_step()

def run(self) -> None:
    """Run the evaluator across all steps and splits.

    This method should be called when the programmer wants to step through
    all phases and splits to arrive to the metrics computed. An alternative
    to running through all splits is to call the run_step method which runs
    only one step at a time.
    """
    self._ready_evaluator()

    with tqdm(total=self.setting.num_split, desc="Evaluating steps") as pbar:
        while self._run_step <= self.setting.num_split:
            logger.info("Running step %d", self._run_step)
            self._evaluate_step()
            pbar.update(1)
            # if is last step, no need to release data anymore
            # since there is no more evaluation that can be done
            # break out of the loop
            if self._run_step == self.setting.num_split:
                break
            self._data_release_step()

def metric_results(
    self,
    level: MetricLevelEnum | Literal["macro", "micro", "window", "user"] = MetricLevelEnum.MACRO,
    only_current_timestamp: None | bool = False,
    filter_timestamp: None | int = None,
    filter_algo: None | str = None,
) -> pd.DataFrame:
    """Results of the metrics computed.

    Computes the metrics of all algorithms based on the level specified and
    return the results in a pandas DataFrame. The results can be filtered
    based on the algorithm name and the current timestamp.

    Specifics
    ---------
    - User level: User level metrics computed across all timestamps.
    - Window level: Window level metrics computed across all timestamps. This can
        be viewed as a macro level metric in the context of a single window, where
        the scores of each user is averaged within the window.
    - Macro level: Macro level metrics computed for entire timeline. This
        score is computed by averaging the scores of all windows, treating each
        window equally.
    - Micro level: Micro level metrics computed for entire timeline. This
        score is computed by averaging the scores of all users, treating each
        user and the timestamp the user is in as unique contribution to the
        overall score.

    Args:
        level: Level of the metric to compute, defaults to "macro".
        only_current_timestamp: Filter only the current timestamp, defaults to False.
        filter_timestamp: Timestamp value to filter on, defaults to None.
            If both `only_current_timestamp` and `filter_timestamp` are provided,
            `filter_timestamp` will be used.
        filter_algo: Algorithm name to filter on, defaults to None.

    Returns:
        Dataframe representation of the metric.
    """
    if isinstance(level, str) and not MetricLevelEnum.has_value(level):
        raise ValueError("Invalid level specified")
    level = MetricLevelEnum(level)

    if only_current_timestamp and filter_timestamp:
        raise ValueError("Cannot specify both only_current_timestamp and filter_timestamp.")

    timestamp = None
    if only_current_timestamp:
        timestamp = self._current_timestamp

    if filter_timestamp:
        timestamp = filter_timestamp

    return self._acc.df_metric(filter_algo=filter_algo, filter_timestamp=timestamp, level=level)

def plot_macro_level_metric(self) -> None:
    df = self.metric_results("macro")
    df = df.reset_index()
    ax = sns.barplot(
        data=df,
        x="metric",
        y="macro_score",
        hue="algorithm",
        edgecolor="black"
    )

    ax.set_xlabel("Metric")
    ax.set_ylabel("Macro score")
    ax.set_title("Macro-level scores by metric and algorithm")

    for container in ax.containers:
        ax.bar_label(container, fmt='%.4f', padding=3, fontsize=8)
    plt.legend(
        title="Algorithm",
        loc="upper center",
        bbox_to_anchor=(0.5, -0.1),
    )
    ax.grid(axis="y", alpha=0.3, linestyle="--")
    plt.show()

def plot_micro_level_metric(self) -> None:
    df = self.metric_results("micro")
    df = df.reset_index()
    ax = sns.barplot(
        data=df,
        x="metric",
        y="micro_score",
        hue="algorithm",
        edgecolor="black"
    )

    ax.set_xlabel("Metric")
    ax.set_ylabel("Micro score")
    ax.set_title("Micro-level scores by metric and algorithm")

    for container in ax.containers:
        ax.bar_label(container, fmt='%.4f', padding=3, fontsize=8)
    plt.legend(
        title="Algorithm",
        loc="upper center",
        bbox_to_anchor=(0.5, -0.1),
    )
    ax.grid(axis="y", alpha=0.3, linestyle="--")
    plt.show()

def plot_window_level_metric(self) -> None:
    df = self.metric_results("window")
    df = df.reset_index()
    metrics = df["metric"].unique()
    n_metrics = len(metrics)

    fig, axes = plt.subplots(n_metrics, 1, figsize=(10, 7), sharey=False)
    if n_metrics == 1:
        axes = [axes]

    fig.suptitle("Window-level scores over time", fontsize=14, fontweight="bold")

    for ax, metric in zip(axes, metrics):
        # Filter data for this metric
        metric_df = df[df["metric"] == metric]

        # Plot line for each algorithm
        sns.lineplot(
            data=metric_df,
            x="timestamp",
            y="window_score",
            hue="algorithm",
            marker="o",
            markersize=6,
            linewidth=2,
            ax=ax,
        )
        ax.set_xlabel("Timestamp (epoch)")
        ax.set_ylabel(f"{metric} score")
        ax.grid(axis="both", alpha=0.3, linestyle="--")

        # Remove individual legends
        if ax.get_legend() is not None:
            ax.get_legend().remove()

    # Create single shared legend at bottom
    handles, labels = axes[0].get_legend_handles_labels()

    fig.legend(
        handles,
        labels,
        title="Algorithm",
        loc="lower center",
        bbox_to_anchor=(0.5, -0.15),
        ncol=1,  # vertical stacking
        frameon=True,
        fontsize=9,
    )
    plt.show()

def restore(self) -> None:
    """Restore the generators before pickling.

    This method is used to restore the generators after loading the object
    from a pickle file.
    """
    self.setting.restore(self._run_step)
    logger.debug("Generators restored")

def current_step(self) -> int:
    """Return the current step of the evaluator.

    Returns:
        Current step of the evaluator.
    """
    return self._run_step