bbbbbbenchh

This commit is contained in:
2026-01-18 01:59:55 +03:00
parent 1648a8206a
commit 69e871a639
16 changed files with 188 additions and 221 deletions
@@ -29,7 +29,7 @@ msgid ""
"взаимодействие с пользователем, координируя работу всех компонентов: "
"роутеров, обработчиков и системных сообщений."
msgstr ""
"The ``App`` object is the core of your console application. It handles "
"The ``App`` object is the implementations of your console application. It handles "
"configuration, lifecycle management, command processing, and user "
"interaction, coordinating the work of all components: routers, handlers, "
"and system messages."
@@ -109,7 +109,7 @@ msgstr "How Does It Work?"
#: ../../root/dependency_injection.rst:51
msgid "В основе DI в Argenta лежат **провайдеры** и **контейнер**."
msgstr "At the core of DI in Argenta are **providers** and a **container**."
msgstr "At the implementations of DI in Argenta are **providers** and a **container**."
#: ../../root/dependency_injection.rst:53
msgid ""
+1 -1
View File
@@ -49,7 +49,7 @@ msgstr ""
"The main purpose of flags is to provide a way to change the command's "
"logic without reworking it. A command can operate in several modes: "
"standard, verbose, debug, or simplified. Flags switch these modes on user"
" demand, keeping the core functionality unchanged."
" demand, keeping the implementations functionality unchanged."
#: ../../root/flags.rst:17
msgid "Опциональность и удобство"
+2 -2
View File
@@ -1,9 +1,9 @@
from argenta import App, Orchestrator
from argenta.app import DynamicDividingLine
from argenta.app import StaticDividingLine
from .handlers import router
app = App(initial_message="metrics", prompt=">>> ", dividing_line=DynamicDividingLine('~'))
app = App(initial_message="metrics", prompt=">>> ", dividing_line=StaticDividingLine('~', length=120))
orchestrator = Orchestrator()
+4 -1
View File
@@ -1 +1,4 @@
from .core import *
from .pre_cycle_setup import *
from .most_similar_command import *
from .finds_appropriate_handler import *
from .validate_routers_for_collisions import *
-4
View File
@@ -1,4 +0,0 @@
from .pre_cycle_setup import *
from .validate_routers_for_collisions import *
from .most_similar_command import *
from .finds_appropriate_handler import *
@@ -1,9 +1,12 @@
__all__ = [
"Benchmark",
"Benchmarks",
"BenchmarkResult"
"BenchmarkResult",
"BenchmarkGroupResult"
]
import io
from contextlib import redirect_stdout
from dataclasses import dataclass
import time
import gc
@@ -13,7 +16,8 @@ from typing import Callable, override
from .exceptions import BenchmarkNotFound, BenchmarksNotFound
BenchmarkAsFunc = Callable[[], float]
FuncForBenchmark = Callable[[], None]
MILLISECONDS_IN_SECONDS = 1000
@dataclass(frozen=True, slots=True)
@@ -37,7 +41,7 @@ class BenchmarkGroupResult:
class Benchmark:
def __init__(
self,
func: BenchmarkAsFunc,
func: FuncForBenchmark,
*,
type_: str,
name: str,
@@ -53,6 +57,7 @@ class Benchmark:
was_gc_enabled = gc.isenabled()
gc.disable()
with redirect_stdout(io.StringIO()):
start = time.perf_counter()
self.func()
end = time.perf_counter()
@@ -61,18 +66,19 @@ class Benchmark:
gc.enable()
gc.collect()
return end - start
return (end - start) * MILLISECONDS_IN_SECONDS
else:
with redirect_stdout(io.StringIO()):
start = time.perf_counter()
self.func()
end = time.perf_counter()
return end - start
return (end - start) * MILLISECONDS_IN_SECONDS
def multiple_runs(self, iterations: int, is_gc_disabled: bool = False) -> tuple[float]:
def multiple_runs(self, iterations: int, is_gc_disabled: bool = False) -> tuple[float, ...]:
run_attempts: list[float] = []
for _ in range(iterations):
run_attempts.append(self.single_run(is_gc_disabled))
return tuple(*run_attempts)
return tuple(run_attempts)
@override
def __repr__(self) -> str:
@@ -93,8 +99,8 @@ class Benchmarks:
self,
type_: str,
description: str = ""
) -> Callable[[BenchmarkAsFunc], BenchmarkAsFunc]:
def decorator(func: BenchmarkAsFunc) -> BenchmarkAsFunc:
) -> Callable[[FuncForBenchmark], FuncForBenchmark]:
def decorator(func: FuncForBenchmark) -> FuncForBenchmark:
benchmark = Benchmark(
func,
type_=type_,
@@ -102,7 +108,7 @@ class Benchmarks:
description=description or f'description for {func.__name__} with type {type_}',
)
self._benchmarks.append(benchmark)
self._benchmarks_paired_by_name[type_] = benchmark
self._benchmarks_paired_by_name[func.__name__] = benchmark
self._benchmarks_grouped_by_type.setdefault(type_, []).append(benchmark)
return func
return decorator
@@ -111,11 +117,11 @@ class Benchmarks:
benchmark = self.get_benchmark_by_name(name)
if not benchmark:
raise BenchmarkNotFound(name)
run_attempts: tuple[float] = benchmark.multiple_runs(iterations, is_gc_disables)
run_attempts: tuple[float, ...] = benchmark.multiple_runs(iterations, is_gc_disables)
avg = statistics.mean(run_attempts)
median = statistics.median(run_attempts)
std_dev = statistics.stdev(run_attempts) if len(run_attempts) > 1 else 0
avg = round(statistics.mean(run_attempts), 4)
median = round(statistics.median(run_attempts), 4)
std_dev = round(statistics.stdev(run_attempts) if len(run_attempts) > 1 else 0, 4)
return BenchmarkResult(
type_=benchmark.type_,
@@ -142,6 +148,12 @@ class Benchmarks:
benchmark_results=benchmark_results
)
def run_benchmarks_grouped_by_type(self) -> list[BenchmarkGroupResult]:
results: list[BenchmarkGroupResult] = []
for type_, benchmarks in self._benchmarks_grouped_by_type.items():
results.append(self.run_benchmarks_by_type(type_))
return results
def get_benchmarks_by_type(self, type_: str) -> list[Benchmark]:
return self._benchmarks_grouped_by_type.get(type_, [])
+3
View File
@@ -0,0 +1,3 @@
from metrics.benchmarks.core.models import Benchmarks
benchmarks = Benchmarks()
@@ -11,12 +11,11 @@ from argenta.command import Flag, Flags
from argenta.response import Response
from argenta.router import Router
from ..models import benchmark
from ..utils import get_time_of_finds_appropriate_handler
from .entity import benchmarks
@benchmark(type_="finds_appropriate_handler", description="Simple command (no flags)")
def benchmark_simple_command() -> float:
@benchmarks.register(type_="finds_appropriate_handler", description="Simple command (no flags)")
def benchmark_simple_command() -> None:
router = Router()
@router.command(Command('test'))
@@ -24,11 +23,11 @@ def benchmark_simple_command() -> float:
pass
input_cmd = InputCommand.parse('test')
return get_time_of_finds_appropriate_handler(router, input_cmd)
router.finds_appropriate_handler(input_cmd)
@benchmark(type_="finds_appropriate_handler", description="Command with flags (3 flags)")
def benchmark_command_with_flags() -> float:
@benchmarks.register(type_="finds_appropriate_handler", description="Command with flags (3 flags)")
def benchmark_command_with_flags() -> None:
router = Router()
@router.command(Command('test', flags=Flags([Flag('a'), Flag('b'), Flag('c')])))
@@ -36,11 +35,11 @@ def benchmark_command_with_flags() -> float:
pass
input_cmd = InputCommand.parse('test -a -b -c')
return get_time_of_finds_appropriate_handler(router, input_cmd)
router.finds_appropriate_handler(input_cmd)
@benchmark(type_="finds_appropriate_handler", description="Many commands (50 commands)")
def benchmark_many_commands() -> float:
@benchmarks.register(type_="finds_appropriate_handler", description="Many commands (50 commands)")
def benchmark_many_commands() -> None:
router = Router()
for i in range(50):
@@ -49,11 +48,11 @@ def benchmark_many_commands() -> float:
pass
input_cmd = InputCommand.parse('cmd25')
return get_time_of_finds_appropriate_handler(router, input_cmd)
router.finds_appropriate_handler(input_cmd)
@benchmark(type_="finds_appropriate_handler", description="Command with many flags (20 flags)")
def benchmark_command_with_many_flags() -> float:
@benchmarks.register(type_="finds_appropriate_handler", description="Command with many flags (20 flags)")
def benchmark_command_with_many_flags() -> None:
router = Router()
flags = Flags([Flag(f'flag{i}') for i in range(20)])
@@ -63,11 +62,11 @@ def benchmark_command_with_many_flags() -> float:
pass
input_cmd = InputCommand.parse('test ' + ' '.join(f'-flag{i}' for i in range(10)))
return get_time_of_finds_appropriate_handler(router, input_cmd)
router.finds_appropriate_handler(input_cmd)
@benchmark(type_="finds_appropriate_handler", description="Extreme (100 commands, 10 flags each)")
def benchmark_extreme_router() -> float:
@benchmarks.register(type_="finds_appropriate_handler", description="Extreme (100 commands, 10 flags each)")
def benchmark_extreme_router() -> None:
router = Router()
for i in range(100):
@@ -78,4 +77,4 @@ def benchmark_extreme_router() -> float:
pass
input_cmd = InputCommand.parse('cmd50 -f50_0 -f50_1 -f50_2')
return get_time_of_finds_appropriate_handler(router, input_cmd)
router.finds_appropriate_handler(input_cmd)
@@ -14,8 +14,7 @@ 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_most_similar_command
from .entity import benchmarks
def setup_app_with_commands(command_count: int, aliases_per_command: int = 0) -> App:
@@ -35,31 +34,31 @@ def setup_app_with_commands(command_count: int, aliases_per_command: int = 0) ->
return app
@benchmark(type_="most_similar_command", description="Few commands (10 commands, no match)")
def benchmark_few_commands() -> float:
@benchmarks.register(type_="most_similar_command", description="Few commands (10 commands, no match)")
def benchmark_few_commands() -> None:
app = setup_app_with_commands(10)
return get_time_of_most_similar_command(app, "unknown")
app._most_similar_command("unknown")
@benchmark(type_="most_similar_command", description="Many commands (50 commands, no match)")
def benchmark_many_commands() -> float:
@benchmarks.register(type_="most_similar_command", description="Many commands (50 commands, no match)")
def benchmark_many_commands() -> None:
app = setup_app_with_commands(50)
return get_time_of_most_similar_command(app, "unknown")
app._most_similar_command("unknown")
@benchmark(type_="most_similar_command", description="Many aliases (20 commands, 10 aliases each)")
def benchmark_many_aliases() -> float:
@benchmarks.register(type_="most_similar_command", description="Many aliases (20 commands, 10 aliases each)")
def benchmark_many_aliases() -> None:
app = setup_app_with_commands(20, aliases_per_command=10)
return get_time_of_most_similar_command(app, "unknown")
app._most_similar_command("unknown")
@benchmark(type_="most_similar_command", description="Partial match (50 commands, prefix match)")
def benchmark_partial_match() -> float:
@benchmarks.register(type_="most_similar_command", description="Partial match (50 commands, prefix match)")
def benchmark_partial_match() -> None:
app = setup_app_with_commands(50)
return get_time_of_most_similar_command(app, "comm")
app._most_similar_command("comm")
@benchmark(type_="most_similar_command", description="Extreme (100 commands, 20 aliases each)")
def benchmark_extreme_commands() -> float:
@benchmarks.register(type_="most_similar_command", description="Extreme (100 commands, 20 aliases each)")
def benchmark_extreme_commands() -> None:
app = setup_app_with_commands(100, aliases_per_command=20)
return get_time_of_most_similar_command(app, "comm")
app._most_similar_command("comm")
@@ -11,12 +11,11 @@ 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 .entity import benchmarks
@benchmark(type_="pre_cycle_setup", description="With no aliases")
def benchmark_no_aliases() -> float:
@benchmarks.register(type_="pre_cycle_setup", description="With no aliases")
def benchmark_no_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@@ -33,12 +32,11 @@ def benchmark_no_aliases() -> float:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
app._pre_cycle_setup()
@benchmark(type_="pre_cycle_setup", description="With few aliases (6 total)")
def benchmark_few_aliases() -> float:
@benchmarks.register(type_="pre_cycle_setup", description="With few aliases (6 total)")
def benchmark_few_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@@ -55,12 +53,11 @@ def benchmark_few_aliases() -> float:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
app._pre_cycle_setup()
@benchmark(type_="pre_cycle_setup", description="With many aliases (15 total)")
def benchmark_many_aliases() -> float:
@benchmarks.register(type_="pre_cycle_setup", description="With many aliases (15 total)")
def benchmark_many_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@@ -77,12 +74,11 @@ def benchmark_many_aliases() -> float:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
app._pre_cycle_setup()
@benchmark(type_="pre_cycle_setup", description="With very many aliases (60 total)")
def benchmark_very_many_aliases() -> float:
@benchmarks.register(type_="pre_cycle_setup", description="With very many aliases (60 total)")
def benchmark_very_many_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@@ -99,12 +95,11 @@ def benchmark_very_many_aliases() -> float:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
app._pre_cycle_setup()
@benchmark(type_="pre_cycle_setup", description="With extreme aliases (300 total)")
def benchmark_extreme_aliases() -> float:
@benchmarks.register(type_="pre_cycle_setup", description="With extreme aliases (300 total)")
def benchmark_extreme_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@@ -121,5 +116,4 @@ def benchmark_extreme_aliases() -> float:
pass
app.include_router(router)
execution_time = get_time_of_pre_cycle_setup(app)
return execution_time
app._pre_cycle_setup()
-110
View File
@@ -1,110 +0,0 @@
__all__ = [
"get_time_of_pre_cycle_setup",
"get_time_of_validate_routers_for_collisions",
"get_time_of_most_similar_command",
"get_time_of_finds_appropriate_handler",
"get_kernel_version",
"get_gpu_info"
]
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
import pynvml
from argenta import App
from argenta.router import Router
from argenta.command.models import InputCommand
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_time_of_validate_routers_for_collisions(app: App) -> float:
app._setup_system_router() # pyright: ignore[reportPrivateUsage]
start = time.perf_counter()
with redirect_stdout(io.StringIO()):
app._validate_routers_for_collisions() # pyright: ignore[reportPrivateUsage]
end = time.perf_counter()
return (end - start) * 1000
def get_time_of_most_similar_command(app: App, unknown_command: str) -> float:
start = time.perf_counter()
with redirect_stdout(io.StringIO()):
app._most_similar_command(unknown_command) # pyright: ignore[reportPrivateUsage]
end = time.perf_counter()
return (end - start) * 1000
def get_time_of_finds_appropriate_handler(router: Router, input_command: InputCommand) -> float:
start = time.perf_counter()
with redirect_stdout(io.StringIO()):
router.finds_appropriate_handler(input_command)
end = time.perf_counter()
return (end - start) * 1000
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"
@@ -11,13 +11,11 @@ from argenta.command.models import Command
from argenta.response import Response
from argenta.router import Router
from ..utils import get_time_of_validate_routers_for_collisions
from ..models import benchmark
from .entity import benchmarks
@benchmark(type_="validate_routers_for_collisions", description="With few routers (3 routers, 1 command each)")
def benchmark_few_routers() -> float:
@benchmarks.register(type_="validate_routers_for_collisions", description="With few routers (3 routers, 1 command each)")
def benchmark_few_routers() -> None:
app = App(override_system_messages=True)
for i in range(3):
@@ -29,11 +27,12 @@ def benchmark_few_routers() -> float:
app.include_router(router)
return get_time_of_validate_routers_for_collisions(app)
app._setup_system_router()
app._validate_routers_for_collisions()
@benchmark(type_="validate_routers_for_collisions", description="With many routers (10 routers, 1 command each)")
def benchmark_many_routers() -> float:
@benchmarks.register(type_="validate_routers_for_collisions", description="With many routers (10 routers, 1 command each)")
def benchmark_many_routers() -> None:
app = App(override_system_messages=True)
for i in range(10):
@@ -45,11 +44,12 @@ def benchmark_many_routers() -> float:
app.include_router(router)
return get_time_of_validate_routers_for_collisions(app)
app._setup_system_router()
app._validate_routers_for_collisions()
@benchmark(type_="validate_routers_for_collisions", description="With many commands per router (3 routers, 10 commands each)")
def benchmark_many_commands_per_router() -> float:
@benchmarks.register(type_="validate_routers_for_collisions", description="With many commands per router (3 routers, 10 commands each)")
def benchmark_many_commands_per_router() -> None:
app = App(override_system_messages=True)
for i in range(3):
@@ -62,11 +62,12 @@ def benchmark_many_commands_per_router() -> float:
app.include_router(router)
return get_time_of_validate_routers_for_collisions(app)
app._setup_system_router()
app._validate_routers_for_collisions()
@benchmark(type_="validate_routers_for_collisions", description="With many aliases (3 routers, 5 commands, 10 aliases each)")
def benchmark_many_aliases_per_command() -> float:
@benchmarks.register(type_="validate_routers_for_collisions", description="With many aliases (3 routers, 5 commands, 10 aliases each)")
def benchmark_many_aliases_per_command() -> None:
app = App(override_system_messages=True)
for i in range(3):
@@ -79,11 +80,12 @@ def benchmark_many_aliases_per_command() -> float:
app.include_router(router)
return get_time_of_validate_routers_for_collisions(app)
app._setup_system_router()
app._validate_routers_for_collisions()
@benchmark(type_="validate_routers_for_collisions", description="Extreme (20 routers, 10 commands, 20 aliases each)")
def benchmark_extreme_routers() -> float:
@benchmarks.register(type_="validate_routers_for_collisions", description="Extreme (20 routers, 10 commands, 20 aliases each)")
def benchmark_extreme_routers() -> None:
app = App(override_system_messages=True)
for i in range(20):
@@ -96,4 +98,5 @@ def benchmark_extreme_routers() -> float:
app.include_router(router)
return get_time_of_validate_routers_for_collisions(app)
app._setup_system_router()
app._validate_routers_for_collisions()
+12 -7
View File
@@ -9,8 +9,9 @@ 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, get_gpu_info
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:")
@@ -39,10 +40,10 @@ def all_print_handler(_: Response) -> None:
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()
type_grouped_benchmarks: list[BenchmarkGroupResult] = registered_benchmarks.run_benchmarks_grouped_by_type()
for type_, benchmarks in type_paired_benchmarks.items():
header_text = Text(f"TYPE: {type_.upper()}", style="bold magenta")
for results in type_grouped_benchmarks:
header_text = Text(f"TYPE: {results.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)
@@ -50,13 +51,17 @@ def all_print_handler(_: Response) -> None:
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")
for benchmark in benchmarks:
for benchmark in results.benchmark_results:
table.add_row(
benchmark.name,
benchmark.description,
str(benchmark.iterations),
str(benchmark.avg_time)
str(benchmark.avg_time),
str(benchmark.median_time),
str(benchmark.std_dev),
)
console.print(table)
+63
View File
@@ -0,0 +1,63 @@
__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"