evaluate.py

import hashlib
import re
from datetime import datetime, timezone
from pathlib import Path

import gradio as gr
import numpy as np
import pandas as pd
import pyarrow.parquet as pq
from huggingface_hub import CommitOperationAdd, HfApi

from config import (
    ACTIVATED_COL,
    CLINICAL_COLS,
    DATASET_REPO_ID,
    DEFAULT_LEADERBOARD_CONTEXT,
    FILE_MD5_COL,
    FINAL_LEADERBOARD_CONTEXT,
    FINAL_SUBMISSION_FILE_DIR,
    FUTURE_INSULIN_OPTIONS,
    HF_TOKEN,
    HORIZONS,
    METRIC_BASE_COLS,
    PRED_COLS,
    SUBMISSION_COOLDOWN_SECONDS,
    get_leaderboard_entry_dir,
    get_max_visible_submissions,
    get_leaderboard_name,
)
from data import (
    build_submission_metrics_row,
    get_reference_path,
    get_local_user_submission_history,
    get_user_metrics_repo_path,
    get_user_submission_history,
    write_user_metrics_history_file,
)
_MAX_SUBMISSION_NAME_LEN = 120
_MAX_CONTACT_EMAIL_LEN = 254
_MAX_WEBSITE_LEN = 200
_MAX_NOTES_LEN = 500
_JOIN_KEYS = ["id", "source_file", "date"]
_TARGET_COLS = [f"target_{horizon}" for horizon in HORIZONS]
_SUBMISSION_REQUIRED_COLS = _JOIN_KEYS + PRED_COLS
_GROUND_TRUTH_REQUIRED_COLS = _JOIN_KEYS + _TARGET_COLS
_EMPTY_TEXT_VALUE = "N/A"
_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
_PARQUET_BATCH_SIZE = 250_000
_MISSING_PREDICTION_STRINGS = {"", "n/a", "na", "nan", "none", "null"}
_VISIBILITY_REFRESH_NOTICE = "If the leaderboard does not reflect this change right away, please refresh the page."


def _emit_progress(progress_callback, percent: float, message: str) -> None:
    if progress_callback is not None:
        progress_callback(percent, message)


def validate_submission_name(name: str) -> str | None:
    """Returns an error message, or None if the value is acceptable."""
    name = (name or "").strip()
    if not name or name.lower() in _MISSING_PREDICTION_STRINGS:
        return "⚠️ Please provide a submission name."
    if len(name) > _MAX_SUBMISSION_NAME_LEN:
        return f"⚠️ Submission name must be {_MAX_SUBMISSION_NAME_LEN} characters or fewer."
    if re.search(r"[<>]", name):
        return "⚠️ Submission name contains invalid characters (< or >)."
    return None


def validate_contact_email(email: str) -> str | None:
    """Returns an error message, or None if the value is acceptable."""
    email = (email or "").strip()
    if not email or email.lower() in _MISSING_PREDICTION_STRINGS:
        return "⚠️ Please provide a contact email."
    if len(email) > _MAX_CONTACT_EMAIL_LEN:
        return f"⚠️ Contact email must be {_MAX_CONTACT_EMAIL_LEN} characters or fewer."
    if re.search(r"[<>\s]", email):
        return "⚠️ Contact email contains invalid characters or spaces."
    if not re.match(r"^[^@\s<>]+@[^@\s<>]+\.[^@\s<>]+$", email):
        return "⚠️ Please provide a valid contact email address."
    return None


def validate_website(url: str) -> str | None:
    """Returns an error message, or None if the value is acceptable."""
    url = (url or "").strip()
    if not url or url == "N/A":
        return None
    if len(url) > _MAX_WEBSITE_LEN:
        return f"⚠️ Website URL must be {_MAX_WEBSITE_LEN} characters or fewer."
    if re.search(r"""[<>"']""", url):
        return "⚠️ Website URL contains invalid characters (< > \" ')."
    if not re.match(r'^https?://', url, re.IGNORECASE):
        return "⚠️ Website URL must start with http:// or https://."
    return None


