src/envs.py

import logging
import os

from huggingface_hub import HfApi

logger = logging.getLogger(__name__)

# clone / pull the lmeh eval data
H4_TOKEN = os.environ.get("H4_TOKEN", None)
GIT_TOKEN = os.environ.get("GIT_TOKEN", None)
GIT_REPO = os.environ.get("GIT_REPO", None)
HF_TOKEN = os.environ.get("HF_TOKEN", None)
lvkaokao_TOKEN = os.environ.get("lvkaokao_TOKEN", None)
INC4AI_TOKEN = os.environ.get("INC4AI_TOKEN", None)

# Azure DevOps PAT for CI pipeline dispatch (optional — dispatcher disabled without it)
AZP_DEVOPS_PAT = os.environ.get("AZP_DEVOPS_PAT", "")

# Automatic retry of failed CI runs (default: disabled).
# Set to "1" or "true" to enable the dispatcher's reconcile-and-retry logic.
ENABLE_RETRY = os.environ.get("ENABLE_RETRY", "").lower() in ("1", "true", "yes")

# Whitelist: comma-separated usernames. If empty, all authenticated users can submit.
_whitelist_raw = os.environ.get("WHITELIST", "").strip()
WHITELIST = set(filter(None, (x.strip() for x in _whitelist_raw.split(",")))) if _whitelist_raw else None

# Size-limit whitelist: comma-separated usernames.
# When non-empty, only Intel-org users whose username is in this list can skip the
# model-size cap.  When empty/unset, the original behaviour applies (all Intel-org
# users skip the cap).
_size_whitelist_raw = os.environ.get("SIZE_WHITELIST", "").strip()
SIZE_WHITELIST = set(filter(None, (x.strip() for x in _size_whitelist_raw.split(",")))) if _size_whitelist_raw else None


REPO_ID = "Intel/low-bit-leaderboard"
QUEUE_REPO = "Intel/ld_requests"
DYNAMIC_INFO_REPO = "Intel/dynamic_model_information"
RESULTS_REPO = "Intel/ld_results"

IS_PUBLIC = os.environ.get("IS_PUBLIC", "true").lower() == "true"

CACHE_PATH = os.getenv("HF_HOME", "./cache_hf")
CACHE_GIT = "cache_git"

EVAL_REQUESTS_PATH = os.path.join(CACHE_PATH, "eval-queue")
EVAL_RESULTS_PATH = os.path.join(CACHE_PATH, "eval-results")

# GIT_REQUESTS_PATH = os.path.join(CACHE_GIT, "requests")
GIT_REQUESTS_PATH = os.path.join(CACHE_GIT, "pending_requests")
GIT_STATUS_PATH = os.path.join(CACHE_GIT, "status")
GIT_RESULTS_PATH = os.path.join(CACHE_GIT, "results")

DYNAMIC_INFO_PATH = os.path.join(CACHE_PATH, "dynamic-info")
DYNAMIC_INFO_FILE_PATH = os.path.join(DYNAMIC_INFO_PATH, "model_infos.json")

EVAL_REQUESTS_PATH_PRIVATE = "eval-queue-private"
EVAL_RESULTS_PATH_PRIVATE = "eval-results-private"

PATH_TO_COLLECTION = os.environ.get("PATH_TO_COLLECTION", "")

# Rate limit variables — 5 submissions per user per day (across all queues combined)
RATE_LIMIT_PERIOD = 1
RATE_LIMIT_QUOTA = 5
HAS_HIGHER_RATE_LIMIT = ["TheBloke"]

API = HfApi(token=HF_TOKEN)

import threading
from git import Repo

GLOBAL_COND = threading.Condition()


def _init_git_repo():
    """Clone or pull the git evaluation repo. Called once at startup."""
    if not os.path.exists(CACHE_GIT):
        logger.info("Cloning git repo from %s", GIT_REPO)
        repo = Repo.clone_from(url=GIT_REPO, to_path=CACHE_GIT)
    else:
        logger.info("Loading git repo from local dir: %s", CACHE_GIT)
        repo = Repo.init(CACHE_GIT)
    # Ensure pull strategy is set (newer git requires this)
    with repo.config_writer() as cw:
        cw.set_value("pull", "rebase", "false")
    branch = repo.active_branch.name
    repo.remotes.origin.pull(branch)
    return repo


REPO = _init_git_repo()