ruff format

This commit is contained in:
2026-02-12 14:18:53 +03:00
parent de6d35205c
commit 732a4456b7
16 changed files with 394 additions and 289 deletions
+6 -6
View File
@@ -1,18 +1,18 @@
from argenta import App, Orchestrator, Command
from argenta.app import DynamicDividingLine
from argenta import App, Command, Orchestrator
from .handlers import router
app = App(initial_message="metrics", exit_command=Command('exit', aliases=['quit']))
app = App(initial_message="metrics", exit_command=Command("exit", aliases=["quit"]))
orchestrator = Orchestrator()
def main() -> None:
app.include_router(router)
app.set_description_message_pattern(
lambda command, description: f'[bold cyan]▸[/bold cyan] [bold white]{command}[/bold white] [dim]│[/dim] [yellow italic]{description}[/yellow italic]'
lambda command,
description: f"[bold cyan]▸[/bold cyan] [bold white]{command}[/bold white] [dim]│[/dim] [yellow italic]{description}[/yellow italic]"
)
orchestrator.run_repl(app)
# orchestrator.run_repl(app)
if __name__ == "__main__":
+2
View File
@@ -13,8 +13,10 @@ class BenchmarksNotFound(Exception):
def __str__(self) -> str:
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) -> str:
return f"Benchmarks with name '{self.benchmark_name}' already exists"
+25 -30
View File
@@ -1,16 +1,11 @@
__all__ = [
"Benchmark",
"Benchmarks",
"BenchmarkResult",
"BenchmarkGroupResult"
]
__all__ = ["Benchmark", "Benchmarks", "BenchmarkResult", "BenchmarkGroupResult"]
import gc
import io
import statistics
import time
from contextlib import redirect_stdout
from dataclasses import dataclass
import time
import gc
import statistics
from typing import Callable, override
from .exceptions import BenchmarkNotFound, BenchmarksNotFound, BenchmarksWithSameNameAlreadyExists
@@ -40,14 +35,7 @@ class BenchmarkGroupResult:
class Benchmark:
def __init__(
self,
func: FuncForBenchmark,
*,
type_: str,
name: str,
description: str
) -> None:
def __init__(self, func: FuncForBenchmark, *, type_: str, name: str, description: str) -> None:
self.func = func
self.type_ = type_
self.name = name
@@ -78,11 +66,11 @@ class Benchmark:
@override
def __repr__(self) -> str:
return f'Benchmark<{self.type_=}, {self.name=}, {self.description=}>'
return f"Benchmark<{self.type_=}, {self.name=}, {self.description=}>"
@override
def __str__(self) -> str:
return f'benchmark {self.name} with type {self.type_}'
return f"benchmark {self.name} with type {self.type_}"
class Benchmarks:
@@ -92,16 +80,14 @@ class Benchmarks:
self._benchmarks_paired_by_name: dict[str, Benchmark] = {}
def register(
self,
type_: str,
description: str = ""
self, type_: str, description: str = ""
) -> Callable[[FuncForBenchmark], FuncForBenchmark]:
def decorator(func: FuncForBenchmark) -> FuncForBenchmark:
benchmark = Benchmark(
func,
type_=type_,
name=func.__name__,
description=description or f'description for {func.__name__} with type {type_}',
description=description or f"description for {func.__name__} with type {type_}",
)
if self._benchmarks_paired_by_name.get(func.__name__):
raise BenchmarksWithSameNameAlreadyExists(func.__name__)
@@ -110,9 +96,12 @@ class Benchmarks:
self._benchmarks.append(benchmark)
self._benchmarks_grouped_by_type.setdefault(type_, []).append(benchmark)
return func
return decorator
def run_benchmark_by_name(self, name: str, iterations: int = 100, is_gc_disables: bool = False) -> BenchmarkResult:
def run_benchmark_by_name(
self, name: str, iterations: int = 100, is_gc_disables: bool = False
) -> BenchmarkResult:
benchmark = self.get_benchmark_by_name(name)
if not benchmark:
raise BenchmarkNotFound(name)
@@ -130,28 +119,34 @@ class Benchmarks:
is_gc_disabled=is_gc_disables,
avg_time=avg,
median_time=median,
std_dev=std_dev
std_dev=std_dev,
)
def run_benchmarks_by_type(self, type_: str, iterations: int = 100, is_gc_disabled: bool = False) -> BenchmarkGroupResult:
def run_benchmarks_by_type(
self, type_: str, iterations: int = 100, is_gc_disabled: bool = False
) -> BenchmarkGroupResult:
benchmarks = self.get_benchmarks_by_type(type_)
if not benchmarks:
raise BenchmarksNotFound(type_)
benchmark_results: list[BenchmarkResult] = []
for benchmark in benchmarks:
benchmark_results.append(self.run_benchmark_by_name(benchmark.name, iterations, is_gc_disabled))
benchmark_results.append(
self.run_benchmark_by_name(benchmark.name, iterations, is_gc_disabled)
)
return BenchmarkGroupResult(
type_=type_,
iterations=iterations,
is_gc_disabled=is_gc_disabled,
benchmark_results=benchmark_results
benchmark_results=benchmark_results,
)
def run_benchmarks_grouped_by_type(self, iterations: int = 100, is_gc_disabled: bool = False) -> list[BenchmarkGroupResult]:
def run_benchmarks_grouped_by_type(
self, iterations: int = 100, is_gc_disabled: bool = False
) -> list[BenchmarkGroupResult]:
results: list[BenchmarkGroupResult] = []
for type_, benchmarks in self._benchmarks_grouped_by_type.items():
for type_, _ in self._benchmarks_grouped_by_type.items():
results.append(self.run_benchmarks_by_type(type_, iterations, is_gc_disabled))
return results
+21 -16
View File
@@ -3,11 +3,11 @@ __all__ = [
"benchmark_command_with_flags",
"benchmark_many_commands",
"benchmark_command_with_many_flags",
"benchmark_extreme_router"
"benchmark_extreme_router",
]
from argenta.command.models import Command, InputCommand
from argenta.command import Flag, Flags
from argenta.command.models import Command, InputCommand
from argenta.response import Response
from argenta.router import Router
@@ -18,11 +18,11 @@ from .entity import benchmarks
def benchmark_simple_command() -> None:
router = Router()
@router.command(Command('test'))
@router.command(Command("test"))
def handler(_res: Response) -> None:
pass
input_cmd = InputCommand.parse('test')
input_cmd = InputCommand.parse("test")
router.finds_appropriate_handler(input_cmd)
@@ -30,11 +30,11 @@ def benchmark_simple_command() -> None:
def benchmark_command_with_flags() -> None:
router = Router()
@router.command(Command('test', flags=Flags([Flag('a'), Flag('b'), Flag('c')])))
@router.command(Command("test", flags=Flags([Flag("a"), Flag("b"), Flag("c")])))
def handler(_res: Response) -> None:
pass
input_cmd = InputCommand.parse('test -a -b -c')
input_cmd = InputCommand.parse("test -a -b -c")
router.finds_appropriate_handler(input_cmd)
@@ -43,38 +43,43 @@ def benchmark_many_commands() -> None:
router = Router()
for i in range(50):
@router.command(Command(f'cmd{i}'))
@router.command(Command(f"cmd{i}"))
def handler(_res: Response) -> None:
pass
input_cmd = InputCommand.parse('cmd25')
input_cmd = InputCommand.parse("cmd25")
router.finds_appropriate_handler(input_cmd)
@benchmarks.register(type_="finds_appropriate_handler", description="Command with many flags (20 flags)")
@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)])
flags = Flags([Flag(f"flag{i}") for i in range(20)])
@router.command(Command('test', flags=flags))
@router.command(Command("test", flags=flags))
def handler(_res: Response) -> None:
pass
input_cmd = InputCommand.parse('test ' + ' '.join(f'-flag{i}' for i in range(10)))
input_cmd = InputCommand.parse("test " + " ".join(f"-flag{i}" for i in range(10)))
router.finds_appropriate_handler(input_cmd)
@benchmarks.register(type_="finds_appropriate_handler", description="Extreme (100 commands, 10 flags each)")
@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):
flags = Flags([Flag(f'f{i}_{j}') for j in range(10)])
flags = Flags([Flag(f"f{i}_{j}") for j in range(10)])
@router.command(Command(f'cmd{i}', flags=flags))
@router.command(Command(f"cmd{i}", flags=flags))
def handler(_res: Response) -> None:
pass
input_cmd = InputCommand.parse('cmd50 -f50_0 -f50_1 -f50_2')
input_cmd = InputCommand.parse("cmd50 -f50_0 -f50_1 -f50_2")
router.finds_appropriate_handler(input_cmd)
+10 -26
View File
@@ -7,7 +7,7 @@ __all__ = [
"benchmark_validate_regex_complex",
"benchmark_validate_multiple_flags_10",
"benchmark_validate_multiple_flags_50",
"benchmark_validate_extreme_100_flags"
"benchmark_validate_extreme_100_flags",
]
import re
@@ -58,14 +58,8 @@ def benchmark_validate_regex_complex() -> None:
@benchmarks.register(type_="flag_validation", description="Multiple flags validation (10 flags)")
def benchmark_validate_multiple_flags_10() -> None:
flags = [
Flag(f"flag{i}", possible_values=PossibleValues.ALL)
for i in range(10)
]
input_flags = [
InputFlag(f"flag{i}", input_value=f"value{i}")
for i in range(10)
]
flags = [Flag(f"flag{i}", possible_values=PossibleValues.ALL) for i in range(10)]
input_flags = [InputFlag(f"flag{i}", input_value=f"value{i}") for i in range(10)]
for flag, input_flag in zip(flags, input_flags):
flag.validate_input_flag_value(input_flag.input_value)
@@ -73,30 +67,20 @@ def benchmark_validate_multiple_flags_10() -> None:
@benchmarks.register(type_="flag_validation", description="Multiple flags validation (50 flags)")
def benchmark_validate_multiple_flags_50() -> None:
flags = [
Flag(f"flag{i}", possible_values=PossibleValues.ALL)
for i in range(50)
]
input_flags = [
InputFlag(f"flag{i}", input_value=f"value{i}")
for i in range(50)
]
flags = [Flag(f"flag{i}", possible_values=PossibleValues.ALL) for i in range(50)]
input_flags = [InputFlag(f"flag{i}", input_value=f"value{i}") for i in range(50)]
for flag, input_flag in zip(flags, input_flags):
flag.validate_input_flag_value(input_flag.input_value)
@benchmarks.register(type_="flag_validation", description="Extreme (100 flags with regex validation)")
@benchmarks.register(
type_="flag_validation", description="Extreme (100 flags with regex validation)"
)
def benchmark_validate_extreme_100_flags() -> None:
pattern = re.compile(r"^[a-zA-Z0-9_-]+$")
flags = [
Flag(f"flag{i}", possible_values=pattern)
for i in range(100)
]
input_flags = [
InputFlag(f"flag{i}", input_value=f"valid_value_{i}")
for i in range(100)
]
flags = [Flag(f"flag{i}", possible_values=pattern) for i in range(100)]
input_flags = [InputFlag(f"flag{i}", input_value=f"valid_value_{i}") for i in range(100)]
for flag, input_flag in zip(flags, input_flags):
flag.validate_input_flag_value(input_flag.input_value)
+10 -4
View File
@@ -5,7 +5,7 @@ __all__ = [
"benchmark_command_with_mixed_prefixes",
"benchmark_command_with_long_values",
"benchmark_command_with_quoted_values",
"benchmark_extreme_many_flags"
"benchmark_extreme_many_flags",
]
from argenta.command.models import InputCommand
@@ -23,12 +23,16 @@ def benchmark_command_with_few_flags() -> None:
InputCommand.parse("start -a -b -c")
@benchmarks.register(type_="input_command_parse", description="Command with flags and values (5 flags)")
@benchmarks.register(
type_="input_command_parse", description="Command with flags and values (5 flags)"
)
def benchmark_command_with_flags_and_values() -> None:
InputCommand.parse("start --host localhost --port 8080 --debug --verbose -c config.json")
@benchmarks.register(type_="input_command_parse", description="Command with mixed prefixes (-, --, ---)")
@benchmarks.register(
type_="input_command_parse", description="Command with mixed prefixes (-, --, ---)"
)
def benchmark_command_with_mixed_prefixes() -> None:
InputCommand.parse("cmd -a --bb ---ccc -d value --ee value2 ---fff value3")
@@ -40,7 +44,9 @@ def benchmark_command_with_long_values() -> None:
InputCommand.parse(cmd)
@benchmarks.register(type_="input_command_parse", description="Command with quoted values (5 flags)")
@benchmarks.register(
type_="input_command_parse", description="Command with quoted values (5 flags)"
)
def benchmark_command_with_quoted_values() -> None:
InputCommand.parse("cmd --text 'hello world' --path '/usr/local/bin' --msg \"test message\"")
+20 -8
View File
@@ -3,7 +3,7 @@ __all__ = [
"benchmark_many_commands_most_similar",
"benchmark_many_aliases",
"benchmark_partial_match",
"benchmark_extreme_commands"
"benchmark_extreme_commands",
]
from argenta import App
@@ -19,9 +19,11 @@ def setup_app_with_commands(command_count: int, aliases_per_command: int = 0) ->
router = Router()
for i in range(command_count):
aliases = {f'alias{i}_{j}' for j in range(aliases_per_command)} if aliases_per_command else set()
aliases = (
{f"alias{i}_{j}" for j in range(aliases_per_command)} if aliases_per_command else set()
)
@router.command(Command(f'command{i}', aliases=aliases))
@router.command(Command(f"command{i}", aliases=aliases))
def handler(_res: Response) -> None:
pass
@@ -29,31 +31,41 @@ def setup_app_with_commands(command_count: int, aliases_per_command: int = 0) ->
return app
@benchmarks.register(type_="most_similar_command", description="Few commands (10 commands, no match)")
@benchmarks.register(
type_="most_similar_command", description="Few commands (10 commands, no match)"
)
def benchmark_few_commands() -> None:
app = setup_app_with_commands(10)
app._most_similar_command("unknown")
@benchmarks.register(type_="most_similar_command", description="Many commands (50 commands, no match)")
@benchmarks.register(
type_="most_similar_command", description="Many commands (50 commands, no match)"
)
def benchmark_many_commands_most_similar() -> None:
app = setup_app_with_commands(50)
app._most_similar_command("unknown")
@benchmarks.register(type_="most_similar_command", description="Many aliases (20 commands, 10 aliases each)")
@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)
app._most_similar_command("unknown")
@benchmarks.register(type_="most_similar_command", description="Partial match (50 commands, prefix match)")
@benchmarks.register(
type_="most_similar_command", description="Partial match (50 commands, prefix match)"
)
def benchmark_partial_match() -> None:
app = setup_app_with_commands(50)
app._most_similar_command("comm")
@benchmarks.register(type_="most_similar_command", description="Extreme (100 commands, 20 aliases each)")
@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)
app._most_similar_command("comm")
+16 -16
View File
@@ -3,7 +3,7 @@ __all__ = [
"benchmark_with_many_aliases",
"benchmark_few_aliases",
"benchmark_extreme_aliases",
"benchmark_very_many_aliases"
"benchmark_very_many_aliases",
]
from argenta import App
@@ -19,15 +19,15 @@ def benchmark_no_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1'))
@router.command(Command("command1"))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2'))
@router.command(Command("command2"))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3'))
@router.command(Command("command3"))
def handler3(_res: Response) -> None:
pass
@@ -40,15 +40,15 @@ def benchmark_few_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1', aliases={'c1', 'cmd1'}))
@router.command(Command("command1", aliases={"c1", "cmd1"}))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2', aliases={'c2', 'cmd2'}))
@router.command(Command("command2", aliases={"c2", "cmd2"}))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3', aliases={'c3', 'cmd3'}))
@router.command(Command("command3", aliases={"c3", "cmd3"}))
def handler3(_res: Response) -> None:
pass
@@ -61,15 +61,15 @@ def benchmark_with_many_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1', aliases={'c1', 'cmd1', 'com1', 'first', 'one'}))
@router.command(Command("command1", aliases={"c1", "cmd1", "com1", "first", "one"}))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2', aliases={'c2', 'cmd2', 'com2', 'second', 'two'}))
@router.command(Command("command2", aliases={"c2", "cmd2", "com2", "second", "two"}))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3', aliases={'c3', 'cmd3', 'com3', 'third', 'three'}))
@router.command(Command("command3", aliases={"c3", "cmd3", "com3", "third", "three"}))
def handler3(_res: Response) -> None:
pass
@@ -82,15 +82,15 @@ def benchmark_very_many_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1', aliases={f'alias1_{i}' for i in range(20)}))
@router.command(Command("command1", aliases={f"alias1_{i}" for i in range(20)}))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2', aliases={f'alias2_{i}' for i in range(20)}))
@router.command(Command("command2", aliases={f"alias2_{i}" for i in range(20)}))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3', aliases={f'alias3_{i}' for i in range(20)}))
@router.command(Command("command3", aliases={f"alias3_{i}" for i in range(20)}))
def handler3(_res: Response) -> None:
pass
@@ -103,15 +103,15 @@ def benchmark_extreme_aliases() -> None:
app = App(override_system_messages=True)
router = Router()
@router.command(Command('command1', aliases={f'alias1_{i}' for i in range(100)}))
@router.command(Command("command1", aliases={f"alias1_{i}" for i in range(100)}))
def handler1(_res: Response) -> None:
pass
@router.command(Command('command2', aliases={f'alias2_{i}' for i in range(100)}))
@router.command(Command("command2", aliases={f"alias2_{i}" for i in range(100)}))
def handler2(_res: Response) -> None:
pass
@router.command(Command('command3', aliases={f'alias3_{i}' for i in range(100)}))
@router.command(Command("command3", aliases={f"alias3_{i}" for i in range(100)}))
def handler3(_res: Response) -> None:
pass
@@ -3,7 +3,7 @@ __all__ = [
"benchmark_many_routers",
"benchmark_many_commands_per_router",
"benchmark_many_aliases_per_command",
"benchmark_extreme_routers"
"benchmark_extreme_routers",
]
from argenta import App
@@ -14,14 +14,17 @@ from argenta.router import Router
from .entity import benchmarks
@benchmarks.register(type_="validate_routers_for_collisions", description="With few routers (3 routers, 1 command each)")
@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):
router = Router()
@router.command(Command(f'cmd{i}'))
@router.command(Command(f"cmd{i}"))
def handler(_res: Response) -> None:
pass
@@ -31,14 +34,17 @@ def benchmark_few_routers() -> None:
app._validate_routers_for_collisions()
@benchmarks.register(type_="validate_routers_for_collisions", description="With many routers (10 routers, 1 command each)")
@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):
router = Router()
@router.command(Command(f'cmd{i}'))
@router.command(Command(f"cmd{i}"))
def handler(_res: Response) -> None:
pass
@@ -48,7 +54,10 @@ def benchmark_many_routers() -> None:
app._validate_routers_for_collisions()
@benchmarks.register(type_="validate_routers_for_collisions", description="With many commands per router (3 routers, 10 commands each)")
@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)
@@ -56,7 +65,8 @@ def benchmark_many_commands_per_router() -> None:
router = Router()
for j in range(10):
@router.command(Command(f'cmd{i}_{j}'))
@router.command(Command(f"cmd{i}_{j}"))
def handler(_res: Response) -> None:
pass
@@ -66,7 +76,10 @@ def benchmark_many_commands_per_router() -> None:
app._validate_routers_for_collisions()
@benchmarks.register(type_="validate_routers_for_collisions", description="With many aliases (3 routers, 5 commands, 10 aliases each)")
@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)
@@ -74,7 +87,10 @@ def benchmark_many_aliases_per_command() -> None:
router = Router()
for j in range(5):
@router.command(Command(f'cmd{i}_{j}', aliases={f'alias{i}_{j}_{k}' for k in range(10)}))
@router.command(
Command(f"cmd{i}_{j}", aliases={f"alias{i}_{j}_{k}" for k in range(10)})
)
def handler(_res: Response) -> None:
pass
@@ -84,7 +100,10 @@ def benchmark_many_aliases_per_command() -> None:
app._validate_routers_for_collisions()
@benchmarks.register(type_="validate_routers_for_collisions", description="Extreme (20 routers, 10 commands, 20 aliases each)")
@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)
@@ -92,7 +111,10 @@ def benchmark_extreme_routers() -> None:
router = Router()
for j in range(10):
@router.command(Command(f'cmd{i}_{j}', aliases={f'alias{i}_{j}_{k}' for k in range(20)}))
@router.command(
Command(f"cmd{i}_{j}", aliases={f"alias{i}_{j}_{k}" for k in range(20)})
)
def handler(_res: Response) -> None:
pass
+56 -33
View File
@@ -5,17 +5,18 @@ from pathlib import Path
from rich.console import Console
from argenta.command import Flag, PossibleValues, Flags
from argenta.command import Flag, Flags, PossibleValues
from argenta.command.flag import ValidationStatus
from argenta.command.models import Command
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 .services.report_table_generator import ReportTableGenerator
from .services.system_info_reader import get_system_info
from .services.diagram_generator import DiagramGenerator
from .services.release_generator import ReleaseGenerator
from .services.report_table_generator import ReportTableGenerator
from .services.system_info_reader import get_system_info
console = Console()
router = Router(title="Metrics commands:", disable_redirect_stdout=True)
@@ -27,22 +28,30 @@ POSITIVE_INTEGER_PATTERN = re.compile(r"^[1-9]\d*$")
Command(
"run-all",
description="Print all benchmarks results",
flags=Flags([
Flag('without-gc', possible_values=PossibleValues.NEITHER),
Flag('without-system-info', possible_values=PossibleValues.NEITHER)
])
flags=Flags(
[
Flag("without-gc", possible_values=PossibleValues.NEITHER),
Flag("without-system-info", possible_values=PossibleValues.NEITHER),
]
),
)
)
def all_print_handler(response: Response) -> None:
report_generator = ReportTableGenerator(get_system_info())
without_system_info = response.input_flags.get_flag_by_name("without-system-info", with_status=ValidationStatus.VALID)
without_system_info = response.input_flags.get_flag_by_name(
"without-system-info", with_status=ValidationStatus.VALID
)
if not without_system_info:
console.print(report_generator.generate_system_info_header())
console.print(report_generator.generate_system_info_table())
is_gc_disabled = response.input_flags.get_flag_by_name("without-gc", with_status=ValidationStatus.VALID)
type_grouped_benchmarks: list[BenchmarkGroupResult] = registered_benchmarks.run_benchmarks_grouped_by_type(is_gc_disabled=bool(is_gc_disabled))
is_gc_disabled = response.input_flags.get_flag_by_name(
"without-gc", with_status=ValidationStatus.VALID
)
type_grouped_benchmarks: list[BenchmarkGroupResult] = (
registered_benchmarks.run_benchmarks_grouped_by_type(is_gc_disabled=bool(is_gc_disabled))
)
for benchmark_group_result in type_grouped_benchmarks:
console.print(report_generator.generate_benchmark_table_header(benchmark_group_result))
@@ -67,11 +76,13 @@ def list_types_handler(_: Response) -> None:
Command(
"run-type",
description="Run benchmarks by specific type",
flags=Flags([
Flag('type', possible_values=registered_benchmarks.get_types()),
Flag('without-gc', possible_values=PossibleValues.NEITHER),
Flag('without-system-info', possible_values=PossibleValues.NEITHER)
])
flags=Flags(
[
Flag("type", possible_values=registered_benchmarks.get_types()),
Flag("without-gc", possible_values=PossibleValues.NEITHER),
Flag("without-system-info", possible_values=PossibleValues.NEITHER),
]
),
)
)
def run_type_handler(response: Response) -> None:
@@ -94,13 +105,19 @@ def run_type_handler(response: Response) -> None:
report_generator = ReportTableGenerator(get_system_info())
without_system_info = response.input_flags.get_flag_by_name("without-system-info", with_status=ValidationStatus.VALID)
without_system_info = response.input_flags.get_flag_by_name(
"without-system-info", with_status=ValidationStatus.VALID
)
if not without_system_info:
console.print(report_generator.generate_system_info_header())
console.print(report_generator.generate_system_info_table())
is_gc_disabled = response.input_flags.get_flag_by_name("without-gc", with_status=ValidationStatus.VALID, default=False)
benchmark_group_result = registered_benchmarks.run_benchmarks_by_type(benchmark_type, is_gc_disabled=bool(is_gc_disabled))
is_gc_disabled = response.input_flags.get_flag_by_name(
"without-gc", with_status=ValidationStatus.VALID, default=False
)
benchmark_group_result = registered_benchmarks.run_benchmarks_by_type(
benchmark_type, is_gc_disabled=bool(is_gc_disabled)
)
console.print(report_generator.generate_benchmark_table_header(benchmark_group_result))
console.print(report_generator.generate_benchmark_report_table(benchmark_group_result))
@@ -113,22 +130,21 @@ def release_generate_handler(_: Response) -> None:
console.print(f"[cyan]Generating release report for version:[/cyan] [bold]{lib_version}[/bold]")
console.print("[dim]Running benchmarks (1000 iterations, GC disabled)...[/dim]\n")
type_grouped_benchmarks: list[BenchmarkGroupResult] = registered_benchmarks.run_benchmarks_grouped_by_type(
iterations=1000,
is_gc_disabled=True
type_grouped_benchmarks: list[BenchmarkGroupResult] = (
registered_benchmarks.run_benchmarks_grouped_by_type(iterations=1000, is_gc_disabled=True)
)
release_generator = ReleaseGenerator(lib_version)
output_dir = release_generator.generate_release(type_grouped_benchmarks)
console.print(f"[green]✓[/green] Benchmarks completed. Generating release report...\n")
console.print("[green]✓[/green] Benchmarks completed. Generating release report...\n")
for benchmark_group in type_grouped_benchmarks:
console.print(f"[cyan]Generated for:[/cyan] [bold]{benchmark_group.type_}[/bold]")
console.print(f" [green]✓[/green] {benchmark_group.type_}_comparison.png")
console.print(f" [green]✓[/green] {benchmark_group.type_}.json\n")
console.print(f"[bold green]✓ Release report generated successfully[/bold green]")
console.print("[bold green]✓ Release report generated successfully[/bold green]")
console.print(f"[cyan]Output directory:[/cyan] [bold]{output_dir}[/bold]")
@@ -136,26 +152,33 @@ def release_generate_handler(_: Response) -> None:
Command(
"diagrams-generate",
description="Generate diagrams for all benchmarks",
flags=Flags([
Flag('without-gc', possible_values=PossibleValues.NEITHER),
Flag('iterations', possible_values=POSITIVE_INTEGER_PATTERN)
])
flags=Flags(
[
Flag("without-gc", possible_values=PossibleValues.NEITHER),
Flag("iterations", possible_values=POSITIVE_INTEGER_PATTERN),
]
),
)
)
def diagrams_generate_handler(response: Response) -> None:
iterations = 100
iterations_flag = response.input_flags.get_flag_by_name("iterations", with_status=ValidationStatus.VALID)
iterations_flag = response.input_flags.get_flag_by_name(
"iterations", with_status=ValidationStatus.VALID
)
if iterations_flag:
iterations = int(iterations_flag.input_value)
is_gc_disabled = bool(response.input_flags.get_flag_by_name("without-gc", with_status=ValidationStatus.VALID))
is_gc_disabled = bool(
response.input_flags.get_flag_by_name("without-gc", with_status=ValidationStatus.VALID)
)
console.print("[cyan]Running all benchmarks...[/cyan]")
console.print(f"[dim]Iterations: {iterations}, GC Disabled: {is_gc_disabled}[/dim]\n")
type_grouped_benchmarks: list[BenchmarkGroupResult] = registered_benchmarks.run_benchmarks_grouped_by_type(
iterations=iterations,
is_gc_disabled=is_gc_disabled
type_grouped_benchmarks: list[BenchmarkGroupResult] = (
registered_benchmarks.run_benchmarks_grouped_by_type(
iterations=iterations, is_gc_disabled=is_gc_disabled
)
)
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
@@ -164,7 +187,7 @@ def diagrams_generate_handler(response: Response) -> None:
diagram_generator = DiagramGenerator(output_dir)
console.print(f"[green]✓[/green] Benchmarks completed. Generating diagrams...\n")
console.print("[green]✓[/green] Benchmarks completed. Generating diagrams...\n")
generated_count = 0
+1 -1
View File
@@ -1,6 +1,6 @@
from .diagram_generator import DiagramGenerator
from .release_generator import ReleaseGenerator
from .report_table_generator import ReportTableGenerator
from .system_info_reader import get_system_info
from .release_generator import ReleaseGenerator
__all__ = ["DiagramGenerator", "ReportTableGenerator", "get_system_info", "ReleaseGenerator"]
+83 -31
View File
@@ -12,8 +12,8 @@ class DiagramGenerator:
def __init__(self, output_dir: Path | str) -> None:
self.output_dir: Path = Path(output_dir) if isinstance(output_dir, str) else output_dir
matplotlib.use('Agg')
plt.style.use('seaborn-v0_8-whitegrid')
matplotlib.use("Agg")
plt.style.use("seaborn-v0_8-whitegrid")
def generate_comparison_diagram(self, benchmark_group: BenchmarkGroupResult) -> Path:
results = benchmark_group.benchmark_results
@@ -27,7 +27,7 @@ class DiagramGenerator:
max_value = max(
max(avg_times) if avg_times else 0,
max(median_times) if median_times else 0,
max(std_devs) if std_devs else 0
max(std_devs) if std_devs else 0,
)
y_limit = max_value / 0.85 if max_value > 0 else 1.0
@@ -41,34 +41,77 @@ class DiagramGenerator:
x_median = [x + bar_width for x in x_positions]
fig, ax = plt.subplots(figsize=(16, 8))
fig.patch.set_facecolor('white')
fig.patch.set_facecolor("white")
bars_std = ax.bar(x_std_dev, std_devs, bar_width, label='Std Deviation',
color='#2ecc71', alpha=0.9, edgecolor='#27ae60', linewidth=1.5)
bars_avg = ax.bar(x_avg, avg_times, bar_width, label='Average Time',
color='#3498db', alpha=0.9, edgecolor='#2980b9', linewidth=1.5)
bars_median = ax.bar(x_median, median_times, bar_width, label='Median Time',
color='#e74c3c', alpha=0.9, edgecolor='#c0392b', linewidth=1.5)
bars_std = ax.bar(
x_std_dev,
std_devs,
bar_width,
label="Std Deviation",
color="#2ecc71",
alpha=0.9,
edgecolor="#27ae60",
linewidth=1.5,
)
bars_avg = ax.bar(
x_avg,
avg_times,
bar_width,
label="Average Time",
color="#3498db",
alpha=0.9,
edgecolor="#2980b9",
linewidth=1.5,
)
bars_median = ax.bar(
x_median,
median_times,
bar_width,
label="Median Time",
color="#e74c3c",
alpha=0.9,
edgecolor="#c0392b",
linewidth=1.5,
)
for bar_group in [bars_std, bars_avg, bars_median]:
for bar in bar_group:
height = bar.get_height()
ax.text(
bar.get_x() + bar.get_width() / 2.,
bar.get_x() + bar.get_width() / 2.0,
height,
f'{height:.3f}',
ha='center', va='bottom', fontsize=9, fontweight='bold'
f"{height:.3f}",
ha="center",
va="bottom",
fontsize=9,
fontweight="bold",
)
ax.set_ylabel('Time (ms)', fontsize=14, fontweight='bold', labelpad=10)
ax.set_ylabel("Time (ms)", fontsize=14, fontweight="bold", labelpad=10)
title_text = f'{benchmark_group.type_.replace("_", " ").title()}'
metadata_text = f'Iterations: {benchmark_group.iterations} | GC: {"Disabled" if benchmark_group.is_gc_disabled else "Enabled"}'
title_text = f"{benchmark_group.type_.replace('_', ' ').title()}"
metadata_text = f"Iterations: {benchmark_group.iterations} | GC: {'Disabled' if benchmark_group.is_gc_disabled else 'Enabled'}"
ax.text(0.5, 1.08, title_text, transform=ax.transAxes,
fontsize=18, fontweight='bold', ha='center', color='#2c3e50')
ax.text(0.5, 1.03, metadata_text, transform=ax.transAxes,
fontsize=12, ha='center', color='#7f8c8d', style='italic')
ax.text(
0.5,
1.08,
title_text,
transform=ax.transAxes,
fontsize=18,
fontweight="bold",
ha="center",
color="#2c3e50",
)
ax.text(
0.5,
1.03,
metadata_text,
transform=ax.transAxes,
fontsize=12,
ha="center",
color="#7f8c8d",
style="italic",
)
ax.set_xticks(x_positions)
ax.set_xticklabels([])
@@ -79,23 +122,32 @@ class DiagramGenerator:
text_x_pos,
y_limit * 0.02,
desc,
rotation=90, va='bottom', ha='right', fontsize=10,
color='#2c3e50'
rotation=90,
va="bottom",
ha="right",
fontsize=10,
color="#2c3e50",
)
ax.set_ylim(0, y_limit)
legend = ax.legend(loc='upper left', fontsize=12, framealpha=0.95,
edgecolor='#34495e', fancybox=True, shadow=True)
legend.get_frame().set_facecolor('#ecf0f1')
legend = ax.legend(
loc="upper left",
fontsize=12,
framealpha=0.95,
edgecolor="#34495e",
fancybox=True,
shadow=True,
)
legend.get_frame().set_facecolor("#ecf0f1")
ax.grid(axis='y', alpha=0.4, linestyle='--', linewidth=0.8)
ax.grid(axis="y", alpha=0.4, linestyle="--", linewidth=0.8)
ax.set_axisbelow(True)
ax.spines['top'].set_visible(False)
ax.spines['right'].set_visible(False)
ax.spines['left'].set_color('#7f8c8d')
ax.spines['bottom'].set_color('#7f8c8d')
ax.spines["top"].set_visible(False)
ax.spines["right"].set_visible(False)
ax.spines["left"].set_color("#7f8c8d")
ax.spines["bottom"].set_color("#7f8c8d")
plt.tight_layout()
@@ -104,7 +156,7 @@ class DiagramGenerator:
self.output_dir.mkdir(parents=True, exist_ok=True)
plt.savefig(output_path, dpi=200, bbox_inches='tight', facecolor='white')
plt.savefig(output_path, dpi=200, bbox_inches="tight", facecolor="white")
plt.close(fig)
return output_path
+3 -3
View File
@@ -36,14 +36,14 @@ class ReleaseGenerator:
"description": br.description,
"avg_time": br.avg_time,
"median_time": br.median_time,
"std_dev": br.std_dev
"std_dev": br.std_dev,
}
for br in benchmark_group.benchmark_results
]
],
}
json_path = type_dir / f"{benchmark_group.type_}.json"
with open(json_path, 'w', encoding='utf-8') as f:
with open(json_path, "w", encoding="utf-8") as f:
json.dump(json_data, f, indent=2, ensure_ascii=False)
return self.output_dir
+19 -11
View File
@@ -3,7 +3,7 @@ from rich.table import Table
from rich.text import Text
from ..benchmarks.core.models import BenchmarkGroupResult
from metrics.services.system_info_reader import SystemInfo
from .system_info_reader import SystemInfo
class ReportTableGenerator:
@@ -12,11 +12,15 @@ class ReportTableGenerator:
self._cached_benchmark_tables: dict[int, Table] = {}
self._cached_system_info_table: Table | None = None
def generate_benchmark_report_table(self, benchmark_group_result: BenchmarkGroupResult) -> Table:
def generate_benchmark_report_table(
self, benchmark_group_result: BenchmarkGroupResult
) -> Table:
if cached_result := self._cached_benchmark_tables.get(id(benchmark_group_result)):
return cached_result
table = Table(show_header=True, header_style="bold cyan", border_style="blue", show_lines=True)
table = Table(
show_header=True, header_style="bold cyan", border_style="blue", show_lines=True
)
table.add_column("Description", style="dim")
table.add_column("Avg Time", justify="right", style="bold yellow")
table.add_column("Median Time", justify="right", style="bold yellow")
@@ -34,18 +38,22 @@ class ReportTableGenerator:
@staticmethod
def generate_benchmark_table_header(benchmark_group_result: BenchmarkGroupResult) -> Panel:
header_text = Text(f"TYPE: {benchmark_group_result.type_.upper()} ; "
header_text = Text(
f"TYPE: {benchmark_group_result.type_.upper()} ; "
f"ITERATIONS: {benchmark_group_result.iterations} ; "
f"GC {"DISABLED" if benchmark_group_result.is_gc_disabled else "ENABLED"} ; "
f"GC {'DISABLED' if benchmark_group_result.is_gc_disabled else 'ENABLED'} ; "
f"ALL TIME IN MS",
style="bold magenta")
style="bold magenta",
)
return Panel(header_text, expand=False, border_style="magenta")
def generate_system_info_table(self) -> Table:
if self._cached_system_info_table is not None:
return self._cached_system_info_table
table = Table(show_header=True, header_style="bold cyan", border_style="blue", show_lines=True)
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")
@@ -55,10 +63,10 @@ class ReportTableGenerator:
table.add_row("CPU", self.system_info.cpu_info.name)
table.add_row("CPU Physical Cores", str(self.system_info.cpu_info.physical_cores))
table.add_row("CPU Logical Cores", str(self.system_info.cpu_info.logical_cores))
table.add_row("CPU Max Frequency", str(self.system_info.cpu_info.max_frequency) + ' GHz')
table.add_row("Total RAM", str(self.system_info.memory_info.total_ram) + ' GB')
table.add_row("Used RAM", str(self.system_info.memory_info.used_ram) + ' GB')
table.add_row("Available RAM", str(self.system_info.memory_info.available_ram) + ' GB')
table.add_row("CPU Max Frequency", str(self.system_info.cpu_info.max_frequency) + " GHz")
table.add_row("Total RAM", str(self.system_info.memory_info.total_ram) + " GB")
table.add_row("Used RAM", str(self.system_info.memory_info.used_ram) + " GB")
table.add_row("Available RAM", str(self.system_info.memory_info.available_ram) + " GB")
table.add_row("Python Version", self.system_info.python_info.version)
table.add_row("Python Implementation", self.system_info.python_info.implementation)
table.add_row("Python Compiler", self.system_info.python_info.compiler)
+33 -38
View File
@@ -1,28 +1,19 @@
__all__ = [
"SystemInfo",
"get_system_info"
]
__all__ = ["SystemInfo", "get_system_info"]
from dataclasses import dataclass
import platform
import sys
from dataclasses import dataclass
import cpuinfo
import psutil
@dataclass(frozen=True, slots=True)
class SystemInfo:
os_info: OSInfo
cpu_info: CPUInfo
memory_info: MemoryInfo
python_info: PythonInfo
@dataclass(frozen=True, slots=True)
class OSInfo:
name: str
kernel_version: str
@dataclass(frozen=True, slots=True)
class CPUInfo:
name: str
@@ -31,12 +22,14 @@ class CPUInfo:
logical_cores: int
max_frequency: float
@dataclass(frozen=True, slots=True)
class MemoryInfo:
total_ram: float # in GB
used_ram: float # in GB
available_ram: float # in GB
@dataclass(frozen=True, slots=True)
class PythonInfo:
version: str
@@ -44,18 +37,6 @@ class PythonInfo:
compiler: str
def get_system_info() -> SystemInfo:
os_info = get_os_info()
cpu_info = get_cpu_info()
memory_info = get_memory_info()
python_info = get_python_info()
return SystemInfo(
os_info=os_info,
cpu_info=cpu_info,
memory_info=memory_info,
python_info=python_info,
)
def get_os_info() -> OSInfo:
system = platform.system()
@@ -73,22 +54,17 @@ def get_os_info() -> OSInfo:
kernel_version=kernel_version,
)
elif system == "Darwin":
return OSInfo(
kernel_version=platform.release(),
name=f"macOS {platform.mac_ver()[0]}"
)
return OSInfo(kernel_version=platform.release(), name=f"macOS {platform.mac_ver()[0]}")
else:
return OSInfo(
kernel_version=platform.release(),
name=platform.system()
)
return OSInfo(kernel_version=platform.release(), name=platform.system())
def get_cpu_info() -> CPUInfo:
cpu_info = cpuinfo.get_cpu_info()
cpu_name = cpu_info["brand_raw"]
cpu_architecture = cpu_info["arch"]
cpu_physical_cores = psutil.cpu_count(logical=False)
cpu_logical_cores = psutil.cpu_count(logical=True)
cpu_physical_cores = psutil.cpu_count(logical=False) or 0
cpu_logical_cores = psutil.cpu_count(logical=True) or 0
cpu_freq = psutil.cpu_freq()
cpu_max_frequency = cpu_freq.max
@@ -98,9 +74,10 @@ def get_cpu_info() -> CPUInfo:
architecture=cpu_architecture,
physical_cores=cpu_physical_cores,
logical_cores=cpu_logical_cores,
max_frequency=cpu_max_frequency
max_frequency=cpu_max_frequency,
)
def get_memory_info() -> MemoryInfo:
mem = psutil.virtual_memory()
total_ram = round(mem.total / (1024**3), 2)
@@ -113,14 +90,32 @@ def get_memory_info() -> MemoryInfo:
available_ram=available_ram,
)
def get_python_info() -> PythonInfo:
python_version = platform.python_version()
python_implementation = platform.python_implementation()
python_compiler = platform.python_compiler()
return PythonInfo(
version=python_version,
implementation=python_implementation,
compiler=python_compiler
version=python_version, implementation=python_implementation, compiler=python_compiler
)
@dataclass(frozen=True, slots=True)
class SystemInfo:
os_info: OSInfo
cpu_info: CPUInfo
memory_info: MemoryInfo
python_info: PythonInfo
def get_system_info() -> SystemInfo:
os_info = get_os_info()
cpu_info = get_cpu_info()
memory_info = get_memory_info()
python_info = get_python_info()
return SystemInfo(
os_info=os_info,
cpu_info=cpu_info,
memory_info=memory_info,
python_info=python_info,
)
+1
View File
@@ -61,6 +61,7 @@ line-length=100
[tool.pyright]
typeCheckingMode = "strict"
reportMissingTypeStubs = false
[[tool.pyright.executionEnvironments]]
root = "tests/"