def validate_notes(notes: str) -> str | None:
    """Returns an error message, or None if the value is acceptable."""
    notes = (notes or "").strip()
    if not notes or notes == "N/A":
        return None
    if len(notes) > _MAX_NOTES_LEN:
        return f"⚠️ Notes must be {_MAX_NOTES_LEN} characters or fewer."
    if re.search(r'[<>]', notes):
        return "⚠️ Notes contain invalid characters (< or >)."
    return None


def _normalize_optional_text(value: str | None) -> str:
    return (value or "").strip() or _EMPTY_TEXT_VALUE


def _normalize_future_insulin_response(value: str | None) -> str:
    value = (value or "").strip()
    for option in FUTURE_INSULIN_OPTIONS:
        if value.lower() == option.lower():
            return option
    return value


def validate_future_insulin_response(value: str | None) -> str | None:
    value = _normalize_future_insulin_response(value)
    if value not in FUTURE_INSULIN_OPTIONS:
        return "⚠️ Please answer whether future insulin values were used."
    return None


def _validate_submission_details(contact_email: str, submission_name: str, website: str, notes: str) -> str | None:
    err = validate_contact_email(contact_email)
    if err:
        return err
    err = validate_submission_name(submission_name)
    if err:
        return err
    err = validate_website(website)
    if err:
        return err
    return validate_notes(notes)


def _coerce_uploaded_file_path(file_path):
    return getattr(file_path, "name", file_path)


def _open_parquet_file(file_path, label: str) -> pq.ParquetFile:
    try:
        return pq.ParquetFile(file_path)
    except Exception as exc:
        raise ValueError(f"Could not read {label} parquet file: {exc}") from exc


def _validate_parquet_columns(
    parquet_file: pq.ParquetFile,
    required_cols: list[str],
    label: str,
) -> str | None:
    available_cols = set(parquet_file.schema_arrow.names)
    missing_cols = [col for col in required_cols if col not in available_cols]
    if missing_cols:
        return f"⚠️ {label} is missing columns: {', '.join(missing_cols)}."
    return None


def _validate_submission_row_count(
    submission_parquet: pq.ParquetFile,
    reference_parquet: pq.ParquetFile,
) -> str | None:
    n_submitted = submission_parquet.metadata.num_rows
    n_reference = reference_parquet.metadata.num_rows
    if n_submitted != n_reference:
        return (
            f"⚠️ Submission has {n_submitted:,} rows but the reference file has {n_reference:,} rows. "
            "Ensure your file is derived from the predictions template without adding or removing rows."
        )
    return None


def _iter_aligned_batches(
    ground_truth_batches,
    submission_batches,
):
    ground_truth_batch = None
    submission_batch = None
    ground_truth_offset = 0
    submission_offset = 0

    while True:
        if ground_truth_batch is None or ground_truth_offset >= ground_truth_batch.num_rows:
            ground_truth_batch = next(ground_truth_batches, None)
            ground_truth_offset = 0
        if submission_batch is None or submission_offset >= submission_batch.num_rows:
            submission_batch = next(submission_batches, None)
            submission_offset = 0

        if ground_truth_batch is None and submission_batch is None:
            return
        if ground_truth_batch is None or submission_batch is None:
            raise ValueError("Reference and submission parquet streams ended at different row counts.")

        row_count = min(
            ground_truth_batch.num_rows - ground_truth_offset,
            submission_batch.num_rows - submission_offset,
        )
        yield (
            ground_truth_batch.slice(ground_truth_offset, row_count),
            submission_batch.slice(submission_offset, row_count),
        )
        ground_truth_offset += row_count
        submission_offset += row_count


def _validate_batch_keys(ground_truth_batch, submission_batch, row_offset: int) -> str | None:
    for col in _JOIN_KEYS:
        if not ground_truth_batch.column(col).equals(submission_batch.column(col)):
            return (
                f"⚠️ Submission keys do not match the template near row {row_offset + 1:,}. "
                "Check that the id, source_file, and date columns are unmodified and in the original order."
            )
    return None


def _empty_scores() -> dict[str, float]:
    return {f"{metric_name}_{horizon}": float("nan") for horizon in HORIZONS for metric_name in METRIC_BASE_COLS}


