kolo/Argenta
1
0
Fork 0
mirror of https://github.com/koloideal/Argenta.git synced 2026-06-10 10:05:28 +03:00
Code Issues Packages Projects Releases Wiki Activity

benchs

Browse Source
This commit is contained in:
kolo
2026-01-17 00:42:00 +03:00
parent 00cb94ffe2
commit 6aa6b0f179
12 changed files with 201 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
metrics/__init__.py
-1
View File
@@ -1 +0,0 @@
from .benchmarks import *
metrics/__main__.py
+8 -41
View File
@@ -1,48 +1,15 @@
from concurrent.futures import ProcessPoolExecutor
import os
from argenta import App, Orchestrator
from argenta.app import DynamicDividingLine
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.text import Text
from .handlers import router
from metrics.utils import run_benchmark, BenchmarkResult
from .registry import Benchmarks, Benchmark
app = App(initial_message="metrics", prompt=">>> ", dividing_line=DynamicDividingLine('~'))
orchestrator = Orchestrator()
def main():
console = Console()
all_benchmarks: list[Benchmark] = Benchmarks.get_benchmarks()
workers = os.cpu_count() or 1
with ProcessPoolExecutor(max_workers=workers) as executor:
results = executor.map(run_benchmark, all_benchmarks)
type_paired_benchmarks: dict[str, list[BenchmarkResult]] = {}
for result in results:
type_paired_benchmarks.setdefault(result.type_, []).append(result)
for type_, benchmarks in type_paired_benchmarks.items():
header_text = Text(f"TYPE: {type_.upper()}", style="bold magenta")
console.print(Panel(header_text, expand=False, border_style="magenta"))
table = Table(show_header=True, header_style="bold cyan", border_style="blue", show_lines=True)
table.add_column("Name", style="green")
table.add_column("Description", style="dim")
table.add_column("Iterations", justify="right")
table.add_column("Avg Time (ms)", justify="right", style="bold yellow")
for benchmark in benchmarks:
table.add_row(
benchmark.name,
benchmark.description,
str(benchmark.iterations),
str(benchmark.avg_time)
)
console.print(table)
console.print()
def main() -> None:
app.include_router(router)
orchestrator.start_polling(app)
if __name__ == "__main__":
metrics/benchmarks/__init__.py
+1 -1
View File
@@ -1 +1 @@
from .pre_cycle_setup import *
from .core import *
metrics/benchmarks/core/__init__.py
+1
View File
@@ -0,0 +1 @@
from .pre_cycle_setup import *
metrics/benchmarks/pre_cycle_setup.py → metrics/benchmarks/core/pre_cycle_setup.py
+2 -2
View File
@@ -7,12 +7,12 @@ __all__ = [
]
from argenta import App
from argenta.router import Router
from argenta.command.models import Command
from argenta.response import Response
from argenta.router import Router
from ..models import benchmark
from ..utils import get_time_of_pre_cycle_setup
from ..registry import benchmark
@benchmark(type_="pre_cycle_setup", description="With no aliases")
metrics/registry.py → metrics/benchmarks/models.py
+12
View File
@@ -1,14 +1,26 @@
__all__ = [
"Benchmark",
"Benchmarks",
"BenchmarkResult",
"benchmark"
]
from dataclasses import dataclass
from decimal import Decimal
from typing import Callable, ClassVar, overload, override
BenchmarkAsFunc = Callable[[], float]
@dataclass(frozen=True)
class BenchmarkResult:
type_: str
name: str
description: str
iterations: int
avg_time: Decimal
class Benchmark:
def __init__(
self,
metrics/benchmarks/utils.py
+89
View File
@@ -0,0 +1,89 @@
__all__ = [
"get_time_of_pre_cycle_setup",
"attempts_to_average",
"run_benchmark",
"run_all_benchmarks",
"get_kernel_version"
]
import io
import os
import platform
import sys
import time
from concurrent.futures import ProcessPoolExecutor
from contextlib import redirect_stdout
from decimal import ROUND_HALF_UP, Decimal
from argenta import App
from .models import Benchmark, BenchmarkResult, Benchmarks
def get_time_of_pre_cycle_setup(app: App) -> float:
start = time.perf_counter()
with redirect_stdout(io.StringIO()):
app._pre_cycle_setup() # pyright: ignore[reportPrivateUsage]
end = time.perf_counter()
return (end - start) * 1000 # as milliseconds
def get_kernel_version() -> dict[str, str]:
system = platform.system()
if system == "Windows":
ver = sys.getwindowsversion()
kernel_version = f"{ver.major}.{ver.minor}.{ver.build}"
if ver.build >= 22000:
product_name = "Windows 11"
else:
product_name = "Windows 10"
return {
'kernel_version': kernel_version,
'product_name': product_name
}
elif system == "Linux":
return {
'kernel_version': platform.release(),
'product_name': platform.system()
}
elif system == "Darwin":
return {
'kernel_version': platform.release(),
'product_name': f"macOS {platform.mac_ver()[0]}"
}
else:
return {
'kernel_version': platform.release(),
'product_name': platform.system(),
}
def attempts_to_average(bench_attempts: list[float], iterations: int) -> Decimal:
return Decimal(sum(bench_attempts) / iterations).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
def run_all_benchmarks() -> dict[str, list[BenchmarkResult]]:
all_benchmarks: list[Benchmark] = Benchmarks.get_benchmarks()
workers = os.cpu_count() or 1
with ProcessPoolExecutor(max_workers=workers) as executor:
results = executor.map(run_benchmark, all_benchmarks)
type_paired_benchmarks: dict[str, list[BenchmarkResult]] = {}
for result in results:
type_paired_benchmarks.setdefault(result.type_, []).append(result)
return type_paired_benchmarks
def run_benchmark(benchmark: Benchmark) -> BenchmarkResult:
bench_attempts: list[float] = []
for _ in range(benchmark.iterations):
bench_attempts.append(benchmark.run())
avg = attempts_to_average(bench_attempts, benchmark.iterations)
return BenchmarkResult(benchmark.type_, benchmark.name, benchmark.description, benchmark.iterations, avg)
metrics/handlers.py
+70
View File
@@ -0,0 +1,70 @@
import platform
import cpuinfo
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from argenta.command.models import Command
from argenta.response import Response
from argenta.router import Router
from .benchmarks.models import BenchmarkResult
from .benchmarks.utils import run_all_benchmarks, get_kernel_version
console = Console()
router = Router(title="Metrics commands:")
@router.command(Command("all-print", description="Print all benchmarks results"))
def all_print_handler(_: Response) -> None:
cpu_info = cpuinfo.get_cpu_info()
os_info = get_kernel_version()
table = Table(show_header=True, header_style="bold cyan", border_style="blue", show_lines=True)
table.add_column("Parameter", style="green")
table.add_column("Value", style="yellow")
table.add_row("OS", platform.system())
table.add_row("OS Name", os_info['product_name'])
table.add_row("OS Kernel Version", os_info['kernel_version'])
table.add_row("Architecture", cpu_info['arch'])
table.add_row("Processor", cpu_info['brand_raw'])
table.add_row("Python Version", cpu_info['python_version'])
table.add_row("Python Implementation", platform.python_implementation())
header_text = Text("SYSTEM INFO", style="bold magenta")
console.print(Panel(header_text, expand=False, border_style="magenta"))
console.print(table, end="\n\n")
type_paired_benchmarks: dict[str, list[BenchmarkResult]] = run_all_benchmarks()
for type_, benchmarks in type_paired_benchmarks.items():
header_text = Text(f"TYPE: {type_.upper()}", style="bold magenta")
console.print(Panel(header_text, expand=False, border_style="magenta"))
table = Table(show_header=True, header_style="bold cyan", border_style="blue", show_lines=True)
table.add_column("Name", style="green")
table.add_column("Description", style="dim")
table.add_column("Iterations", justify="right")
table.add_column("Avg Time (ms)", justify="right", style="bold yellow")
for benchmark in benchmarks:
table.add_row(
benchmark.name,
benchmark.description,
str(benchmark.iterations),
str(benchmark.avg_time)
)
console.print(table)
@router.command(Command("release-generate", description="Generate release report"))
def release_generate_handler(_: Response) -> None:
console.print("[yellow]Release report generation not implemented yet[/yellow]")
@router.command(Command("diagrams-generate", description="Generate diagrams"))
def diagrams_generate_handler(_: Response) -> None:
console.print("[yellow]Diagrams generation not implemented yet[/yellow]")
metrics/utils.py
-44
View File
@@ -1,44 +0,0 @@
__all__ = [
"get_time_of_pre_cycle_setup",
"attempts_to_average",
"run_benchmark",
"BenchmarkResult"
]
import io
from contextlib import redirect_stdout
import time
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from argenta import App
from metrics.registry import Benchmark
def get_time_of_pre_cycle_setup(app: App) -> float:
start = time.perf_counter()
with redirect_stdout(io.StringIO()):
app._pre_cycle_setup() # pyright: ignore[reportPrivateUsage]
end = time.perf_counter()
return (end - start) * 1000 # as milliseconds
def attempts_to_average(bench_attempts: list[float], iterations: int) -> Decimal:
return Decimal(sum(bench_attempts) / iterations).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
@dataclass(frozen=True)
class BenchmarkResult:
type_: str
name: str
description: str
iterations: int
avg_time: Decimal
def run_benchmark(benchmark: Benchmark) -> BenchmarkResult:
bench_attempts: list[float] = []
for _ in range(benchmark.iterations):
bench_attempts.append(benchmark.run())
avg = attempts_to_average(bench_attempts, benchmark.iterations)
return BenchmarkResult(benchmark.type_, benchmark.name, benchmark.description, benchmark.iterations, avg)