Spaces:
Running
Running
| import logging | |
| from datetime import datetime, timezone | |
| import pandas as pd | |
| from huggingface_hub import HfApi | |
| from about import TOKEN, REGISTRATION_REPO | |
| from components.registration.config import PARTICIPANTS_FILE_NAME, PARTICIPANT_COLUMNS | |
| from components.registration.utils import get_user_teams | |
| from components.submission.config import SUBMISSION_COLUMNS, SUBMISSIONS_FILE_NAME | |
| from components.utils import load_data_from_dataset, push_data_to_dataset | |
| logger = logging.getLogger(__name__) | |
| def submit_prediction( | |
| team_name: str, | |
| hypothesis: str, | |
| file_path: str, | |
| uploader_username: str, | |
| note: str | None, | |
| link_to_methods: str | None, | |
| ) -> tuple[bool, str]: | |
| """Save a validated submission; return (success, message).""" | |
| try: | |
| # 1. Load existing submissions and generate a new submission_id | |
| submissions_df = load_data_from_dataset( | |
| REGISTRATION_REPO, SUBMISSIONS_FILE_NAME, SUBMISSION_COLUMNS | |
| ) | |
| submission_id = len(submissions_df) + 1 | |
| # 2. Upload the CSV file to Registrations/submissions/{team_name}_{submission_id}.csv | |
| remote_file_path = f"Registrations/submissions/{team_name}_{submission_id}.csv" | |
| api = HfApi(token=TOKEN) | |
| with open(file_path, "rb") as f: | |
| api.upload_file( | |
| path_or_fileobj=f, | |
| path_in_repo=remote_file_path, | |
| repo_id=REGISTRATION_REPO, | |
| repo_type="dataset", | |
| commit_message=f"Submission {submission_id} from {team_name} – {datetime.now(timezone.utc).isoformat()}", | |
| ) | |
| # 3. Append the new record to submissions.csv | |
| new_record = pd.DataFrame( | |
| [ | |
| { | |
| "submission_id": submission_id, | |
| "team_name": team_name, | |
| "submission_username": uploader_username, | |
| "created_at": datetime.now(timezone.utc).isoformat(), | |
| "hypothesis": hypothesis, | |
| "submission_note": note or "", | |
| "method": link_to_methods or "", | |
| "file_path": remote_file_path, | |
| } | |
| ] | |
| ) | |
| updated_df = pd.concat([submissions_df, new_record], ignore_index=True) | |
| push_data_to_dataset(updated_df, REGISTRATION_REPO, SUBMISSIONS_FILE_NAME) | |
| return ( | |
| True, | |
| f"Submission #{submission_id} submitted for hypothesis {hypothesis}.", | |
| ) | |
| except Exception as exc: | |
| logger.error("Failed to submit prediction for %s: %s", team_name, exc) | |
| return False, f"Submission failed: {exc}" | |
| def get_team_submission_count(team_name: str, hypothesis: str) -> int: | |
| """Get the number of submissions for a team and a given hypothesis.""" | |
| submissions_df = load_data_from_dataset( | |
| REGISTRATION_REPO, SUBMISSIONS_FILE_NAME, SUBMISSION_COLUMNS | |
| ) | |
| return submissions_df[ | |
| (submissions_df["team_name"] == team_name) | |
| & (submissions_df["hypothesis"] == hypothesis) | |
| ].shape[0] | |
| def validate_user_team_registration(username: str, team_name: str) -> bool: | |
| """ | |
| Verify that the given username is registered as part of the team with the given team_name. | |
| :param username: The username to validate. | |
| :param team_name: The team to validate. | |
| :return: True is the username is registered as part of the team with the given team_name. False otherwise. | |
| """ | |
| user_teams = get_user_teams(username=username) | |
| return len(user_teams[user_teams["team_name"] == team_name]) > 0 | |
| def validate_user_registration(username: str, team_name: str) -> str: | |
| """ | |
| Verify that the given username is validated as part of the team with the given team_name. | |
| :param username: The username to validate. | |
| :param team_name: The team to validate. | |
| :return: True is the username is registered and validated as part of the team with the given team_name. False otherwise. | |
| """ | |
| participants_df = load_data_from_dataset( | |
| REGISTRATION_REPO, PARTICIPANTS_FILE_NAME, PARTICIPANT_COLUMNS | |
| ) | |
| if participants_df.empty: | |
| return "not registered" | |
| record = participants_df[ | |
| (participants_df["username"] == username) | |
| & (participants_df["team_name"] == team_name) | |
| ] | |
| if record.empty: | |
| return "not registered" | |
| elif record.iloc[0]["registration_status"] == "Validated": | |
| return "validated" | |
| elif record.iloc[0]["registration_status"] == "Pending": | |
| return "pending" | |
| else: | |
| return "not validated" | |