def _empty_score_accumulators():
    return {
        horizon: {
            "count": 0,
            "squared_error_sum": 0.0,
            "absolute_error_sum": 0.0,
            "absolute_relative_error_sum": 0.0,
            "relative_error_count": 0,
            "zone_counts": {metric_name: 0 for metric_name in CLINICAL_COLS},
        }
        for horizon in HORIZONS
    }


def _as_float_numpy(batch, column_name: str):
    return batch.column(column_name).to_numpy(zero_copy_only=False).astype(float, copy=False)


def _prediction_values_and_missing(batch, column_name: str, row_offset: int):
    series = batch.column(column_name).to_pandas()
    missing = series.isna()

    if not pd.api.types.is_numeric_dtype(series):
        stripped = series.astype("string").str.strip().str.lower()
        missing = missing | stripped.isin(_MISSING_PREDICTION_STRINGS).fillna(False)

    numeric = pd.to_numeric(series.mask(missing), errors="coerce")
    invalid = (~missing) & numeric.isna()
    if invalid.any():
        row_number = row_offset + int(np.flatnonzero(invalid.to_numpy(dtype=bool))[0]) + 1
        return None, None, (
            f"⚠️ Column '{column_name}' contains a non-numeric prediction near row {row_number:,}. "
            "Use numeric glucose predictions or leave the entire horizon column N/A."
        )

    values = numeric.to_numpy(dtype=float, na_value=np.nan)
    missing_mask = missing.to_numpy(dtype=bool)
    non_finite = (~missing_mask) & ~np.isfinite(values)
    if np.any(non_finite):
        row_number = row_offset + int(np.flatnonzero(non_finite)[0]) + 1
        return None, None, (
            f"⚠️ Column '{column_name}' contains a non-finite prediction near row {row_number:,}. "
            "Use finite numeric glucose predictions or leave the entire horizon column N/A."
        )

    return values, missing_mask, None


def _empty_horizon_availability():
    return {horizon: {"available": 0, "missing": 0} for horizon in HORIZONS}


def _record_horizon_availability(availability, horizon: int, available_count: int, missing_count: int) -> str | None:
    state = availability[horizon]
    state["available"] += int(available_count)
    state["missing"] += int(missing_count)

    if state["available"] > 0 and state["missing"] > 0:
        return (
            f"⚠️ Column 'pred_{horizon}' is partially filled. "
            "For each prediction horizon, submit predictions for every row or leave the entire column N/A."
        )
    return None


def _validate_horizon_availability(availability) -> str | None:
    submitted_horizons = [horizon for horizon, state in availability.items() if state["available"] > 0]
    if not submitted_horizons:
        return (
            "⚠️ At least one prediction horizon must be fully populated. "
            "The submitted file leaves every prediction column N/A."
        )
    return None


def _calculate_dts_zone_counts(pred_glucose, true_glucose) -> dict[str, int]:
    pred = np.asarray(pred_glucose, dtype=float)
    ref = np.asarray(true_glucose, dtype=float)

    if len(ref) == 0:
        return {metric_name: 0 for metric_name in CLINICAL_COLS}

    b_up = np.where(ref <= 50, 60, (540 / 450) * (ref - 50) + 60)
    c_up = np.where(ref <= 50, 86.5, (513.5 / 297) * (ref - 50) + 86.5)
    d_up = np.where(ref <= 50, 124, (476 / 191) * (ref - 50) + 124)
    e_up = np.where(ref <= 50, 179, (421 / 117) * (ref - 50) + 179)

    b_low = np.where(ref <= 62.5, 0, (430 / 537.5) * (ref - 62.5) + 50)
    c_low = np.where(ref <= 97.5, 0, (257 / 502.5) * (ref - 97.5) + 50)
    d_low = np.where(ref <= 153, 0, (147 / 447) * (ref - 153) + 50)
    e_low = np.where(ref <= 238, 0, (76 / 362) * (ref - 238) + 50)

    zone_a = (pred <= b_up) & (pred >= b_low)
    zone_b = ((pred <= c_up) & (pred > b_up)) | ((pred < b_low) & (pred >= c_low))
    zone_c = ((pred <= d_up) & (pred > c_up)) | ((pred < c_low) & (pred >= d_low))
    zone_d = ((pred <= e_up) & (pred > d_up)) | ((pred < d_low) & (pred >= e_low))
    zone_e = (pred > e_up) | (pred < e_low)

    return {
        "DTS_A_ZONE_PERCENT": int(np.count_nonzero(zone_a)),
        "DTS_B_ZONE_PERCENT": int(np.count_nonzero(zone_b)),
        "DTS_C_ZONE_PERCENT": int(np.count_nonzero(zone_c)),
        "DTS_D_ZONE_PERCENT": int(np.count_nonzero(zone_d)),
        "DTS_E_ZONE_PERCENT": int(np.count_nonzero(zone_e)),
    }


