app.py

"""
Error Tester: a PyHARP app for exercising HARP's Gradio error handling.

The app performs no model inference. Its process endpoint deliberately raises
or returns selected failure shapes so HARP can be checked against the user-facing
messages in src/utils/Errors.h.
"""

import spaces
from pyharp import *

from typing import Optional, Tuple
import gradio as gr

import os
import tempfile
import time
import wave


model_card = ModelCard(
    name="Error Tester",
    description=(
        "Raises controlled Gradio errors, runtime errors, quota messages, "
        "Space-status messages, and malformed output payloads for testing "
        "HARP's error handling."
    ),
    author="TEAMuP",
    tags=["example", "test", "error handling", "v3"],
)


ERROR_FAMILIES = [
    "Success",
    "Gradio runtime error",
    "Python runtime error",
    "ZeroGPU quota error",
    "Space status error",
    "Output payload error",
]

ERROR_VARIANTS = [
    "Default message",
    "Empty detail",
    "Long detail",
    "Custom detail",
]

OUTPUT_PAYLOADS = [
    "Valid empty labels",
    "Malformed label JSON",
    "Missing audio file",
    "Unsupported audio extension",
]

DEFAULT_DETAIL = (
    "Synthetic failure from the PyHARP Error Tester. This is expected and is "
    "intended to test HARP's error popup."
)


def write_silent_wav(duration_seconds: float = 0.25, sample_rate: int = 44100) -> str:
    """Create a small WAV file using only the Python standard library."""

    fd, path = tempfile.mkstemp(prefix="harp_error_tester_", suffix=".wav")
    os.close(fd)

    frame_count = int(duration_seconds * sample_rate)

    with wave.open(path, "wb") as wav:
        wav.setnchannels(1)
        wav.setsampwidth(2)
        wav.setframerate(sample_rate)
        wav.writeframes(b"\x00\x00" * frame_count)

    return path


def build_detail(variant: str, custom_detail: str) -> str:
    if variant == "Empty detail":
        return ""

    if variant == "Long detail":
        return (
            DEFAULT_DETAIL
            + " "
            + "This deliberately long diagnostic text should be shortened in "
            + "HARP's transient status area while remaining available in the "
            + "final error dialog. "
            + "x" * 260
        )

    if variant == "Custom detail" and custom_detail.strip():
        return custom_detail.strip()

    return DEFAULT_DETAIL


def build_labels(message: str) -> LabelList:
    labels = LabelList()
    labels.labels.append(
        AudioLabel(
            t=0.0,
            label="success",
            duration=0.25,
            description=message,
            color=OutputLabel.rgb_color_to_int(20, 120, 80),
            amplitude=0.5,
        )
    )
    return labels

@spaces.GPU
def process_fn(
    input_audio_path: Optional[str],
    error_family: str,
    error_variant: str,
    output_payload: str,
    delay_seconds: float,
    custom_detail: str,
) -> Tuple[str, LabelList]:
    if delay_seconds > 0:
        time.sleep(float(delay_seconds))

    detail = build_detail(error_variant, custom_detail)

    if error_family == "Gradio runtime error":
        raise gr.Error(detail or "The Gradio server reported a runtime error.")

    if error_family == "Python runtime error":
        raise RuntimeError(detail or "Python RuntimeError without additional detail.")

    if error_family == "ZeroGPU quota error":
        quota_detail = detail or "GPU quota has been exceeded for this Space."
        raise gr.Error(f"ZeroGPU quota exceeded: {quota_detail}")

    if error_family == "Space status error":
        status_detail = detail or "Check its status on HF."
        raise gr.Error(f"Your Space is in error. {status_detail}")

    if error_family == "Output payload error":
        audio_path = write_silent_wav()

        if output_payload == "Malformed label JSON":
            return audio_path, {"labels": "this should be a list"}

        if output_payload == "Missing audio file":
            return "/tmp/harp_error_tester_missing_output.wav", LabelList()

        if output_payload == "Unsupported audio extension":
            fd, path = tempfile.mkstemp(prefix="harp_error_tester_", suffix=".txt")
            with os.fdopen(fd, "w") as f:
                f.write("This is not an audio file.\n")
            return path, LabelList()

    audio_path = input_audio_path or write_silent_wav()
    labels = build_labels(f"Selected family: {error_family}. No error was raised.")

    return audio_path, labels


with gr.Blocks() as demo:
    input_components = [
        gr.Audio(
            type="filepath",
            label="Optional Input Audio",
        )
        .harp_required(False)
        .set_info("Passed through on success. A silent WAV is generated if empty."),
        gr.Dropdown(
            choices=ERROR_FAMILIES,
            value="Gradio runtime error",
            label="Error Family",
            info="The broad error path to exercise."
        ),
        gr.Dropdown(
            choices=ERROR_VARIANTS,
            value="Default message",
            label="Error Detail",
            info="Controls the detail text carried by raised errors."
        ),
        gr.Dropdown(
            choices=OUTPUT_PAYLOADS,
            value="Valid empty labels",
            label="Output Payload",
            info="Used only when Error Family is Output payload error."
        ),
        gr.Slider(
            minimum=0,
            maximum=30,
            step=1,
            value=0,
            label="Delay (s)",
            info="Stalls before producing the selected response."
        ),
        gr.Textbox(
            value="",
            label="Custom Detail",
            info="Used when Error Detail is Custom detail."
        ),
    ]

    output_components = [
        gr.Audio(
            type="filepath",
            label="Output Audio",
        ).set_info("Silent or pass-through audio for successful responses."),
        gr.JSON(
            label="Output Labels",
        ).set_info("Labels returned on successful responses."),
    ]

    app = build_endpoint(
        model_card=model_card,
        input_components=input_components,
        output_components=output_components,
        process_fn=process_fn,
    )

demo.queue().launch(share=True, show_error=True, pwa=True)