This commit is contained in:
2026-01-18 18:41:03 +03:00
parent ba9a7b5539
commit e7d064908f
11 changed files with 167 additions and 93 deletions
+1 -1
View File
@@ -3,7 +3,7 @@ from argenta.app import StaticDividingLine
from .handlers import router
app = App(initial_message="metrics", prompt=">>> ", dividing_line=StaticDividingLine('~', length=120))
app = App(initial_message="metrics", prompt=">>> ", dividing_line=StaticDividingLine('~', length=70))
orchestrator = Orchestrator()
+6
View File
@@ -12,3 +12,9 @@ class BenchmarksNotFound(Exception):
def __str__(self):
return f"Benchmarks with type '{self.type_}' not found"
class BenchmarksWithSameNameAlreadyExists(Exception):
def __init__(self, benchmark_name: str):
self.benchmark_name = benchmark_name
def __str__(self):
return f"Benchmarks with name '{self.benchmark_name}' already exists"
+9 -3
View File
@@ -13,8 +13,7 @@ import gc
import statistics
from typing import Callable, override
from .exceptions import BenchmarkNotFound, BenchmarksNotFound
from .exceptions import BenchmarkNotFound, BenchmarksNotFound, BenchmarksWithSameNameAlreadyExists
FuncForBenchmark = Callable[[], None]
MILLISECONDS_IN_SECONDS = 1000
@@ -35,6 +34,8 @@ class BenchmarkResult:
@dataclass(frozen=True, slots=True)
class BenchmarkGroupResult:
type_: str
iterations: int
is_gc_disabled: bool
benchmark_results: list[BenchmarkResult]
@@ -107,8 +108,11 @@ class Benchmarks:
name=func.__name__,
description=description or f'description for {func.__name__} with type {type_}',
)
self._benchmarks.append(benchmark)
if self._benchmarks_paired_by_name.get(func.__name__):
raise BenchmarksWithSameNameAlreadyExists(func.__name__)
self._benchmarks_paired_by_name[func.__name__] = benchmark
self._benchmarks.append(benchmark)
self._benchmarks_grouped_by_type.setdefault(type_, []).append(benchmark)
return func
return decorator
@@ -145,6 +149,8 @@ class Benchmarks:
return BenchmarkGroupResult(
type_=type_,
iterations=iterations,
is_gc_disabled=is_gc_disabled,
benchmark_results=benchmark_results
)
+2 -5
View File
@@ -1,14 +1,11 @@
__all__ = [
"benchmark_few_commands",
"benchmark_many_commands",
"benchmark_many_commands_most_similar",
"benchmark_many_aliases",
"benchmark_partial_match",
"benchmark_extreme_commands"
]
import io
from contextlib import redirect_stdout
from argenta import App
from argenta.command.models import Command
from argenta.response import Response
@@ -39,7 +36,7 @@ def benchmark_few_commands() -> None:
@benchmarks.register(type_="most_similar_command", description="Many commands (50 commands, no match)")
def benchmark_many_commands() -> None:
def benchmark_many_commands_most_similar() -> None:
app = setup_app_with_commands(50)
app._most_similar_command("unknown")
+4 -9
View File
@@ -11,7 +11,6 @@ from argenta.response import Response
from argenta.router import Router
from .benchmarks.core.models import BenchmarkGroupResult
from .benchmarks.entity import benchmarks as registered_benchmarks
from .utils import get_kernel_version, get_gpu_info
console = Console()
router = Router(title="Metrics commands:")
@@ -43,22 +42,18 @@ def all_print_handler(_: Response) -> None:
type_grouped_benchmarks: list[BenchmarkGroupResult] = registered_benchmarks.run_benchmarks_grouped_by_type()
for results in type_grouped_benchmarks:
header_text = Text(f"TYPE: {results.type_.upper()}", style="bold magenta")
header_text = Text(f"TYPE: {results.type_.upper()} ; ITERATIONS: {results.iterations} ; ALL TIME IN MS", 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")
table.add_column("Median Time (ms)", justify="right", style="bold yellow")
table.add_column("Stdev (ms)", justify="right", style="bold yellow")
table.add_column("Avg Time", justify="right", style="bold yellow")
table.add_column("Median Time", justify="right", style="bold yellow")
table.add_column("Stdev", justify="right", style="bold yellow")
for benchmark in results.benchmark_results:
table.add_row(
benchmark.name,
benchmark.description,
str(benchmark.iterations),
str(benchmark.avg_time),
str(benchmark.median_time),
str(benchmark.std_dev),
View File
+114
View File
@@ -0,0 +1,114 @@
__all__ = [
"SystemInfo",
"SystemInfoGetter",
"get_system_info"
]
from dataclasses import dataclass
import platform
import sys
from typing import Protocol
import cpuinfo
import psutil
@dataclass(frozen=True, slots=True)
class SystemInfo:
os_info: OSInfo
cpu_info: CPUInfo
memory_info: MemoryInfo
python_version: str
python_implementation: str
@dataclass(frozen=True, slots=True)
class OSInfo:
os_name: str
kernel_version: str
@dataclass(frozen=True, slots=True)
class CPUInfo:
cpu_name: str
cpu_architecture: str
cpu_physical_cores: int
cpu_logical_cores: int
cpu_max_frequency: float
cpu_base_frequency: float
@dataclass(frozen=True, slots=True)
class MemoryInfo:
total_ram: float # in GB
available_ram: float # in GB
l1_cache: float
l2_cache: float
l3_cache: float
@dataclass(frozen=True, slots=True)
class PythonInfo:
python_version: str
python_implementation: str
python_compiler: str
class SystemInfoGetter(Protocol):
def __call__(self) -> SystemInfo:
raise NotImplementedError
def get_system_info() -> SystemInfo:
os_info = get_os_info()
os_name = os_info.os_name
os_kernel_version = os_info.kernel_version
cpu_info = cpuinfo.get_cpu_info()
cpu_architecture = cpu_info["arch"]
cpu_name = cpu_info["brand_raw"]
gpu_name = get_gpu_name()
total_ram = psutil.virtual_memory().total / (1024 ** 3)
python_version = platform.python_version()
python_implementation = platform.python_implementation()
return SystemInfo(
os_name=os_name,
kernel_version=os_kernel_version,
cpu_architecture=cpu_architecture,
cpu_name=cpu_name,
gpu_name=gpu_name,
total_ram=total_ram,
python_version=python_version,
python_implementation=python_implementation,
)
def get_os_info() -> OSInfo:
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 OSInfo(
os_name=product_name,
kernel_version=kernel_version,
)
elif system == "Darwin":
return OSInfo(
kernel_version=platform.release(),
os_name=f"macOS {platform.mac_ver()[0]}"
)
else:
return OSInfo(
kernel_version=platform.release(),
os_name=platform.system()
)
-63
View File
@@ -1,63 +0,0 @@
__all__ = [
"get_kernel_version",
"get_gpu_info"
]
import platform
import sys
import pynvml
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 get_gpu_info() -> str:
try:
pynvml.nvmlInit()
device_count = pynvml.nvmlDeviceGetCount()
if device_count == 0:
return "N/A"
handle = pynvml.nvmlDeviceGetHandleByIndex(0)
name = pynvml.nvmlDeviceGetName(handle)
if isinstance(name, bytes):
name = name.decode("utf-8")
pynvml.nvmlShutdown()
return name
except pynvml.NVMLError:
return "N/A"