def _update_score_accumulators(accumulators, availability, ground_truth_batch, submission_batch, row_offset: int) -> str | None:
    for horizon in HORIZONS:
        pred_vals, pred_missing, pred_error = _prediction_values_and_missing(
            submission_batch,
            f"pred_{horizon}",
            row_offset,
        )
        if pred_error:
            return pred_error

        available_count = int(np.count_nonzero(~pred_missing))
        missing_count = int(np.count_nonzero(pred_missing))
        availability_error = _record_horizon_availability(
            availability,
            horizon,
            available_count,
            missing_count,
        )
        if availability_error:
            return availability_error
        if available_count == 0:
            continue

        true_vals = _as_float_numpy(ground_truth_batch, f"target_{horizon}")
        valid_mask = ~(np.isnan(true_vals) | pred_missing)
        if not np.any(valid_mask):
            continue

        true_vals = true_vals[valid_mask]
        pred_vals = pred_vals[valid_mask]
        errors = pred_vals - true_vals
        accumulator = accumulators[horizon]
        accumulator["count"] += len(true_vals)
        accumulator["squared_error_sum"] += float(np.dot(errors, errors))
        accumulator["absolute_error_sum"] += float(np.sum(np.abs(errors)))

        relative_error_mask = true_vals != 0
        if np.any(relative_error_mask):
            accumulator["absolute_relative_error_sum"] += float(
                np.sum(np.abs(errors[relative_error_mask] / true_vals[relative_error_mask]))
            )
            accumulator["relative_error_count"] += int(np.count_nonzero(relative_error_mask))

        for metric_name, count in _calculate_dts_zone_counts(pred_vals, true_vals).items():
            accumulator["zone_counts"][metric_name] += count
    return None


def _finalize_scores(accumulators) -> dict[str, float]:
    scores: dict[str, float] = {}
    for horizon in HORIZONS:
        accumulator = accumulators[horizon]
        count = accumulator["count"]
        if count == 0:
            scores.update({f"{metric_name}_{horizon}": float("nan") for metric_name in METRIC_BASE_COLS})
            continue

        for metric_name, zone_count in accumulator["zone_counts"].items():
            scores[f"{metric_name}_{horizon}"] = round((zone_count / count) * 100, 2)
        relative_error_count = accumulator["relative_error_count"]
        scores[f"MARD_{horizon}"] = (
            round((accumulator["absolute_relative_error_sum"] / relative_error_count) * 100, 2)
            if relative_error_count
            else float("nan")
        )
        scores[f"RMSE_{horizon}"] = round(
            float(np.sqrt(accumulator["squared_error_sum"] / count)),
            2,
        )
        scores[f"MAE_{horizon}"] = round(accumulator["absolute_error_sum"] / count, 2)
    return scores


