evals: move scripts/release to script/evals and add README
This commit is contained in:
Aleix Conchillo Flaqué
45
scripts/evals/README.md
Normal file
45
scripts/evals/README.md
Normal file
@@ -0,0 +1,45 @@
|
||||
# Pipecat Evals
|
||||
|
||||
This directory contains a set of utilities to help test Pipecat, specifically
|
||||
its examples.
|
||||
|
||||
## Release Evals
|
||||
|
||||
Before any Pipecat release, we make sure that all (or most) of the examples work
|
||||
flawlessly. We have 100+ examples, and checking each one manually was very
|
||||
time-consuming (and painful!), especially because we aim to release often.
|
||||
|
||||
To make this process easier, we designed these "release evals," which do the
|
||||
following:
|
||||
|
||||
- Start one of the foundational examples (the user bot)
|
||||
- Start an eval bot
|
||||
|
||||
The user bot (i.e. the example) introduces itself, and the eval bot then asks a
|
||||
question. The user bot replies, and the eval bot verifies the response.
|
||||
|
||||
For example, the eval bot might ask:
|
||||
|
||||
"What's 2 plus 2?"
|
||||
|
||||
The user bot replies:
|
||||
|
||||
"2 plus 2 is 4."
|
||||
|
||||
The eval bot (powered by an LLM) evaluates the response and emits a result. It
|
||||
also explains why it thinks the answer is valid or invalid.
|
||||
|
||||
To run the release evals:
|
||||
|
||||
```sh
|
||||
python run-release-evals.py -a -v
|
||||
```
|
||||
|
||||
This runs all the evals and stores logs and audio (`-a`) for each test.
|
||||
|
||||
You can also specify which tests to run. For example, to run all `07` series
|
||||
tests:
|
||||
|
||||
```sh
|
||||
python run-release-evals.py -p 07 -a -v
|
||||
```
|
||||
290
scripts/evals/eval.py
Normal file
290
scripts/evals/eval.py
Normal file
@@ -0,0 +1,290 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import io
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import wave
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
import aiofiles
|
||||
from loguru import logger
|
||||
from utils import (
|
||||
EvalResult,
|
||||
load_module_from_path,
|
||||
print_begin_test,
|
||||
print_end_test,
|
||||
print_test_results,
|
||||
)
|
||||
|
||||
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
||||
from pipecat.adapters.schemas.tools_schema import ToolsSchema
|
||||
from pipecat.audio.vad.silero import SileroVADAnalyzer
|
||||
from pipecat.audio.vad.vad_analyzer import VADParams
|
||||
from pipecat.frames.frames import EndTaskFrame
|
||||
from pipecat.pipeline.pipeline import Pipeline
|
||||
from pipecat.pipeline.runner import PipelineRunner
|
||||
from pipecat.pipeline.task import PipelineParams, PipelineTask
|
||||
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
|
||||
from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
|
||||
from pipecat.processors.frame_processor import FrameDirection
|
||||
from pipecat.services.cartesia.tts import CartesiaTTSService
|
||||
from pipecat.services.deepgram.stt import DeepgramSTTService
|
||||
from pipecat.services.llm_service import FunctionCallParams
|
||||
from pipecat.services.openai.llm import OpenAILLMService
|
||||
from pipecat.transports.services.daily import DailyParams, DailyTransport
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
|
||||
FOUNDATIONAL_DIR = SCRIPT_DIR.parent.parent / "examples" / "foundational"
|
||||
|
||||
sys.path.insert(0, os.path.abspath(FOUNDATIONAL_DIR))
|
||||
|
||||
EVAL_PROMPT = ""
|
||||
|
||||
PIPELINE_IDLE_TIMEOUT_SECS = 30
|
||||
|
||||
|
||||
class EvalRunner:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
pattern: str = "",
|
||||
record_audio: bool = False,
|
||||
name: Optional[str] = None,
|
||||
log_level: str = "DEBUG",
|
||||
):
|
||||
self._pattern = f".*{pattern}.*" if pattern else ""
|
||||
self._record_audio = record_audio
|
||||
self._log_level = log_level
|
||||
self._total_success = 0
|
||||
self._tests: List[EvalResult] = []
|
||||
self._queue = asyncio.Queue()
|
||||
|
||||
# We to save runner files.
|
||||
name = name or f"{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
||||
self._runs_dir = os.path.join(SCRIPT_DIR, "test-runs", name)
|
||||
self._logs_dir = os.path.join(self._runs_dir, "logs")
|
||||
self._recordings_dir = os.path.join(self._runs_dir, "recordings")
|
||||
os.makedirs(self._logs_dir, exist_ok=True)
|
||||
os.makedirs(self._recordings_dir, exist_ok=True)
|
||||
|
||||
async def assert_eval(self, params: FunctionCallParams):
|
||||
reasoning = params.arguments["reasoning"]
|
||||
logger.debug(f"🧠 EVAL REASONING: {reasoning}")
|
||||
await self._queue.put(params.arguments["result"])
|
||||
await params.result_callback(None)
|
||||
await params.llm.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
|
||||
|
||||
async def assert_eval_false(self):
|
||||
await self._queue.put(False)
|
||||
|
||||
async def run_eval(self, example_file: str, prompt: str, eval: Optional[str] = None):
|
||||
if not re.match(self._pattern, example_file):
|
||||
return
|
||||
|
||||
# Store logs
|
||||
filename = self._log_file_name(example_file)
|
||||
log_file_id = logger.add(filename, level=self._log_level)
|
||||
|
||||
print_begin_test(example_file)
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
try:
|
||||
await asyncio.wait(
|
||||
[
|
||||
asyncio.create_task(run_example_pipeline(example_file)),
|
||||
asyncio.create_task(run_eval_pipeline(self, example_file, prompt, eval)),
|
||||
],
|
||||
timeout=90,
|
||||
)
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"ERROR: Unable to run {example_file}: {e}")
|
||||
|
||||
try:
|
||||
result = await asyncio.wait_for(self._queue.get(), timeout=1.0)
|
||||
except asyncio.TimeoutError:
|
||||
result = False
|
||||
|
||||
if result:
|
||||
self._total_success += 1
|
||||
|
||||
eval_time = time.time() - start_time
|
||||
|
||||
self._tests.append(EvalResult(name=example_file, result=result, time=eval_time))
|
||||
|
||||
print_end_test(example_file, result, eval_time)
|
||||
|
||||
logger.remove(log_file_id)
|
||||
|
||||
def print_results(self):
|
||||
print_test_results(self._tests, self._total_success, self._runs_dir)
|
||||
|
||||
async def save_audio(self, name: str, audio: bytes, sample_rate: int, num_channels: int):
|
||||
if len(audio) > 0:
|
||||
filename = self._recording_file_name(name)
|
||||
with io.BytesIO() as buffer:
|
||||
with wave.open(buffer, "wb") as wf:
|
||||
wf.setsampwidth(2)
|
||||
wf.setnchannels(num_channels)
|
||||
wf.setframerate(sample_rate)
|
||||
wf.writeframes(audio)
|
||||
async with aiofiles.open(filename, "wb") as file:
|
||||
await file.write(buffer.getvalue())
|
||||
logger.debug(f"Saving {name} audio to {filename}")
|
||||
else:
|
||||
logger.warning(f"There's no audio to save for {name}")
|
||||
|
||||
def _base_file_name(self, example_file: str):
|
||||
base_name = os.path.splitext(example_file)[0]
|
||||
return f"{base_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
||||
|
||||
def _log_file_name(self, example_file: str):
|
||||
base_name = self._base_file_name(example_file)
|
||||
return os.path.join(self._logs_dir, f"{base_name}.log")
|
||||
|
||||
def _recording_file_name(self, example_file: str):
|
||||
base_name = self._base_file_name(example_file)
|
||||
return os.path.join(self._recordings_dir, f"{base_name}.wav")
|
||||
|
||||
|
||||
async def run_example_pipeline(example_file: str):
|
||||
room_url = os.getenv("DAILY_SAMPLE_ROOM_URL")
|
||||
|
||||
script_path = FOUNDATIONAL_DIR / example_file
|
||||
|
||||
module = load_module_from_path(script_path)
|
||||
|
||||
transport = DailyTransport(
|
||||
room_url,
|
||||
None,
|
||||
"Pipecat",
|
||||
DailyParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
vad_analyzer=SileroVADAnalyzer(),
|
||||
),
|
||||
)
|
||||
|
||||
await module.run_example(transport, argparse.Namespace(), True)
|
||||
|
||||
|
||||
async def run_eval_pipeline(
|
||||
eval_runner: EvalRunner, example_file: str, prompt: str, eval: Optional[str]
|
||||
):
|
||||
logger.info(f"Starting eval bot")
|
||||
|
||||
room_url = os.getenv("DAILY_SAMPLE_ROOM_URL")
|
||||
|
||||
transport = DailyTransport(
|
||||
room_url,
|
||||
None,
|
||||
"Pipecat Eval",
|
||||
DailyParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=2.0)),
|
||||
),
|
||||
)
|
||||
|
||||
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
|
||||
|
||||
tts = CartesiaTTSService(
|
||||
api_key=os.getenv("CARTESIA_API_KEY"),
|
||||
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
|
||||
)
|
||||
|
||||
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
|
||||
|
||||
llm.register_function("assert_eval", eval_runner.assert_eval)
|
||||
|
||||
eval_function = FunctionSchema(
|
||||
name="assert_eval",
|
||||
description="Called when the user answers a question.",
|
||||
properties={
|
||||
"result": {
|
||||
"type": "boolean",
|
||||
"description": "The result of the eval",
|
||||
},
|
||||
"reasoning": {
|
||||
"type": "string",
|
||||
"description": "Why the answer was considered correct or invalid",
|
||||
},
|
||||
},
|
||||
required=["result", "reasoning"],
|
||||
)
|
||||
tools = ToolsSchema(standard_tools=[eval_function])
|
||||
|
||||
# See if we need to include an eval prompt.
|
||||
eval_prompt = ""
|
||||
if eval:
|
||||
eval_prompt = f"The answer is correct if the user says [{eval}]."
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": f"You are an LLM eval, be extremly brief. Your goal is to only ask one question: {prompt}. Call the eval function only if the user answers the question and check if the answer is correct (words as numbers are valid). {eval_prompt}",
|
||||
},
|
||||
]
|
||||
|
||||
context = OpenAILLMContext(messages, tools)
|
||||
context_aggregator = llm.create_context_aggregator(context)
|
||||
|
||||
audio_buffer = AudioBufferProcessor()
|
||||
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
transport.input(), # Transport user input
|
||||
stt, # STT
|
||||
context_aggregator.user(), # User responses
|
||||
llm, # LLM
|
||||
tts, # TTS
|
||||
transport.output(), # Transport bot output
|
||||
audio_buffer,
|
||||
context_aggregator.assistant(), # Assistant spoken responses
|
||||
]
|
||||
)
|
||||
|
||||
task = PipelineTask(
|
||||
pipeline,
|
||||
params=PipelineParams(
|
||||
allow_interruptions=True,
|
||||
audio_in_sample_rate=16000,
|
||||
audio_out_sample_rate=16000,
|
||||
),
|
||||
idle_timeout_secs=PIPELINE_IDLE_TIMEOUT_SECS,
|
||||
)
|
||||
|
||||
@audio_buffer.event_handler("on_audio_data")
|
||||
async def on_audio_data(buffer, audio, sample_rate, num_channels):
|
||||
await eval_runner.save_audio(example_file, audio, sample_rate, num_channels)
|
||||
|
||||
@transport.event_handler("on_client_connected")
|
||||
async def on_client_connected(transport, client):
|
||||
logger.info(f"Client connected")
|
||||
await audio_buffer.start_recording()
|
||||
|
||||
@transport.event_handler("on_client_disconnected")
|
||||
async def on_client_disconnected(transport, client):
|
||||
logger.info(f"Client disconnected")
|
||||
await task.cancel()
|
||||
|
||||
@task.event_handler("on_idle_timeout")
|
||||
async def on_pipeline_idle_timeout(task):
|
||||
await eval_runner.assert_eval_false()
|
||||
|
||||
runner = PipelineRunner()
|
||||
|
||||
await runner.run(task)
|
||||
147
scripts/evals/run-release-evals.py
Normal file
147
scripts/evals/run-release-evals.py
Normal file
@@ -0,0 +1,147 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from eval import EvalRunner
|
||||
from loguru import logger
|
||||
from utils import check_env_variables
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
# Math
|
||||
PROMPT_SIMPLE_MATH = "A simple math addition."
|
||||
|
||||
# Weather
|
||||
PROMPT_WEATHER = "What's the weather in San Francisco?"
|
||||
EVAL_WEATHER = (
|
||||
"Something specific about the current weather in San Francisco, including the degrees."
|
||||
)
|
||||
|
||||
# Online search
|
||||
PROMPT_ONLINE_SEARCH = "What's the date right now in London?"
|
||||
EVAL_ONLINE_SEARCH = f"Today is {datetime.now(timezone.utc).strftime('%B %d, %Y')}."
|
||||
|
||||
TESTS_07 = [
|
||||
# 07 series
|
||||
("07-interruptible.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07-interruptible-cartesia-http.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07b-interruptible-langchain.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07c-interruptible-deepgram.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07d-interruptible-elevenlabs.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07d-interruptible-elevenlabs-http.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07e-interruptible-playht.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07e-interruptible-playht-http.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07f-interruptible-azure.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07g-interruptible-openai.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07h-interruptible-openpipe.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07j-interruptible-gladia.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07k-interruptible-lmnt.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07l-interruptible-groq.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07m-interruptible-aws.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07n-interruptible-google.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07o-interruptible-assemblyai.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07q-interruptible-rime.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07q-interruptible-rime-http.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07r-interruptible-riva-nim.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07s-interruptible-google-audio-in.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07t-interruptible-fish.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07v-interruptible-neuphonic.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07v-interruptible-neuphonic-http.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07w-interruptible-fal.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07y-interruptible-minimax.py", PROMPT_SIMPLE_MATH, None),
|
||||
("07z-interruptible-sarvam.py", PROMPT_SIMPLE_MATH, None),
|
||||
# Needs a local XTTS docker instance running.
|
||||
# ("07i-interruptible-xtts.py", PROMPT_SIMPLE_MATH, None),
|
||||
# Needs a Krisp license.
|
||||
# ("07p-interruptible-krisp.py", PROMPT_SIMPLE_MATH, None),
|
||||
# Needs GPU resources.
|
||||
# ("07u-interruptible-ultravox.py", PROMPT_SIMPLE_MATH, None),
|
||||
]
|
||||
|
||||
TESTS_14 = [
|
||||
("14-function-calling.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14a-function-calling-anthropic.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14b-function-calling-anthropic-video.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14d-function-calling-video.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14e-function-calling-gemini.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14f-function-calling-groq.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14g-function-calling-grok.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14h-function-calling-azure.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14i-function-calling-fireworks.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14j-function-calling-nim.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14n-function-calling-perplexity.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14q-function-calling-qwen.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("14r-function-calling-aws.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
# Currently not working.
|
||||
# ("14c-function-calling-together.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
# ("14j-function-calling-nim.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
# ("14k-function-calling-cerebras.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
# ("14l-function-calling-deepseek.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
# ("14m-function-calling-openrouter.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
# ("14o-function-calling-gemini-openai-format.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
# ("14p-function-calling-gemini-vertex-ai.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
]
|
||||
|
||||
TESTS_19 = [
|
||||
("19-openai-realtime-beta.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("19a-azure-realtime-beta.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
]
|
||||
|
||||
TESTS_26 = [
|
||||
("26-gemini-multimodal-live.py", PROMPT_SIMPLE_MATH, None),
|
||||
("26a-gemini-multimodal-live-transcription.py", PROMPT_SIMPLE_MATH, None),
|
||||
("26b-gemini-multimodal-live-function-calling.py", PROMPT_WEATHER, EVAL_WEATHER),
|
||||
("26c-gemini-multimodal-live-video.py", PROMPT_SIMPLE_MATH, None),
|
||||
("26e-gemini-multimodal-google-search.py", PROMPT_ONLINE_SEARCH, EVAL_ONLINE_SEARCH),
|
||||
# Currently not working.
|
||||
# ("26d-gemini-multimodal-live-text.py", PROMPT_SIMPLE_MATH, None),
|
||||
]
|
||||
|
||||
TESTS = [
|
||||
*TESTS_07,
|
||||
*TESTS_14,
|
||||
*TESTS_19,
|
||||
*TESTS_26,
|
||||
]
|
||||
|
||||
|
||||
async def main(args: argparse.Namespace):
|
||||
if not check_env_variables():
|
||||
return
|
||||
|
||||
# Log level
|
||||
logger.remove(0)
|
||||
log_level = "TRACE" if args.verbose >= 2 else "DEBUG"
|
||||
if args.verbose:
|
||||
logger.add(sys.stderr, level=log_level)
|
||||
|
||||
runner = EvalRunner(
|
||||
name=args.name,
|
||||
pattern=args.pattern,
|
||||
record_audio=args.audio,
|
||||
log_level=log_level,
|
||||
)
|
||||
|
||||
for test, prompt, eval in TESTS:
|
||||
await runner.run_eval(test, prompt, eval)
|
||||
|
||||
runner.print_results()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="Pipecat Eval Runner")
|
||||
parser.add_argument("--audio", "-a", action="store_true", help="Record audio for each test")
|
||||
parser.add_argument("--name", "-n", help="Name for the current runner (e.g. 'v.0.0.68')")
|
||||
parser.add_argument("--pattern", "-p", help="Only run tests that match the pattern")
|
||||
parser.add_argument("--verbose", "-v", action="count", default=0)
|
||||
args = parser.parse_args()
|
||||
|
||||
asyncio.run(main(args))
|
||||
84
scripts/evals/utils.py
Normal file
84
scripts/evals/utils.py
Normal file
@@ -0,0 +1,84 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Sequence
|
||||
|
||||
GREEN = "\033[92m"
|
||||
RED = "\033[91m"
|
||||
RESET = "\033[0m"
|
||||
CLEAR = "\033[K"
|
||||
|
||||
|
||||
@dataclass
|
||||
class EvalResult:
|
||||
name: str
|
||||
result: bool
|
||||
time: float
|
||||
|
||||
|
||||
def check_env_variables() -> bool:
|
||||
required_envs = [
|
||||
"CARTESIA_API_KEY",
|
||||
"DEEPGRAM_API_KEY",
|
||||
"OPENAI_API_KEY",
|
||||
"DAILY_SAMPLE_ROOM_URL",
|
||||
]
|
||||
for env in required_envs:
|
||||
if not os.getenv(env):
|
||||
print(f"\nERROR: Environment variable {env} is not defined.\n")
|
||||
print(f"Required environment variables: {required_envs}")
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def print_begin_test(example_file: str):
|
||||
print(f"{example_file:<55} RUNNING...{CLEAR}", end="\r", flush=True)
|
||||
|
||||
|
||||
def print_end_test(example_file: str, passed: bool, time: float):
|
||||
status = f"{GREEN}✅ OK{RESET}" if passed else f"{RED}❌ FAILED{RESET}"
|
||||
print(f"{example_file:<55} {status} ({time:.2f}s){CLEAR}")
|
||||
|
||||
|
||||
def print_test_results(tests: Sequence[EvalResult], total_success: int, location: str):
|
||||
total_count = len(tests)
|
||||
|
||||
bar = "=" * 80
|
||||
|
||||
print()
|
||||
print(f"{GREEN}{bar}{RESET}")
|
||||
print(f"TOTAL NUMBER OF TESTS: {total_count}")
|
||||
print()
|
||||
|
||||
total_time = 0.0
|
||||
total_count = len(tests)
|
||||
for eval in tests:
|
||||
total_time += eval.time
|
||||
print_end_test(eval.name, eval.result, eval.time)
|
||||
|
||||
total_fail = total_count - total_success
|
||||
|
||||
print()
|
||||
print(
|
||||
f"{GREEN}SUCCESS{RESET}: {total_success} | {RED}FAIL{RESET}: {total_fail} | TOTAL TIME: {total_time:.2f}s"
|
||||
)
|
||||
print(f"{GREEN}{bar}{RESET}")
|
||||
print()
|
||||
print(f"Tests output: {location}")
|
||||
|
||||
|
||||
def load_module_from_path(path: str | Path):
|
||||
path = Path(path).resolve()
|
||||
module_name = path.stem
|
||||
|
||||
spec = importlib.util.spec_from_file_location(module_name, str(path))
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
Reference in New Issue
Block a user