From 578453cebeea7e24f07f263dd43da8e90b48dbc9 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 17:03:42 +0000 Subject: [PATCH 1/2] [pre-commit.ci] pre-commit autoupdate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit updates: - [github.com/pre-commit/pre-commit-hooks: v5.0.0 → v6.0.0](https://github.com/pre-commit/pre-commit-hooks/compare/v5.0.0...v6.0.0) - [github.com/astral-sh/ruff-pre-commit: v0.14.14 → v0.15.21](https://github.com/astral-sh/ruff-pre-commit/compare/v0.14.14...v0.15.21) --- .pre-commit-config.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ed4e3a5893..9c4af9f02a 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -14,7 +14,7 @@ ci: - check-ruff-baselined repos: - repo: https://github.com/pre-commit/pre-commit-hooks - rev: v5.0.0 # Use the latest stable version + rev: v6.0.0 # Use the latest stable version hooks: - id: check-ast - id: debug-statements @@ -114,7 +114,7 @@ repos: pass_filenames: false args: [] - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.14.14 + rev: v0.15.21 hooks: - id: ruff args: [--fix, --exit-non-zero-on-fix] From 8d67c595f69e9f0c9726d2cce53810aa1c6fb331 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 17:04:02 +0000 Subject: [PATCH 2/2] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/aiperf/api/api_service.py | 12 ++++---- src/aiperf/common/base_component_service.py | 8 ++++-- src/aiperf/common/base_service.py | 4 ++- .../common/mixins/command_handler_mixin.py | 28 ++++++++++++++----- src/aiperf/common/mixins/hooks_mixin.py | 9 ++++-- src/aiperf/common/mixins/message_bus_mixin.py | 4 ++- .../common/mixins/progress_tracker_mixin.py | 4 ++- src/aiperf/common/mixins/pull_client_mixin.py | 4 ++- .../common/mixins/reply_client_mixin.py | 4 ++- src/aiperf/config/comm/base.py | 24 ++++++++-------- .../multiprocess_service_manager.py | 9 ++++-- src/aiperf/dataset/composer/base.py | 11 +++++--- .../dataset/composer/synthetic_rankings.py | 4 ++- src/aiperf/dataset/dataset_manager.py | 6 ++-- src/aiperf/dataset/generator/image.py | 4 ++- src/aiperf/dataset/generator/prompt.py | 16 +++++++---- src/aiperf/dataset/loader/base_hf_dataset.py | 4 ++- src/aiperf/dataset/loader/hf_asr.py | 12 ++++++-- src/aiperf/dataset/loader/sharegpt.py | 4 ++- src/aiperf/dataset/memory_map_utils.py | 18 ++++++++---- src/aiperf/exporters/metrics_csv_exporter.py | 4 ++- src/aiperf/exporters/metrics_json_exporter.py | 4 ++- .../timeslice_metrics_csv_exporter.py | 8 ++++-- .../timeslice_metrics_json_exporter.py | 8 ++++-- src/aiperf/network_latency/probe.py | 10 +++++-- .../metric_results_processor.py | 8 ++++-- src/aiperf/records/inference_result_parser.py | 6 ++-- src/aiperf/server_metrics/csv_exporter.py | 8 ++++-- src/aiperf/server_metrics/data_collector.py | 4 ++- src/aiperf/server_metrics/json_exporter.py | 8 ++++-- src/aiperf/server_metrics/manager.py | 24 ++++++++++++---- src/aiperf/server_metrics/parquet_exporter.py | 17 +++++++---- src/aiperf/timing/manager.py | 8 ++++-- src/aiperf/timing/phase_orchestrator.py | 10 +++++-- .../dashboard/realtime_metrics_dashboard.py | 5 ++-- src/aiperf/workers/worker_manager.py | 4 ++- src/aiperf/zmq/dealer_request_client.py | 4 ++- src/aiperf/zmq/router_reply_client.py | 9 ++++-- src/aiperf/zmq/sub_client.py | 8 ++++-- src/aiperf/zmq/zmq_base_client.py | 16 ++++++++--- src/aiperf/zmq/zmq_proxy_base.py | 20 +++++++++---- src/aiperf/zmq/zmq_proxy_sockets.py | 4 ++- tests/unit/common/test_aiperf_logger.py | 24 +++++++++++----- tests/unit/orchestrator/test_cell_callback.py | 4 +-- tests/unit/plot/test_dashboard_utils.py | 4 +-- 45 files changed, 292 insertions(+), 126 deletions(-) diff --git a/src/aiperf/api/api_service.py b/src/aiperf/api/api_service.py index c890743c73..4074113e9a 100644 --- a/src/aiperf/api/api_service.py +++ b/src/aiperf/api/api_service.py @@ -151,11 +151,13 @@ async def _start_api_server(self) -> None: self.info(f"AIPerf FastAPI started at {self._base_url}/") self.info( - lambda: " Routes: " - + " | ".join( - r.path - for r in self.app.routes - if hasattr(r, "methods") and r.path not in ("/openapi.json",) + lambda: ( + " Routes: " + + " | ".join( + r.path + for r in self.app.routes + if hasattr(r, "methods") and r.path not in ("/openapi.json",) + ) ) ) diff --git a/src/aiperf/common/base_component_service.py b/src/aiperf/common/base_component_service.py index 5aaa3af213..631aa07f38 100644 --- a/src/aiperf/common/base_component_service.py +++ b/src/aiperf/common/base_component_service.py @@ -67,7 +67,9 @@ async def _heartbeat_task(self) -> None: async def _register_service_on_start(self) -> None: """Register the service with the system controller on startup.""" self.debug( - lambda: f"Attempting to register service {self} ({self.service_id}) with system controller" + lambda: ( + f"Attempting to register service {self} ({self.service_id}) with system controller" + ) ) result = None command_message = RegisterServiceCommand( @@ -97,7 +99,9 @@ async def _register_service_on_start(self) -> None: ) else: self.debug( - lambda: f"Service {self.service_id} registered with system controller" + lambda: ( + f"Service {self.service_id} registered with system controller" + ) ) break diff --git a/src/aiperf/common/base_service.py b/src/aiperf/common/base_service.py index fb07311aea..5d2e29f073 100644 --- a/src/aiperf/common/base_service.py +++ b/src/aiperf/common/base_service.py @@ -181,7 +181,9 @@ async def _kill(self) -> None: ) except Exception as publish_error: self.debug( - lambda e=publish_error: f"Failed to publish BaseServiceErrorMessage during _kill (comms may already be down): {e!r}" + lambda e=publish_error: ( + f"Failed to publish BaseServiceErrorMessage during _kill (comms may already be down): {e!r}" + ) ) self.stop_requested = True self.stopped_event.set() diff --git a/src/aiperf/common/mixins/command_handler_mixin.py b/src/aiperf/common/mixins/command_handler_mixin.py index ae96900c61..bdcce57bcd 100644 --- a/src/aiperf/common/mixins/command_handler_mixin.py +++ b/src/aiperf/common/mixins/command_handler_mixin.py @@ -74,12 +74,16 @@ async def _process_command_message(self, message: CommandMessage) -> None: """ self.trace_or_debug( lambda: f"Received command message: {message}", - lambda: f"Received '{message.command}' command from '{message.service_id}' (id: {message.command_id})", + lambda: ( + f"Received '{message.command}' command from '{message.service_id}' (id: {message.command_id})" + ), ) if message.command_id in self._processed_command_ids: self.trace_or_debug( lambda: f"Received duplicate command message: {message}. Ignoring.", - lambda: f"Received duplicate command message '{message.command}' from '{message.service_id}' (id: {message.command_id}). Ignoring.", + lambda: ( + f"Received duplicate command message '{message.command}' from '{message.service_id}' (id: {message.command_id}). Ignoring." + ), ) # If we receive a duplicate command message, we just send an acknowledged response. await self._publish_command_acknowledged_response(message) @@ -91,7 +95,9 @@ async def _process_command_message(self, message: CommandMessage) -> None: # In the case of a broadcast command, you will receive a command message from yourself. # We ignore these messages. self.debug( - lambda: f"Received broadcast command message from self: {message}. Ignoring." + lambda: ( + f"Received broadcast command message from self: {message}. Ignoring." + ) ) return @@ -317,7 +323,9 @@ async def _process_command_response_message(self, message: CommandResponse) -> N """ self.trace_or_debug( lambda: f"Received command response message: {message}", - lambda: f"Received {message.status} response for command '{message.command}' from '{message.service_id}' (id: {message.command_id})", + lambda: ( + f"Received {message.status} response for command '{message.command}' from '{message.service_id}' (id: {message.command_id})" + ), ) # If the command ID is in the single response futures, we set the result of the future. @@ -328,7 +336,9 @@ async def _process_command_response_message(self, message: CommandResponse) -> N future.set_result(message) else: self.debug( - lambda: f"Already received response for command {message.command_id}, ignoring duplicate from {message.service_id}" + lambda: ( + f"Already received response for command {message.command_id}, ignoring duplicate from {message.service_id}" + ) ) return @@ -343,7 +353,9 @@ async def _process_command_response_message(self, message: CommandResponse) -> N future.set_result(message) else: self.debug( - lambda: f"Already received response for command {message.command_id} from {message.service_id}, ignoring duplicate" + lambda: ( + f"Already received response for command {message.command_id} from {message.service_id}, ignoring duplicate" + ) ) else: if message.command != CommandType.PROFILE_CANCEL: @@ -355,5 +367,7 @@ async def _process_command_response_message(self, message: CommandResponse) -> N # If we reach here, we received a command response that we were not tracking. It is # safe to ignore. self.debug( - lambda: f"Received command response for untracked command: {message}. Ignoring." + lambda: ( + f"Received command response for untracked command: {message}. Ignoring." + ) ) diff --git a/src/aiperf/common/mixins/hooks_mixin.py b/src/aiperf/common/mixins/hooks_mixin.py index 795951af5a..baed7433ff 100644 --- a/src/aiperf/common/mixins/hooks_mixin.py +++ b/src/aiperf/common/mixins/hooks_mixin.py @@ -69,7 +69,9 @@ def __init__(self, **kwargs) -> None: ) self.trace( - lambda: f"Provided hook types: {self._provided_hook_types} for {self.__class__.__name__}" + lambda: ( + f"Provided hook types: {self._provided_hook_types} for {self.__class__.__name__}" + ) ) def _check_hook_type_is_provided(self, hook_type: HookType) -> None: @@ -152,8 +154,9 @@ def for_each_hook_param( ) for param in params: self.trace( - lambda param=param, - type=param_type: f"param: {param}, param_type: {type}" + lambda param=param, type=param_type: ( + f"param: {param}, param_type: {type}" + ) ) if not isinstance(param, param_type): raise ValueError( diff --git a/src/aiperf/common/mixins/message_bus_mixin.py b/src/aiperf/common/mixins/message_bus_mixin.py index 9f7dc2b645..57aa913265 100644 --- a/src/aiperf/common/mixins/message_bus_mixin.py +++ b/src/aiperf/common/mixins/message_bus_mixin.py @@ -50,7 +50,9 @@ def _add_to_subscription_map(hook: Hook, message_type: MessageTypeT) -> None: subscribe_all for efficiency """ self.debug( - lambda: f"Adding subscription for message type: '{message_type}' for hook: {hook}" + lambda: ( + f"Adding subscription for message type: '{message_type}' for hook: {hook}" + ) ) subscription_map.setdefault(message_type, []).append(hook.func) diff --git a/src/aiperf/common/mixins/progress_tracker_mixin.py b/src/aiperf/common/mixins/progress_tracker_mixin.py index edbc5aa623..4e0e308676 100644 --- a/src/aiperf/common/mixins/progress_tracker_mixin.py +++ b/src/aiperf/common/mixins/progress_tracker_mixin.py @@ -80,7 +80,9 @@ def _update_phase_progress( pct = getattr(stats, f"{prefix}_progress_percent") _logger.debug( - lambda: f"Updating {prefix} stats for phase '{stats.phase.title()}': progress_percent: {pct}, finished: {finished}" + lambda: ( + f"Updating {prefix} stats for phase '{stats.phase.title()}': progress_percent: {pct}, finished: {finished}" + ) ) if not pct or finished == 0: diff --git a/src/aiperf/common/mixins/pull_client_mixin.py b/src/aiperf/common/mixins/pull_client_mixin.py index 0918def9dc..2a7f4324bd 100644 --- a/src/aiperf/common/mixins/pull_client_mixin.py +++ b/src/aiperf/common/mixins/pull_client_mixin.py @@ -46,7 +46,9 @@ async def _setup_pull_handler_hooks(self) -> None: def _register_pull_callback(hook: Hook, message_type: MessageTypeT) -> None: self.debug( - lambda: f"Registering pull callback for message type: {message_type} for hook: {hook}" + lambda: ( + f"Registering pull callback for message type: {message_type} for hook: {hook}" + ) ) self.pull_client.register_pull_callback( message_type=message_type, diff --git a/src/aiperf/common/mixins/reply_client_mixin.py b/src/aiperf/common/mixins/reply_client_mixin.py index 16e5dce229..b663beb259 100644 --- a/src/aiperf/common/mixins/reply_client_mixin.py +++ b/src/aiperf/common/mixins/reply_client_mixin.py @@ -40,7 +40,9 @@ async def _setup_request_handler_hooks(self) -> None: def _register_request_handler(hook: Hook, message_type: MessageTypeT) -> None: self.debug( - lambda: f"Registering request handler for message type: {message_type} for hook: {hook}" + lambda: ( + f"Registering request handler for message type: {message_type} for hook: {hook}" + ) ) self.reply_client.register_request_handler( service_id=self.id, diff --git a/src/aiperf/config/comm/base.py b/src/aiperf/config/comm/base.py index 5cd5ec02d9..5897f6fe2e 100644 --- a/src/aiperf/config/comm/base.py +++ b/src/aiperf/config/comm/base.py @@ -121,25 +121,25 @@ def get_address(self, address_type: CommAddress) -> str: # CommAddress to a resolver callable keeps BaseZMQCommunicationConfig.get_address # flat (no match ladder) and makes the mapping easy to extend. _ADDRESS_RESOLVERS: dict[CommAddress, Callable[[BaseZMQCommunicationConfig], str]] = { - CommAddress.EVENT_BUS_PROXY_FRONTEND: lambda c: c.event_bus_proxy_config.resolve_frontend( - c._remote_host + CommAddress.EVENT_BUS_PROXY_FRONTEND: lambda c: ( + c.event_bus_proxy_config.resolve_frontend(c._remote_host) ), - CommAddress.EVENT_BUS_PROXY_BACKEND: lambda c: c.event_bus_proxy_config.resolve_backend( - c._remote_host + CommAddress.EVENT_BUS_PROXY_BACKEND: lambda c: ( + c.event_bus_proxy_config.resolve_backend(c._remote_host) ), - CommAddress.DATASET_MANAGER_PROXY_FRONTEND: lambda c: c.dataset_manager_proxy_config.resolve_frontend( - c._remote_host + CommAddress.DATASET_MANAGER_PROXY_FRONTEND: lambda c: ( + c.dataset_manager_proxy_config.resolve_frontend(c._remote_host) ), - CommAddress.DATASET_MANAGER_PROXY_BACKEND: lambda c: c.dataset_manager_proxy_config.resolve_backend( - c._remote_host + CommAddress.DATASET_MANAGER_PROXY_BACKEND: lambda c: ( + c.dataset_manager_proxy_config.resolve_backend(c._remote_host) ), # Raw inference proxy is always local (within-pod IPC). Workers and record # processors are co-located in the same pod, so remote_host is ignored. - CommAddress.RAW_INFERENCE_PROXY_FRONTEND: lambda c: c.raw_inference_proxy_config.resolve_frontend( - None + CommAddress.RAW_INFERENCE_PROXY_FRONTEND: lambda c: ( + c.raw_inference_proxy_config.resolve_frontend(None) ), - CommAddress.RAW_INFERENCE_PROXY_BACKEND: lambda c: c.raw_inference_proxy_config.resolve_backend( - None + CommAddress.RAW_INFERENCE_PROXY_BACKEND: lambda c: ( + c.raw_inference_proxy_config.resolve_backend(None) ), CommAddress.CREDIT_ROUTER: lambda c: c.credit_router_address, CommAddress.CREDIT_RETURN_ROUTER: lambda c: c.credit_return_router_address, diff --git a/src/aiperf/controller/multiprocess_service_manager.py b/src/aiperf/controller/multiprocess_service_manager.py index df6f3f972b..e98a44c1e6 100644 --- a/src/aiperf/controller/multiprocess_service_manager.py +++ b/src/aiperf/controller/multiprocess_service_manager.py @@ -95,8 +95,9 @@ async def run_service( process.start() self.debug( - lambda pid=process.pid, - type=service_type: f"Service {type} started as process (pid: {pid})" + lambda pid=process.pid, type=service_type: ( + f"Service {type} started as process (pid: {pid})" + ) ) self.multi_process_info.append( @@ -269,7 +270,9 @@ async def _wait_for_process(self, info: MultiProcessRunInfo) -> None: info.process.kill() else: self.debug( - lambda: f"Service {info.service_type} process stopped (pid: {info.process.pid})" + lambda: ( + f"Service {info.service_type} process stopped (pid: {info.process.pid})" + ) ) async def wait_for_all_services_start( diff --git a/src/aiperf/dataset/composer/base.py b/src/aiperf/dataset/composer/base.py index a7b08f0640..3b6aa81545 100644 --- a/src/aiperf/dataset/composer/base.py +++ b/src/aiperf/dataset/composer/base.py @@ -330,7 +330,9 @@ def _inject_context_prompts(self, conversations: list[Conversation]) -> None: if shared_system_prompt: conversation.system_message = shared_system_prompt self.trace( - lambda conv=conversation: f"Set system_message on conversation {conv.session_id}" + lambda conv=conversation: ( + f"Set system_message on conversation {conv.session_id}" + ) ) # Set user context prompt (unique per session) @@ -340,9 +342,10 @@ def _inject_context_prompts(self, conversations: list[Conversation]) -> None: ) conversation.user_context_message = user_context self.trace( - lambda idx=session_index, - conv=conversation: f"Set user_context_message for session {idx} " - f"(conversation {conv.session_id})" + lambda idx=session_index, conv=conversation: ( + f"Set user_context_message for session {idx} " + f"(conversation {conv.session_id})" + ) ) @staticmethod diff --git a/src/aiperf/dataset/composer/synthetic_rankings.py b/src/aiperf/dataset/composer/synthetic_rankings.py index 459ce54dce..0cf5a02ec4 100644 --- a/src/aiperf/dataset/composer/synthetic_rankings.py +++ b/src/aiperf/dataset/composer/synthetic_rankings.py @@ -109,6 +109,8 @@ def _create_turn(self, num_passages: int) -> Turn: self._finalize_turn(turn) self.debug( - lambda: f"[rankings] query_len={len(query_text)} chars, passages={num_passages}" + lambda: ( + f"[rankings] query_len={len(query_text)} chars, passages={num_passages}" + ) ) return turn diff --git a/src/aiperf/dataset/dataset_manager.py b/src/aiperf/dataset/dataset_manager.py index 38688ae792..eb83a6a6ed 100644 --- a/src/aiperf/dataset/dataset_manager.py +++ b/src/aiperf/dataset/dataset_manager.py @@ -293,8 +293,10 @@ def _generate_input_payloads( ) endpoint: EndpointProtocol = EndpointClass(model_endpoint=model_endpoint) self.debug( - lambda: f"Created endpoint protocol for {model_endpoint.endpoint.type}, " - f"class: {endpoint.__class__.__name__}", + lambda: ( + f"Created endpoint protocol for {model_endpoint.endpoint.type}, " + f"class: {endpoint.__class__.__name__}" + ), ) session_payloads_map: dict[str, list] = {} for conversation in self.dataset.values(): diff --git a/src/aiperf/dataset/generator/image.py b/src/aiperf/dataset/generator/image.py index 40b2115a19..8dcdabeb09 100644 --- a/src/aiperf/dataset/generator/image.py +++ b/src/aiperf/dataset/generator/image.py @@ -116,7 +116,9 @@ def generate(self, *args, **kwargs) -> str: image = self._create_source_image(width, height) self.debug( - lambda: f"Generated image from {self.config.source} with width={width}, height={height}" + lambda: ( + f"Generated image from {self.config.source} with width={width}, height={height}" + ) ) base64_image = utils.encode_image(image, image_format) return f"data:image/{image_format.name.lower()};base64,{base64_image}" diff --git a/src/aiperf/dataset/generator/prompt.py b/src/aiperf/dataset/generator/prompt.py index f375817bf5..dc2223a379 100644 --- a/src/aiperf/dataset/generator/prompt.py +++ b/src/aiperf/dataset/generator/prompt.py @@ -148,8 +148,10 @@ def tokenize_chunk(chunk): ] self._corpus_size = len(self._tokenized_corpus) self.debug( - lambda: f"Initialized corpus with {self._corpus_size} tokens " - f"from {len(chunks)} chunks using {num_threads} thread(s)" + lambda: ( + f"Initialized corpus with {self._corpus_size} tokens " + f"from {len(chunks)} chunks using {num_threads} thread(s)" + ) ) def _create_prefix_prompt_pool(self) -> None: @@ -164,7 +166,9 @@ def _create_prefix_prompt_pool(self) -> None: pool_size = self.prefix_prompts.pool_size or 0 self._prefix_prompts = [self.generate_prompt(length) for _ in range(pool_size)] self.debug( - lambda: f"Initialized prefix prompts pool with {len(self._prefix_prompts)} prompts" + lambda: ( + f"Initialized prefix prompts pool with {len(self._prefix_prompts)} prompts" + ) ) def generate( @@ -433,8 +437,10 @@ def generate_user_context_prompt(self, session_index: int) -> str: new_prompt = self.generate_prompt(length) self._user_context_prompts.append(new_prompt) self.debug( - lambda: f"Generated user context prompt #{len(self._user_context_prompts) - 1} " - f"for session {len(self._user_context_prompts) - 1}" + lambda: ( + f"Generated user context prompt #{len(self._user_context_prompts) - 1} " + f"for session {len(self._user_context_prompts) - 1}" + ) ) return self._user_context_prompts[session_index] diff --git a/src/aiperf/dataset/loader/base_hf_dataset.py b/src/aiperf/dataset/loader/base_hf_dataset.py index 93afcfb916..62f1e7b422 100644 --- a/src/aiperf/dataset/loader/base_hf_dataset.py +++ b/src/aiperf/dataset/loader/base_hf_dataset.py @@ -178,7 +178,9 @@ def _extract_audio(self, row: dict[str, Any], audio_column: str) -> list[Audio]: return [Audio(name="", contents=[f"wav,{b64}"])] except (ImportError, OSError, ValueError, RuntimeError) as e: self.debug( - lambda exc=e: f"Failed to encode WAV from column '{audio_column}': {exc.__class__.__name__}: {exc}" + lambda exc=e: ( + f"Failed to encode WAV from column '{audio_column}': {exc.__class__.__name__}: {exc}" + ) ) return [] diff --git a/src/aiperf/dataset/loader/hf_asr.py b/src/aiperf/dataset/loader/hf_asr.py index 83f02b54c5..61de90e365 100644 --- a/src/aiperf/dataset/loader/hf_asr.py +++ b/src/aiperf/dataset/loader/hf_asr.py @@ -74,7 +74,9 @@ def _audio_from_bytes(self, audio_value: _HFAudioBytesRow) -> list[Audio]: return [Audio(name="", contents=[f"wav,{b64}"])] except (ImportError, OSError, ValueError, RuntimeError) as e: self.debug( - lambda exc=e: f"Failed to decode audio bytes: {exc.__class__.__name__}: {exc}" + lambda exc=e: ( + f"Failed to decode audio bytes: {exc.__class__.__name__}: {exc}" + ) ) return [] @@ -90,7 +92,9 @@ def _duration_seconds(self, audio_value: _HFAudioBytesRow) -> float | None: return info.duration except (ImportError, OSError, ValueError, RuntimeError) as e: self.debug( - lambda exc=e: f"Failed to estimate audio duration: {exc.__class__.__name__}: {exc}" + lambda exc=e: ( + f"Failed to estimate audio duration: {exc.__class__.__name__}: {exc}" + ) ) return None @@ -150,6 +154,8 @@ async def convert_to_conversations( f"Check that '{self.audio_column}' contains valid audio data." ) self.debug( - lambda: f"Converted {len(conversations)} rows (skipped {skipped} empty/long)" + lambda: ( + f"Converted {len(conversations)} rows (skipped {skipped} empty/long)" + ) ) return conversations diff --git a/src/aiperf/dataset/loader/sharegpt.py b/src/aiperf/dataset/loader/sharegpt.py index 255f56a11f..a5695c0343 100644 --- a/src/aiperf/dataset/loader/sharegpt.py +++ b/src/aiperf/dataset/loader/sharegpt.py @@ -148,7 +148,9 @@ async def convert_to_conversations( ) self.debug( - lambda: f"Filtered to {len(filtered_dataset)} dataset entries out of {len(dataset)} (skipped {skipped_entries})" + lambda: ( + f"Filtered to {len(filtered_dataset)} dataset entries out of {len(dataset)} (skipped {skipped_entries})" + ) ) return filtered_dataset diff --git a/src/aiperf/dataset/memory_map_utils.py b/src/aiperf/dataset/memory_map_utils.py index 806ab3f7a5..10aee2748b 100644 --- a/src/aiperf/dataset/memory_map_utils.py +++ b/src/aiperf/dataset/memory_map_utils.py @@ -326,12 +326,16 @@ async def _setup(self) -> None: """Open memory-mapped files (read-only).""" self._loop = asyncio.get_running_loop() self.debug( - lambda: f"Opening memory-mapped files: data={self._data_path}, index={self._index_path}" + lambda: ( + f"Opening memory-mapped files: data={self._data_path}, index={self._index_path}" + ) ) self._client = MemoryMapDatasetClient(self._data_path, self._index_path) self.debug( - lambda: f"Memory-mapped client store initialized with " - f"{len(self._client.index.conversation_ids)} conversations" + lambda: ( + f"Memory-mapped client store initialized with " + f"{len(self._client.index.conversation_ids)} conversations" + ) ) async def get_conversation(self, conversation_id: str) -> Conversation: @@ -465,7 +469,9 @@ def __init__(self, data_file_path: Path | str, index_file_path: Path | str) -> N ) _logger.debug( - lambda: f"MemoryMapDatasetClient initialized successfully: data_file={self.data_file_path}, index_file={self.index_file_path}, conversations={len(self.index.conversation_ids)}, size={self.index.total_size} bytes" + lambda: ( + f"MemoryMapDatasetClient initialized successfully: data_file={self.data_file_path}, index_file={self.index_file_path}, conversations={len(self.index.conversation_ids)}, size={self.index.total_size} bytes" + ) ) def __enter__(self) -> "MemoryMapDatasetClient": @@ -546,7 +552,9 @@ def get_conversation(self, conversation_id: str) -> Conversation: conv_bytes = self.data_mmap.read(offset_info.size) _logger.debug( - lambda: f"Loading conversation '{conversation_id}': offset={offset_info.offset}, size={offset_info.size} bytes" + lambda: ( + f"Loading conversation '{conversation_id}': offset={offset_info.offset}, size={offset_info.size} bytes" + ) ) return self._deserialize_conversation(conv_bytes) diff --git a/src/aiperf/exporters/metrics_csv_exporter.py b/src/aiperf/exporters/metrics_csv_exporter.py index e1e32deabe..130d045157 100644 --- a/src/aiperf/exporters/metrics_csv_exporter.py +++ b/src/aiperf/exporters/metrics_csv_exporter.py @@ -28,7 +28,9 @@ def __init__(self, exporter_config: ExporterConfig, **kwargs) -> None: self._percentile_keys = _percentile_keys_from(STAT_KEYS) self.trace_or_debug( lambda: f"Initializing MetricsCsvExporter with config: {exporter_config}", - lambda: f"Initializing MetricsCsvExporter with file path: {self._file_path}", + lambda: ( + f"Initializing MetricsCsvExporter with file path: {self._file_path}" + ), ) def get_export_info(self) -> FileExportInfo: diff --git a/src/aiperf/exporters/metrics_json_exporter.py b/src/aiperf/exporters/metrics_json_exporter.py index a88ab2b304..e42aa9935b 100644 --- a/src/aiperf/exporters/metrics_json_exporter.py +++ b/src/aiperf/exporters/metrics_json_exporter.py @@ -34,7 +34,9 @@ def __init__(self, exporter_config: ExporterConfig, **kwargs) -> None: self._file_path = exporter_config.cfg.artifacts.profile_export_json_file self.trace_or_debug( lambda: f"Initializing MetricsJsonExporter with config: {exporter_config}", - lambda: f"Initializing MetricsJsonExporter with file path: {self._file_path}", + lambda: ( + f"Initializing MetricsJsonExporter with file path: {self._file_path}" + ), ) def get_export_info(self) -> FileExportInfo: diff --git a/src/aiperf/exporters/timeslice_metrics_csv_exporter.py b/src/aiperf/exporters/timeslice_metrics_csv_exporter.py index 161aca6107..9da5f52d49 100644 --- a/src/aiperf/exporters/timeslice_metrics_csv_exporter.py +++ b/src/aiperf/exporters/timeslice_metrics_csv_exporter.py @@ -39,8 +39,12 @@ def __init__(self, exporter_config: ExporterConfig, **kwargs) -> None: exporter_config.cfg.artifacts.profile_export_timeslices_csv_file ) self.trace_or_debug( - lambda: f"Initializing TimesliceMetricsCsvExporter with config: {exporter_config}", - lambda: f"Initializing TimesliceMetricsCsvExporter with file path: {self._file_path}", + lambda: ( + f"Initializing TimesliceMetricsCsvExporter with config: {exporter_config}" + ), + lambda: ( + f"Initializing TimesliceMetricsCsvExporter with file path: {self._file_path}" + ), ) def get_export_info(self) -> FileExportInfo: diff --git a/src/aiperf/exporters/timeslice_metrics_json_exporter.py b/src/aiperf/exporters/timeslice_metrics_json_exporter.py index e3fe3040d8..b352986dde 100644 --- a/src/aiperf/exporters/timeslice_metrics_json_exporter.py +++ b/src/aiperf/exporters/timeslice_metrics_json_exporter.py @@ -44,8 +44,12 @@ def __init__(self, exporter_config: ExporterConfig, **kwargs) -> None: exporter_config.cfg.artifacts.profile_export_timeslices_json_file ) self.trace_or_debug( - lambda: f"Initializing TimesliceMetricsJsonExporter with config: {exporter_config}", - lambda: f"Initializing TimesliceMetricsJsonExporter with file path: {self._file_path}", + lambda: ( + f"Initializing TimesliceMetricsJsonExporter with config: {exporter_config}" + ), + lambda: ( + f"Initializing TimesliceMetricsJsonExporter with file path: {self._file_path}" + ), ) def get_export_info(self) -> FileExportInfo: diff --git a/src/aiperf/network_latency/probe.py b/src/aiperf/network_latency/probe.py index 40d19076cd..3e0313dd9a 100644 --- a/src/aiperf/network_latency/probe.py +++ b/src/aiperf/network_latency/probe.py @@ -107,7 +107,9 @@ async def resolve(self) -> None: self._resolved_family = family self._resolved_host = sockaddr[0] self.debug( - lambda: f"Resolved {self._target_host}:{self._target_port} -> {self._resolved_host}" + lambda: ( + f"Resolved {self._target_host}:{self._target_port} -> {self._resolved_host}" + ) ) except OSError as e: self.warning( @@ -150,8 +152,10 @@ async def probe_once(self) -> None: await writer.wait_closed() except OSError as close_error: self.debug( - lambda err=close_error: f"Error closing probe connection to " - f"{self._target_host}:{self._target_port}: {err!r}" + lambda err=close_error: ( + f"Error closing probe connection to " + f"{self._target_host}:{self._target_port}: {err!r}" + ) ) except OSError as e: error = ErrorDetails.from_exception(e) diff --git a/src/aiperf/post_processors/metric_results_processor.py b/src/aiperf/post_processors/metric_results_processor.py index 941084c832..b7767f87c3 100644 --- a/src/aiperf/post_processors/metric_results_processor.py +++ b/src/aiperf/post_processors/metric_results_processor.py @@ -191,9 +191,11 @@ def _inject_network_adjusted_metrics(self) -> None: if clamped_tags: self.warning( - lambda tags=clamped_tags: f"Network RTT ({rtt_ns / 1e6:.3f} ms) exceeded " - f"some measured latencies; clamped to 0 for: {', '.join(tags)}. " - "The subtracted RTT may be larger than the true network leg." + lambda tags=clamped_tags: ( + f"Network RTT ({rtt_ns / 1e6:.3f} ms) exceeded " + f"some measured latencies; clamped to 0 for: {', '.join(tags)}. " + "The subtracted RTT may be larger than the true network leg." + ) ) async def update_derived_metrics(self) -> None: diff --git a/src/aiperf/records/inference_result_parser.py b/src/aiperf/records/inference_result_parser.py index f9b340de00..7bbda0e22a 100644 --- a/src/aiperf/records/inference_result_parser.py +++ b/src/aiperf/records/inference_result_parser.py @@ -420,8 +420,10 @@ async def _compute_chat_template_token_count( ) except Exception as exc: # noqa: BLE001 - best-effort; fall back to bare encode self.debug( - lambda exc=exc: f"Chat-template tokenization unavailable, " - f"falling back to bare-text encode: {exc!r}" + lambda exc=exc: ( + f"Chat-template tokenization unavailable, " + f"falling back to bare-text encode: {exc!r}" + ) ) return None if not isinstance(tokens, list): diff --git a/src/aiperf/server_metrics/csv_exporter.py b/src/aiperf/server_metrics/csv_exporter.py index 932e6f75c1..57925dbedd 100644 --- a/src/aiperf/server_metrics/csv_exporter.py +++ b/src/aiperf/server_metrics/csv_exporter.py @@ -111,8 +111,12 @@ def __init__(self, exporter_config: ExporterConfig, **kwargs) -> None: super().__init__(exporter_config, **kwargs) self._file_path = exporter_config.cfg.artifacts.server_metrics_export_csv_file self.trace_or_debug( - lambda: f"Initializing ServerMetricsCsvExporter with config: {exporter_config}", - lambda: f"Initializing ServerMetricsCsvExporter with file path: {self._file_path}", + lambda: ( + f"Initializing ServerMetricsCsvExporter with config: {exporter_config}" + ), + lambda: ( + f"Initializing ServerMetricsCsvExporter with file path: {self._file_path}" + ), ) def get_export_info(self) -> FileExportInfo: diff --git a/src/aiperf/server_metrics/data_collector.py b/src/aiperf/server_metrics/data_collector.py index b484f6b0ad..36a91fb2c9 100644 --- a/src/aiperf/server_metrics/data_collector.py +++ b/src/aiperf/server_metrics/data_collector.py @@ -306,7 +306,9 @@ def _parse_metrics_to_records( # cumulatively over server lifetime, not per-benchmark period if family.name not in self._seen_summary_metrics: self.info( - lambda name=family.name: f"Skipping unsupported summary metric: {name}" + lambda name=family.name: ( + f"Skipping unsupported summary metric: {name}" + ) ) self._seen_summary_metrics.add(family.name) continue diff --git a/src/aiperf/server_metrics/json_exporter.py b/src/aiperf/server_metrics/json_exporter.py index 00a290bca6..f046ebd760 100644 --- a/src/aiperf/server_metrics/json_exporter.py +++ b/src/aiperf/server_metrics/json_exporter.py @@ -65,8 +65,12 @@ def __init__(self, exporter_config: ExporterConfig, **kwargs) -> None: super().__init__(exporter_config, **kwargs) self._file_path = exporter_config.cfg.artifacts.server_metrics_export_json_file self.trace_or_debug( - lambda: f"Initializing ServerMetricsJsonExporter with config: {exporter_config}", - lambda: f"Initializing ServerMetricsJsonExporter with file path: {self._file_path}", + lambda: ( + f"Initializing ServerMetricsJsonExporter with config: {exporter_config}" + ), + lambda: ( + f"Initializing ServerMetricsJsonExporter with file path: {self._file_path}" + ), ) def get_export_info(self) -> FileExportInfo: diff --git a/src/aiperf/server_metrics/manager.py b/src/aiperf/server_metrics/manager.py index a6c6be655e..5c8bd50ec9 100644 --- a/src/aiperf/server_metrics/manager.py +++ b/src/aiperf/server_metrics/manager.py @@ -123,7 +123,9 @@ async def _profile_configure_command( for endpoint_url in self._server_metrics_endpoints: self.debug( - lambda url=endpoint_url: f"Server Metrics: Testing reachability of {url}" + lambda url=endpoint_url: ( + f"Server Metrics: Testing reachability of {url}" + ) ) collector = ServerMetricsDataCollector( endpoint_url=endpoint_url, @@ -138,11 +140,15 @@ async def _profile_configure_command( if is_reachable: self._collectors[endpoint_url] = collector self.debug( - lambda url=endpoint_url: f"Server Metrics: Prometheus endpoint {url} is reachable" + lambda url=endpoint_url: ( + f"Server Metrics: Prometheus endpoint {url} is reachable" + ) ) else: self.debug( - lambda url=endpoint_url: f"Server Metrics: Prometheus endpoint {url} is not reachable" + lambda url=endpoint_url: ( + f"Server Metrics: Prometheus endpoint {url} is not reachable" + ) ) except Exception as e: self.error(f"Server Metrics: Exception testing {endpoint_url}: {e}") @@ -168,7 +174,9 @@ async def _profile_configure_command( await collector.initialize() await collector.collect_and_process_metrics() self.debug( - lambda url=endpoint_url: f"Server Metrics: Captured baseline from {url}" + lambda url=endpoint_url: ( + f"Server Metrics: Captured baseline from {url}" + ) ) except Exception as e: self.warning( @@ -262,7 +270,9 @@ async def _on_credit_phase_complete( try: await collector.collect_and_process_metrics() self.debug( - lambda url=endpoint_url: f"Server Metrics: Captured warmup final state from {url}" + lambda url=endpoint_url: ( + f"Server Metrics: Captured warmup final state from {url}" + ) ) except Exception as e: # noqa: BLE001 - one endpoint's scrape failure must not skip the rest self.warning( @@ -316,7 +326,9 @@ async def _handle_profile_complete_command( try: await collector.collect_and_process_metrics() self.debug( - lambda url=endpoint_url: f"Server Metrics: Captured final state from {url}" + lambda url=endpoint_url: ( + f"Server Metrics: Captured final state from {url}" + ) ) except Exception as e: self.warning( diff --git a/src/aiperf/server_metrics/parquet_exporter.py b/src/aiperf/server_metrics/parquet_exporter.py index 4431543a37..cc8a773dc0 100644 --- a/src/aiperf/server_metrics/parquet_exporter.py +++ b/src/aiperf/server_metrics/parquet_exporter.py @@ -112,8 +112,12 @@ def __init__( self._accumulator = server_metrics_accumulator self._time_filter = time_filter self.trace_or_debug( - lambda: f"Initializing ServerMetricsParquetExporter with run cfg: {self.run.cfg}", - lambda: f"Initializing ServerMetricsParquetExporter with file path: {self._file_path}", + lambda: ( + f"Initializing ServerMetricsParquetExporter with run cfg: {self.run.cfg}" + ), + lambda: ( + f"Initializing ServerMetricsParquetExporter with file path: {self._file_path}" + ), ) def get_export_info(self) -> FileExportInfo: @@ -187,7 +191,9 @@ async def export(self) -> FileExportInfo: writer.write_table(table) total_rows += len(batch_rows) self.trace( - lambda: f"Wrote batch of {len(batch_rows):,} rows (total: {total_rows:,})" + lambda: ( + f"Wrote batch of {len(batch_rows):,} rows (total: {total_rows:,})" + ) ) batch_rows = [] @@ -206,8 +212,9 @@ async def export(self) -> FileExportInfo: batch_count = len(batch_rows) current_total = total_rows self.trace( - lambda batch_count=batch_count, - current_total=current_total: f"Wrote batch of {batch_count:,} rows (total: {current_total:,})" + lambda batch_count=batch_count, current_total=current_total: ( + f"Wrote batch of {batch_count:,} rows (total: {current_total:,})" + ) ) batch_rows = [] diff --git a/src/aiperf/timing/manager.py b/src/aiperf/timing/manager.py index 8a84d5de61..e25c4de453 100644 --- a/src/aiperf/timing/manager.py +++ b/src/aiperf/timing/manager.py @@ -88,9 +88,11 @@ async def _on_dataset_configured_notification( ) -> None: """Store dataset metadata and signal configuration ready.""" self.debug( - lambda: f"Received dataset configured notification: " - f"{len(message.metadata.conversations)} conversations, " - f"{message.metadata.sampling_strategy.value} sampling strategy" + lambda: ( + f"Received dataset configured notification: " + f"{len(message.metadata.conversations)} conversations, " + f"{message.metadata.sampling_strategy.value} sampling strategy" + ) ) self._dataset_metadata = message.metadata diff --git a/src/aiperf/timing/phase_orchestrator.py b/src/aiperf/timing/phase_orchestrator.py index d829747c52..955892091e 100644 --- a/src/aiperf/timing/phase_orchestrator.py +++ b/src/aiperf/timing/phase_orchestrator.py @@ -163,8 +163,10 @@ def conversation_source(self) -> ConversationSource: async def _init_orchestrator(self) -> None: """Log configured phases (actual initialization happens per-phase in _execute_phases).""" self.info( - lambda: f"Initialized {len(self._ordered_phase_configs)} phase(s): " - f"{[p.phase.replace('_', ' ').title() for p in self._ordered_phase_configs]}" + lambda: ( + f"Initialized {len(self._ordered_phase_configs)} phase(s): " + f"{[p.phase.replace('_', ' ').title() for p in self._ordered_phase_configs]}" + ) ) @on_start @@ -273,7 +275,9 @@ async def _stop_orchestrator(self) -> None: """ if self._active_runners: self.debug( - lambda: f"Stopping orchestrator with {len(self._active_runners)} active runner(s)" + lambda: ( + f"Stopping orchestrator with {len(self._active_runners)} active runner(s)" + ) ) self._cancel_active_runners() diff --git a/src/aiperf/ui/dashboard/realtime_metrics_dashboard.py b/src/aiperf/ui/dashboard/realtime_metrics_dashboard.py index 2119536107..7d3104dff4 100644 --- a/src/aiperf/ui/dashboard/realtime_metrics_dashboard.py +++ b/src/aiperf/ui/dashboard/realtime_metrics_dashboard.py @@ -98,8 +98,9 @@ def update(self, metrics: list[MetricResult]) -> None: metric for metric in sorted( metrics, - key=lambda m: MetricRegistry.get_class(m.tag).display_order - or sys.maxsize, + key=lambda m: ( + MetricRegistry.get_class(m.tag).display_order or sys.maxsize + ), ) if not self._should_skip(metric) ] diff --git a/src/aiperf/workers/worker_manager.py b/src/aiperf/workers/worker_manager.py index 74e41bca7e..166ac7fbf0 100644 --- a/src/aiperf/workers/worker_manager.py +++ b/src/aiperf/workers/worker_manager.py @@ -73,7 +73,9 @@ def __init__( ), ) self.debug( - lambda: f"Auto-setting max workers to {self.max_workers} due to no max workers specified." + lambda: ( + f"Auto-setting max workers to {self.max_workers} due to no max workers specified." + ) ) # Cap the worker count to the max concurrency, but only if the user is in concurrency mode. diff --git a/src/aiperf/zmq/dealer_request_client.py b/src/aiperf/zmq/dealer_request_client.py index a7c6b7023e..b1298765df 100644 --- a/src/aiperf/zmq/dealer_request_client.py +++ b/src/aiperf/zmq/dealer_request_client.py @@ -156,7 +156,9 @@ async def callback(response_message: Message) -> None: future.set_result(response_message) else: self.debug( - lambda: f"Received response for request {message.request_id} after it was already completed. Ignoring." + lambda: ( + f"Received response for request {message.request_id} after it was already completed. Ignoring." + ) ) await self.request_async(message, callback) diff --git a/src/aiperf/zmq/router_reply_client.py b/src/aiperf/zmq/router_reply_client.py index 790dbbd21e..1e5da1ea9a 100644 --- a/src/aiperf/zmq/router_reply_client.py +++ b/src/aiperf/zmq/router_reply_client.py @@ -98,8 +98,9 @@ def register_request_handler( ) self.debug( - lambda service_id=service_id, - type=message_type: f"Registering request handler for {service_id} with message type {type}" + lambda service_id=service_id, type=message_type: ( + f"Registering request handler for {service_id} with message type {type}" + ) ) self._request_handlers[message_type] = (service_id, handler) @@ -145,7 +146,9 @@ async def _wait_for_response( if response is None: self.warning( - lambda req_id=request_id: f"Got None as response for request {req_id}" + lambda req_id=request_id: ( + f"Got None as response for request {req_id}" + ) ) response = ErrorMessage( request_id=request_id, diff --git a/src/aiperf/zmq/sub_client.py b/src/aiperf/zmq/sub_client.py index 0eef5fed5e..388e99d5fc 100644 --- a/src/aiperf/zmq/sub_client.py +++ b/src/aiperf/zmq/sub_client.py @@ -135,7 +135,9 @@ async def _subscribe_internal( ) else: self.debug( - lambda: f"Adding callback to existing subscription for topic: {topic}" + lambda: ( + f"Adding callback to existing subscription for topic: {topic}" + ) ) self._subscribers.setdefault(topic, []).append(callback) @@ -182,7 +184,9 @@ async def _handle_message(self, topic_bytes: bytes, message_bytes: bytes) -> Non message = Message.from_json(message_bytes) self.trace( - lambda: f"Calling callbacks for message: {message}, {self._subscribers.get(topic)}" + lambda: ( + f"Calling callbacks for message: {message}, {self._subscribers.get(topic)}" + ) ) # Call callbacks with the parsed message object diff --git a/src/aiperf/zmq/zmq_base_client.py b/src/aiperf/zmq/zmq_base_client.py index 5b52c409c7..4eb6f7bb8f 100644 --- a/src/aiperf/zmq/zmq_base_client.py +++ b/src/aiperf/zmq/zmq_base_client.py @@ -96,7 +96,9 @@ async def _initialize_socket(self) -> None: self.scheduler = LoopScheduler() self.socket = self.context.socket(self.socket_type) self.debug( - lambda: f"ZMQ {self.socket_type_name} socket initialized, try {'BIND' if self.bind else 'CONNECT'} to {self.address} ({self.client_id})" + lambda: ( + f"ZMQ {self.socket_type_name} socket initialized, try {'BIND' if self.bind else 'CONNECT'} to {self.address} ({self.client_id})" + ) ) if zmq.IDENTITY in self.socket_ops: @@ -104,7 +106,9 @@ async def _initialize_socket(self) -> None: # otherwise it will not be properly set when the socket is bound/connected self.socket.setsockopt(zmq.IDENTITY, self.socket_ops[zmq.IDENTITY]) self.debug( - lambda: f"Set IDENTITY socket option: {self.socket_ops[zmq.IDENTITY]}" + lambda: ( + f"Set IDENTITY socket option: {self.socket_ops[zmq.IDENTITY]}" + ) ) del self.socket_ops[zmq.IDENTITY] @@ -140,7 +144,9 @@ async def _initialize_socket(self) -> None: self.socket.setsockopt(key, val) self.debug( - lambda: f"ZMQ {self.socket_type_name} socket {'BOUND' if self.bind else 'CONNECTED'} to {self.address} ({self.client_id})" + lambda: ( + f"ZMQ {self.socket_type_name} socket {'BOUND' if self.bind else 'CONNECTED'} to {self.address} ({self.client_id})" + ) ) except Exception as e: @@ -171,7 +177,9 @@ async def _shutdown_socket(self) -> None: self.socket.close() except zmq.ContextTerminated: self.debug( - lambda: f"ZMQ context already terminated, skipping socket close ({self.client_id})" + lambda: ( + f"ZMQ context already terminated, skipping socket close ({self.client_id})" + ) ) return except Exception as e: diff --git a/src/aiperf/zmq/zmq_proxy_base.py b/src/aiperf/zmq/zmq_proxy_base.py index ec2fbdb804..525a5845f9 100644 --- a/src/aiperf/zmq/zmq_proxy_base.py +++ b/src/aiperf/zmq/zmq_proxy_base.py @@ -103,7 +103,9 @@ def __init__( self.config = zmq_proxy_config self.debug( - lambda: f"Proxy Initializing - Frontend: {self.config.frontend_address}, Backend: {self.config.backend_address}" + lambda: ( + f"Proxy Initializing - Frontend: {self.config.frontend_address}, Backend: {self.config.backend_address}" + ) ) self.backend_socket = backend_socket_class( @@ -159,10 +161,14 @@ async def _initialize(self) -> None: """Initialize and start the BaseZMQProxy.""" self.debug("Proxy Initializing Sockets...") self.debug( - lambda: f"Frontend {self.frontend_socket.socket_type.name} socket binding to: {self.config.frontend_address} (for {self.backend_socket.socket_type.name} clients)" + lambda: ( + f"Frontend {self.frontend_socket.socket_type.name} socket binding to: {self.config.frontend_address} (for {self.backend_socket.socket_type.name} clients)" + ) ) self.debug( - lambda: f"Backend {self.backend_socket.socket_type.name} socket binding to: {self.config.backend_address} (for {self.frontend_socket.socket_type.name} services)" + lambda: ( + f"Backend {self.backend_socket.socket_type.name} socket binding to: {self.config.backend_address} (for {self.frontend_socket.socket_type.name} services)" + ) ) try: @@ -234,13 +240,17 @@ async def _monitor_messages(self) -> None: return self.debug( - lambda: f"Proxy Monitor Starting - Capture Address: {self.config.capture_address}" + lambda: ( + f"Proxy Monitor Starting - Capture Address: {self.config.capture_address}" + ) ) capture_socket = self.context.socket(SocketType.SUB) capture_socket.connect(self.config.capture_address) self.debug( - lambda: f"Proxy Monitor Connected to Capture Address: {self.config.capture_address}" + lambda: ( + f"Proxy Monitor Connected to Capture Address: {self.config.capture_address}" + ) ) capture_socket.setsockopt(zmq.SUBSCRIBE, b"") # Subscribe to all messages self.debug("Proxy Monitor Subscribed to all messages") diff --git a/src/aiperf/zmq/zmq_proxy_sockets.py b/src/aiperf/zmq/zmq_proxy_sockets.py index 4e36ee0e35..bc91af2e96 100644 --- a/src/aiperf/zmq/zmq_proxy_sockets.py +++ b/src/aiperf/zmq/zmq_proxy_sockets.py @@ -65,7 +65,9 @@ async def _initialize_socket(self) -> None: # explicitly (the ZMQ default) to make intent unambiguous. self.socket.setsockopt(zmq.XPUB_VERBOSE, 0) self.debug( - lambda: "XPUB socket configured with XPUB_VERBOSE=0 (de-duplicated subscription forwarding) to avoid event-bus-proxy CPU/memory blow-up" + lambda: ( + "XPUB socket configured with XPUB_VERBOSE=0 (de-duplicated subscription forwarding) to avoid event-bus-proxy CPU/memory blow-up" + ) ) # Dynamically set the class name and qualname based on the socket and end type diff --git a/tests/unit/common/test_aiperf_logger.py b/tests/unit/common/test_aiperf_logger.py index 6151f7a10f..e915729f7e 100644 --- a/tests/unit/common/test_aiperf_logger.py +++ b/tests/unit/common/test_aiperf_logger.py @@ -302,7 +302,9 @@ def aiperf_plain_string(): def aiperf_plain_string_lazy(): aiperf_logger.debug( - lambda: "Hello, world! This is a test of an example message that will NOT be printed." + lambda: ( + "Hello, world! This is a test of an example message that will NOT be printed." + ) ) def standard_plain_string(): @@ -337,7 +339,9 @@ def aiperf_plain_string(): def aiperf_plain_string_lazy(): aiperf_logger.info( - lambda: "Hello, world! This is a test of an example message that will be printed." + lambda: ( + "Hello, world! This is a test of an example message that will be printed." + ) ) def standard_plain_string(): @@ -390,9 +394,11 @@ def test_lazy_evaluation_and_formatting_debug(self, aiperf_logger, standard_logg def aiperf_formatting_and_lazy_evaluation(): aiperf_logger.debug( - lambda: "Hello, world! This will NOT be printed %s " - * 100 - % tuple([*["test"] * 100]) + lambda: ( + "Hello, world! This will NOT be printed %s " + * 100 + % tuple([*["test"] * 100]) + ) ) def standard_formatting_no_print(): @@ -417,7 +423,9 @@ def test_lazy_evaluation_and_formatting_and_multiple_args_debug( def aiperf_multiple_args(): aiperf_logger.debug( - lambda: f"Hello Mr {time.time_ns() ** 2} {time.time_ns() ** 2} This will NOT be printed" + lambda: ( + f"Hello Mr {time.time_ns() ** 2} {time.time_ns() ** 2} This will NOT be printed" + ) ) def standard_multiple_args(): @@ -496,7 +504,9 @@ def test_large_messages_debug_math( def aiperf_f_string_math(): aiperf_logger.debug( - lambda: f"Request time: {(large_message.end_perf_ns - large_message.start_perf_ns) / NANOS_PER_SECOND:.2f}" + lambda: ( + f"Request time: {(large_message.end_perf_ns - large_message.start_perf_ns) / NANOS_PER_SECOND:.2f}" + ) ) def standard_f_string_math(): diff --git a/tests/unit/orchestrator/test_cell_callback.py b/tests/unit/orchestrator/test_cell_callback.py index 953ead5085..74ca5276f9 100644 --- a/tests/unit/orchestrator/test_cell_callback.py +++ b/tests/unit/orchestrator/test_cell_callback.py @@ -102,8 +102,8 @@ def _make_executor_returning(results: list[RunResult]) -> MagicMock: """Mocked RunExecutor whose execute() pops successive results.""" iterator = iter(results) executor = MagicMock() - executor.derive_id.side_effect = ( - lambda plan, var_idx, trial: f"id-{var_idx}-{trial}" + executor.derive_id.side_effect = lambda plan, var_idx, trial: ( + f"id-{var_idx}-{trial}" ) executor.execute = AsyncMock(side_effect=lambda _run: next(iterator)) return executor diff --git a/tests/unit/plot/test_dashboard_utils.py b/tests/unit/plot/test_dashboard_utils.py index c7b329f867..88dcf9a1fc 100644 --- a/tests/unit/plot/test_dashboard_utils.py +++ b/tests/unit/plot/test_dashboard_utils.py @@ -404,8 +404,8 @@ def test_excludes_runs_with_missing_metrics(self): run.metadata.experiment_group = "group_a" if i == 1: # Make middle run missing y metric - run.get_metric = ( - lambda name: {"p50": 100} if name == "throughput" else None + run.get_metric = lambda name: ( + {"p50": 100} if name == "throughput" else None ) else: run.get_metric = lambda name: {"p50": 100}