def _validate_submission_format_streaming(
    submission_parquet: pq.ParquetFile,
    template_parquet: pq.ParquetFile,
    progress_callback=None,
) -> str | None:
    availability = _empty_horizon_availability()
    row_offset = 0
    total_rows = max(template_parquet.metadata.num_rows, 1)
    _emit_progress(progress_callback, 18, f"Validating predictions: 0/{total_rows:,} rows")

    template_batches = template_parquet.iter_batches(
        batch_size=_PARQUET_BATCH_SIZE,
        columns=_JOIN_KEYS,
    )
    submission_batches = submission_parquet.iter_batches(
        batch_size=_PARQUET_BATCH_SIZE,
        columns=_SUBMISSION_REQUIRED_COLS,
    )

    for template_batch, submission_batch in _iter_aligned_batches(template_batches, submission_batches):
        key_error = _validate_batch_keys(template_batch, submission_batch, row_offset)
        if key_error:
            return key_error

        for horizon in HORIZONS:
            _, pred_missing, pred_error = _prediction_values_and_missing(
                submission_batch,
                f"pred_{horizon}",
                row_offset,
            )
            if pred_error:
                return pred_error

            availability_error = _record_horizon_availability(
                availability,
                horizon,
                int(np.count_nonzero(~pred_missing)),
                int(np.count_nonzero(pred_missing)),
            )
            if availability_error:
                return availability_error

        row_offset += template_batch.num_rows
        validation_fraction = min(row_offset / total_rows, 1.0)
        _emit_progress(
            progress_callback,
            18 + validation_fraction * 57,
            f"Validating predictions: {min(row_offset, total_rows):,}/{total_rows:,} rows",
        )

    availability_error = _validate_horizon_availability(availability)
    if availability_error:
        return availability_error

    _emit_progress(progress_callback, 78, "Finalizing validation")
    return None


def _compute_scores_streaming(
    submission_parquet: pq.ParquetFile,
    ground_truth_parquet: pq.ParquetFile,
    progress_callback=None,
):
    accumulators = _empty_score_accumulators()
    availability = _empty_horizon_availability()
    row_offset = 0
    total_rows = max(ground_truth_parquet.metadata.num_rows, 1)
    _emit_progress(progress_callback, 18, f"Scoring predictions: 0/{total_rows:,} rows")

    ground_truth_batches = ground_truth_parquet.iter_batches(
        batch_size=_PARQUET_BATCH_SIZE,
        columns=_GROUND_TRUTH_REQUIRED_COLS,
    )
    submission_batches = submission_parquet.iter_batches(
        batch_size=_PARQUET_BATCH_SIZE,
        columns=_SUBMISSION_REQUIRED_COLS,
    )

    for ground_truth_batch, submission_batch in _iter_aligned_batches(ground_truth_batches, submission_batches):
        key_error = _validate_batch_keys(ground_truth_batch, submission_batch, row_offset)
        if key_error:
            return None, key_error
        score_error = _update_score_accumulators(
            accumulators,
            availability,
            ground_truth_batch,
            submission_batch,
            row_offset,
        )
        if score_error:
            return None, score_error
        row_offset += ground_truth_batch.num_rows
        scoring_fraction = min(row_offset / total_rows, 1.0)
        _emit_progress(
            progress_callback,
            18 + scoring_fraction * 57,
            f"Scoring predictions: {min(row_offset, total_rows):,}/{total_rows:,} rows",
        )

    availability_error = _validate_horizon_availability(availability)
    if availability_error:
        return None, availability_error

    _emit_progress(progress_callback, 78, "Finalizing score table")
    return _finalize_scores(accumulators), None


def _open_validated_submission_and_reference(
    file_path,
    context: str,
    reference_required_cols: list[str],
    progress_callback=None,
    reference_label: str = "reference",
):
    _emit_progress(progress_callback, 8, "Opening submission file")
    submission_parquet = _open_parquet_file(file_path, "submission")
    column_error = _validate_parquet_columns(submission_parquet, _SUBMISSION_REQUIRED_COLS, "Submission")
    if column_error:
        return None, None, column_error

    _emit_progress(progress_callback, 12, f"Opening {reference_label} file")
    reference_parquet = _open_parquet_file(get_reference_path(context), reference_label)
    reference_column_error = _validate_parquet_columns(
        reference_parquet,
        reference_required_cols,
        reference_label.capitalize(),
    )
    if reference_column_error:
        return None, None, reference_column_error

    row_count_error = _validate_submission_row_count(submission_parquet, reference_parquet)
    if row_count_error:
        return None, None, row_count_error

    return submission_parquet, reference_parquet, None


