run_jobs.py

import json
import logging
import os
import subprocess
import threading
import time
from pathlib import Path

import pipe
from app_env import (
    HF_GSK_HUB_HF_TOKEN,
    HF_GSK_HUB_KEY,
    HF_GSK_HUB_PROJECT_KEY,
    HF_GSK_HUB_UNLOCK_TOKEN,
    HF_GSK_HUB_URL,
    HF_REPO_ID,
    HF_SPACE_ID,
    HF_WRITE_TOKEN,
)
from io_utils import LOG_FILE, get_submitted_yaml_path, write_log_to_user_file
from isolated_env import prepare_venv
from leaderboard import LEADERBOARD

is_running = False

logger = logging.getLogger(__file__)


def start_process_run_job():
    try:
        logging.debug("Running jobs in thread")
        global thread, is_running
        thread = threading.Thread(target=run_job)
        thread.daemon = True
        is_running = True
        thread.start()

    except Exception as e:
        print("Failed to start thread: ", e)


def stop_thread():
    logging.debug("Stop thread")
    global is_running
    is_running = False


def prepare_env_and_get_command(
    m_id,
    d_id,
    config,
    split,
    inference_token,
    uid,
    label_mapping,
    feature_mapping,
    verbose,
):
    leaderboard_dataset = None
    if os.environ.get("SPACE_ID") == "giskardai/giskard-evaluator":
        leaderboard_dataset = LEADERBOARD

    executable = "giskard_scanner"
    try:
        # Copy the current requirements (might be changed)
        with open("requirements.txt", "r") as f:
            executable = prepare_venv(
                uid,
                "\n".join(f.readlines()),
            )
        logger.info(f"Using {executable} as executable")
    except Exception as e:
        logger.warning(f"Create env failed due to {e}, using the current env as fallback.")
        executable = "giskard_scanner"

    command = [
        executable,
        "--loader",
        "huggingface",
        "--model",
        m_id,
        "--dataset",
        d_id,
        "--dataset_config",
        config,
        "--dataset_split",
        split,
        "--output_format",
        "markdown",
        "--output_portal",
        "huggingface",
        "--feature_mapping",
        json.dumps(feature_mapping),
        "--label_mapping",
        json.dumps(label_mapping),
        "--scan_config",
        get_submitted_yaml_path(uid),
        "--inference_type",
        "hf_inference_api",
        "--inference_api_token",
        inference_token,
        "--persist_scan",
    ]
    # The token to publish post
    if os.environ.get(HF_WRITE_TOKEN):
        command.append("--hf_token")
        command.append(os.environ.get(HF_WRITE_TOKEN))

    # The repo to publish for ranking
    if leaderboard_dataset:
        command.append("--leaderboard_dataset")
        command.append(leaderboard_dataset)

    # The info to upload to Giskard hub
    if os.environ.get(HF_GSK_HUB_KEY):
        command.append("--giskard_hub_api_key")
        command.append(os.environ.get(HF_GSK_HUB_KEY))
        if os.environ.get(HF_GSK_HUB_URL):
            command.append("--giskard_hub_url")
            command.append(os.environ.get(HF_GSK_HUB_URL))
        if os.environ.get(HF_GSK_HUB_PROJECT_KEY):
            command.append("--giskard_hub_project_key")
            command.append(os.environ.get(HF_GSK_HUB_PROJECT_KEY))
        if os.environ.get(HF_GSK_HUB_HF_TOKEN):
            command.append("--giskard_hub_hf_token")
            command.append(os.environ.get(HF_GSK_HUB_HF_TOKEN))
        if os.environ.get(HF_GSK_HUB_UNLOCK_TOKEN):
            command.append("--giskard_hub_unlock_token")
            command.append(os.environ.get(HF_GSK_HUB_UNLOCK_TOKEN))

    if verbose:
        command.append("--verbose")

    eval_str = f"[{m_id}]<{d_id}({config}, {split} set)>"

    write_log_to_user_file(
        uid,
        f"Start local evaluation on {eval_str}. Please wait for your job to start...\n",
    )

    return command


def save_job_to_pipe(task_id, job, description, lock):
    with lock:
        pipe.jobs.append((task_id, job, description))


def pop_job_from_pipe():
    if len(pipe.jobs) == 0:
        return
    job_info = pipe.jobs.pop()
    pipe.current = job_info[2]
    task_id = job_info[0]

    # Link to LOG_FILE
    log_file_path = Path(LOG_FILE)
    if log_file_path.exists():
        log_file_path.unlink()
    os.symlink(f"./tmp/{task_id}.log", LOG_FILE)

    write_log_to_user_file(task_id, f"Running job id {task_id}\n")
    command = prepare_env_and_get_command(*job_info[1])

    with open(f"./tmp/{task_id}.log", "a") as log_file:
        return_code = None
        p = subprocess.Popen(command, stdout=log_file, stderr=subprocess.STDOUT)
        while pipe.current and return_code is None:
            # Wait for finishing
            try:
                return_code = p.wait(timeout=1)
            except subprocess.TimeoutExpired:
                return_code = None

        if not pipe.current:
            # Job interrupted before finishing
            p.kill()

            log_file.write(f"\nJob interrupted by admin at {time.asctime()}\n")

        if return_code:
            log_file.write(f"\nJob finished with {return_code} at {time.asctime()}\n")

    pipe.current = None


def run_job():
    global is_running
    while is_running:
        try:
            pop_job_from_pipe()
            time.sleep(10)
        except KeyboardInterrupt:
            logging.debug("KeyboardInterrupt stop background thread")
            is_running = False
            break