def _score_submission_file(file_path, context: str, progress_callback=None):
    submission_parquet, ground_truth_parquet, validation_error = _open_validated_submission_and_reference(
        file_path,
        context,
        _GROUND_TRUTH_REQUIRED_COLS,
        progress_callback,
        reference_label="ground truth",
    )
    if validation_error:
        return None, validation_error

    return _compute_scores_streaming(submission_parquet, ground_truth_parquet, progress_callback)


def _validate_annual_submission_file(file_path, context: str, progress_callback=None) -> str | None:
    submission_parquet, template_parquet, validation_error = _open_validated_submission_and_reference(
        file_path,
        context,
        _JOIN_KEYS,
        progress_callback,
        reference_label="template",
    )
    if validation_error:
        return validation_error

    return _validate_submission_format_streaming(submission_parquet, template_parquet, progress_callback)


def _calculate_file_md5(file_path) -> str:
    md5 = hashlib.md5()
    with open(file_path, "rb") as submitted_file:
        for chunk in iter(lambda: submitted_file.read(1024 * 1024), b""):
            md5.update(chunk)
    return md5.hexdigest()


def _sanitize_repo_filename(value: str) -> str:
    safe_value = re.sub(r"[^A-Za-z0-9_.-]+", "_", str(value).strip()).strip("._")
    return safe_value or "user"


def _annual_submission_repo_path(username: str, file_md5: str) -> str:
    safe_username = _sanitize_repo_filename(username)
    return f"{FINAL_SUBMISSION_FILE_DIR}/{safe_username}__{file_md5}.parquet"


def _get_remaining_cooldown_seconds(metrics_history_df: pd.DataFrame) -> int | None:
    if metrics_history_df.empty:
        return None

    last_ts = metrics_history_df.iloc[0]["Timestamp"]
    if last_ts in (_EMPTY_TEXT_VALUE, "", "nan"):
        return None

    try:
        last_dt = datetime.strptime(last_ts, _TIMESTAMP_FORMAT).replace(tzinfo=timezone.utc)
    except ValueError:
        return None

    elapsed = (datetime.now(timezone.utc) - last_dt).total_seconds()
    remaining = int(SUBMISSION_COOLDOWN_SECONDS - elapsed)
    return remaining if remaining > 0 else None


def _build_submission_summary(
    username: str,
    action: str,
    scores: dict[str, float],
    context: str,
    file_md5: str,
) -> str:
    if context == FINAL_LEADERBOARD_CONTEXT:
        return (
            f"✅ Submission successful. {action} {username}'s annual competition submission. "
            f"File MD5: {file_md5}. The submitted file has been stored for final evaluation. "
            "Metrics are not shown during the competition and will be released after the competition concludes."
        )

    summary_lines = []
    for horizon in HORIZONS:
        dts_a = scores[f"DTS_A_ZONE_PERCENT_{horizon}"]
        mard = scores[f"MARD_{horizon}"]
        rmse = scores[f"RMSE_{horizon}"]
        if pd.isna(dts_a) or pd.isna(mard) or pd.isna(rmse):
            summary_lines.append(f"  {horizon}min → not included")
        else:
            summary_lines.append(f"  {horizon}min → DTS-A: {dts_a:.1f}%, MARD: {mard:.1f}%, RMSE: {rmse:.1f}")
    return f"✅ Submission successful. {action} {username}'s scores:\n" + "\n".join(summary_lines)


def _append_submission_to_history(
    metrics_history_df: pd.DataFrame,
    scores: dict[str, float],
    timestamp: str,
    used_future_insulin_values: str,
    contact_email: str,
    submission_name: str,
    website: str,
    notes: str,
    file_md5: str = "N/A",
):
    updated_history_df = metrics_history_df.copy()
    new_row_df = build_submission_metrics_row(
        scores,
        timestamp,
        contact_email,
        submission_name,
        website,
        notes,
        used_future_insulin_values=used_future_insulin_values,
        activated=True,
        file_md5=file_md5,
    )
    return pd.concat([new_row_df, updated_history_df], ignore_index=True)


def _commit_user_metrics_history(
    username: str,
    metrics_history_df: pd.DataFrame,
    commit_message: str,
    context: str = DEFAULT_LEADERBOARD_CONTEXT,
):
    metrics_path = write_user_metrics_history_file(
        output_dir=Path(get_leaderboard_entry_dir(context)),
        username=username,
        metrics_history_df=metrics_history_df,
    )

    api = HfApi()
    api.create_commit(
        repo_id=DATASET_REPO_ID,
        repo_type="dataset",
        operations=[
            CommitOperationAdd(
                path_in_repo=get_user_metrics_repo_path(username, context),
                path_or_fileobj=str(metrics_path),
            ),
        ],
        commit_message=commit_message,
        token=HF_TOKEN,
    )


def _normalize_selected_timestamps(timestamps) -> list[str]:
    if timestamps is None:
        return []
    if isinstance(timestamps, str):
        timestamps = [timestamps]
    normalized = []
    for timestamp in timestamps:
        timestamp = (timestamp or "").strip()
        if timestamp and timestamp not in normalized:
            normalized.append(timestamp)
    return normalized


def set_active_submissions(
    profile: gr.OAuthProfile | None,
    timestamps,
    context: str = DEFAULT_LEADERBOARD_CONTEXT,
):
    """Marks saved submissions as selected, or clears the selection."""
    if profile is None:
        return "⚠️ Please log in with Hugging Face to manage submissions."

    selected_timestamps = _normalize_selected_timestamps(timestamps)
    max_selected = get_max_visible_submissions(context)
    if len(selected_timestamps) > max_selected:
        return f"⚠️ Please select at most {max_selected} submissions."

    username = profile.username
    leaderboard_name = get_leaderboard_name(context)

    try:
        metrics_history_df = get_user_submission_history(username, context)
        local_history_df = get_local_user_submission_history(username, context)
        if selected_timestamps and not local_history_df.empty:
            selected_timestamp_set = set(selected_timestamps)
            remote_timestamps = set(metrics_history_df["Timestamp"]) if not metrics_history_df.empty else set()
            local_timestamps = set(local_history_df["Timestamp"])
            if selected_timestamp_set.issubset(local_timestamps) and not selected_timestamp_set.issubset(remote_timestamps):
                metrics_history_df = local_history_df
        if metrics_history_df.empty:
            return "⚠️ No saved submissions found."

        metrics_history_df = metrics_history_df.copy()
        metrics_history_df[ACTIVATED_COL] = False

        missing_timestamps = []
        for timestamp in selected_timestamps:
            matching_rows = metrics_history_df["Timestamp"] == timestamp
            if not matching_rows.any():
                missing_timestamps.append(timestamp)
                continue
            first_match_idx = metrics_history_df.index[matching_rows][0]
            metrics_history_df.loc[first_match_idx, ACTIVATED_COL] = True

        if missing_timestamps:
            return "⚠️ One or more selected submissions could not be found."

        if context == FINAL_LEADERBOARD_CONTEXT:
            if selected_timestamps:
                count = len(selected_timestamps)
                noun = "submission" if count == 1 else "submissions"
                status_message = (
                    f"✅ Saved {count} selected annual competition {noun} for final evaluation. "
                    "Please refresh the page if this selection does not update right away."
                )
                commit_message = f"Update selected {leaderboard_name} entries for {username}"
            else:
                status_message = (
                    "✅ No saved annual competition submissions are selected for final evaluation. "
                    "Please refresh the page if this selection does not update right away."
                )
                commit_message = f"Clear selected {leaderboard_name} entries for {username}"
        elif selected_timestamps:
            count = len(selected_timestamps)
            noun = "submission" if count == 1 else "submissions"
            status_message = f"✅ Showing {count} selected {noun} on {leaderboard_name}. {_VISIBILITY_REFRESH_NOTICE}"
            commit_message = f"Update visible {leaderboard_name} entries for {username}"
        else:
            status_message = f"✅ Your submissions are now hidden from {leaderboard_name}. {_VISIBILITY_REFRESH_NOTICE}"
            commit_message = f"Hide {leaderboard_name} entries for {username}"

        _commit_user_metrics_history(username, metrics_history_df, commit_message, context)
        return status_message

    except Exception as e:
        return f"❌ An error occurred: {str(e)}"


def set_active_submission(
    profile: gr.OAuthProfile | None,
    timestamp: str | None,
    context: str = DEFAULT_LEADERBOARD_CONTEXT,
):
    return set_active_submissions(profile, [timestamp] if timestamp else [], context)


def evaluate_and_submit(
    profile: gr.OAuthProfile | None,
    file_path,
    used_future_insulin_values: str | None,
    contact_email: str,
    submission_name: str,
    website: str = "N/A",
    notes: str = "N/A",
    context: str = DEFAULT_LEADERBOARD_CONTEXT,
    progress_callback=None,
):
    """Validates a submission and securely pushes its saved record to the database."""
    _emit_progress(progress_callback, 2, "Checking submission")
    if profile is None:
        return "⚠️ Please log in with Hugging Face to submit your results."

    username = profile.username

    if not file_path:
        return "⚠️ Please provide a predictions file."

    contact_email = (contact_email or "").strip()
    submission_name = (submission_name or "").strip()
    website = _normalize_optional_text(website)
    notes = _normalize_optional_text(notes)
    used_future_insulin_values = _normalize_future_insulin_response(used_future_insulin_values)

    future_insulin_error = validate_future_insulin_response(used_future_insulin_values)
    if future_insulin_error:
        return future_insulin_error

    details_error = _validate_submission_details(contact_email, submission_name, website, notes)
    if details_error:
        return details_error

    try:
        submitted_file_path = _coerce_uploaded_file_path(file_path)
        file_md5 = _calculate_file_md5(submitted_file_path)
        if context == FINAL_LEADERBOARD_CONTEXT:
            score_error = _validate_annual_submission_file(submitted_file_path, context, progress_callback)
            if score_error:
                return score_error
            scores = _empty_scores()
            _emit_progress(progress_callback, 80, "Preparing submission file storage")
        else:
            scores, score_error = _score_submission_file(submitted_file_path, context, progress_callback)
            if score_error:
                return score_error
        _emit_progress(progress_callback, 80, "Loading submission history")
        metrics_history_df = get_user_submission_history(username, context)

        remaining_cooldown = _get_remaining_cooldown_seconds(metrics_history_df)
        if remaining_cooldown is not None:
            return f"⚠️ Please wait {remaining_cooldown}s before submitting again."

        action = "Updated" if not metrics_history_df.empty else "Added"
        timestamp = datetime.now(timezone.utc).strftime(_TIMESTAMP_FORMAT)
        _emit_progress(progress_callback, 84, "Preparing leaderboard files")
        updated_metrics_history_df = _append_submission_to_history(
            metrics_history_df,
            scores,
            timestamp,
            used_future_insulin_values,
            contact_email,
            submission_name,
            website,
            notes,
            file_md5=file_md5,
        )
        metrics_path = write_user_metrics_history_file(
            output_dir=Path(get_leaderboard_entry_dir(context)),
            username=username,
            metrics_history_df=updated_metrics_history_df,
        )

        commit_progress_message = (
            "Saving annual competition record and submission file"
            if context == FINAL_LEADERBOARD_CONTEXT
            else "Uploading leaderboard update"
        )
        _emit_progress(progress_callback, 90, commit_progress_message)
        operations = [
            CommitOperationAdd(
                path_in_repo=get_user_metrics_repo_path(username, context),
                path_or_fileobj=str(metrics_path),
            ),
        ]
        if context == FINAL_LEADERBOARD_CONTEXT:
            operations.append(
                CommitOperationAdd(
                    path_in_repo=_annual_submission_repo_path(username, file_md5),
                    path_or_fileobj=str(submitted_file_path),
                )
            )

        api = HfApi()
        api.create_commit(
            repo_id=DATASET_REPO_ID,
            repo_type="dataset",
            operations=operations,
            commit_message=f"Update {get_leaderboard_name(context)} entry for {username}",
            token=HF_TOKEN,
        )

        completion_message = (
            "Submission complete. Your annual competition file and file fingerprint have been saved."
            if context == FINAL_LEADERBOARD_CONTEXT
            else "Submission complete. The results will appear shortly."
        )
        _emit_progress(progress_callback, 100, completion_message)
        return _build_submission_summary(username, action, scores, context, file_md5)

    except Exception as e:
        return f"❌ An error occurred: {str(e)}"