Address utility and presentation type errors

Signed-off-by: Jared O'Connell <joconnel@redhat.com>

Author: jaredoconnell

src/guidellm/presentation/data_models.py

@@ -117,13 +117,13 @@ def from_benchmarks(cls, benchmarks: list["GenerativeBenchmark"]):
             range(len(successful_requests)), min(5, len(successful_requests))
         )
         sample_prompts = [
-            successful_requests[i].request_args.replace("\n", " ").replace('"', "'")
-            if successful_requests[i].request_args is not None
-            else ""
+            req.request_args.replace("\n", " ").replace('"', "'")
+            if (req := successful_requests[i]).request_args else ""
             for i in sample_indices
         ]
         sample_outputs = [
-            req.output.replace("\n", " ").replace('"', "'") if (req := successful_requests[i]).output else ""
+            req.output.replace("\n", " ").replace('"', "'")
+            if (req := successful_requests[i]).output else ""
             for i in sample_indices
         ]
 

src/guidellm/utils/encoding.py

@@ -32,7 +32,7 @@
 
     HAS_MSGSPEC = True
 except ImportError:
-    MsgspecDecoder = MsgspecEncoder = None
+    MsgspecDecoder = MsgspecEncoder = None  # type: ignore[misc, assignment] # HAS_MSGSPEC will be checked at runtime
     HAS_MSGSPEC = False
 
 

src/guidellm/utils/imports.py

@@ -3,7 +3,7 @@
 try:
     import orjson as json
 except ImportError:
-    import json
+    import json  # type: ignore[no-redef] # Done only after a failure.
 
 
 __all__ = ["json"]

src/guidellm/utils/registry.py

@@ -65,7 +65,7 @@ class TokenProposal(RegistryMixin):
     :cvar registry_populated: Track whether auto-discovery has completed
     """
 
-    registry: ClassVar[dict[str, RegistryObjT] | None] = None
+    registry: ClassVar[dict[str, RegistryObjT] | None] = None  # type: ignore[misc]
     registry_auto_discovery: ClassVar[bool] = False
     registry_populated: ClassVar[bool] = False
 

src/guidellm/utils/statistics.py

@@ -283,40 +283,12 @@ def from_request_times(
             )
 
         # First convert to timing events based on type
-        events: list[tuple[float, float]] = []
-
-        if distribution_type == "concurrency":
-            # For concurrency, each request adds to concurrency at start
-            # and subtracts at end
-            for (start, end), weight in zip(requests, weights, strict=False):
-                events.append((start, weight))
-                events.append((end, -1 * weight))
-        elif distribution_type == "rate":
-            # For rate, each request is added at the end time only
-            global_start = min(start for start, _ in requests) if requests else 0.0
-            events.append((global_start, 0.0))
-            for (_, end), weight in zip(requests, weights, strict=False):
-                events.append((end, weight))
-        else:
-            raise ValueError(
-                f"Invalid distribution_type '{distribution_type}'. "
-                "Must be 'concurrency' or 'rate'."
-            )
-
-        # Combine any events within epsilon of each other for stability
-        sorted_events = sorted(events, key=lambda event: event[0])
-        flattened_events: list[tuple[float, float]] = (
-            [sorted_events.pop(0)] if sorted_events else []
+        events = DistributionSummary._convert_to_timing_events(
+            requests, distribution_type, weights
         )
-        last_time = flattened_events[0][0] if flattened_events else 0.0
 
-        for time, val in sorted_events:
-            if abs(time - last_time) <= epsilon:
-                last_val = flattened_events[-1][1]
-                flattened_events[-1] = (last_time, last_val + val)
-            else:
-                last_time = time
-                flattened_events.append((time, val))
+        # Combine any events within epsilon of each other for stability
+        flattened_events = DistributionSummary._combine_events(events, epsilon)
 
         # Convert events to value distribution function
         distribution: dict[float, float] = defaultdict(float)
@@ -357,6 +329,53 @@ def from_request_times(
             include_cdf=include_cdf,
         )
 
+    @staticmethod
+    def _convert_to_timing_events(
+        requests: list[tuple[float, float]],
+        distribution_type: Literal["concurrency", "rate"],
+        weights: list[float],
+    ) -> list[tuple[float, float]]:
+        events: list[tuple[float, float]] = []
+
+        if distribution_type == "concurrency":
+            # For concurrency, each request adds to concurrency at start
+            # and subtracts at end
+            for (start, end), weight in zip(requests, weights, strict=False):
+                events.append((start, weight))
+                events.append((end, -1 * weight))
+        elif distribution_type == "rate":
+            # For rate, each request is added at the end time only
+            global_start = min(start for start, _ in requests) if requests else 0.0
+            events.append((global_start, 0.0))
+            for (_, end), weight in zip(requests, weights, strict=False):
+                events.append((end, weight))
+        else:
+            raise ValueError(
+                f"Invalid distribution_type '{distribution_type}'. "
+                "Must be 'concurrency' or 'rate'."
+            )
+        return events
+
+    @staticmethod
+    def _combine_events(
+        events: list[tuple[float, float]],
+        epsilon: float,
+    ) -> list[tuple[float, float]]:
+        sorted_events = sorted(events, key=lambda event: event[0])
+        flattened_events: list[tuple[float, float]] = (
+            [sorted_events.pop(0)] if sorted_events else []
+        )
+        last_time = flattened_events[0][0] if flattened_events else 0.0
+
+        for time, val in sorted_events:
+            if abs(time - last_time) <= epsilon:
+                last_val = flattened_events[-1][1]
+                flattened_events[-1] = (last_time, last_val + val)
+            else:
+                last_time = time
+                flattened_events.append((time, val))
+        return flattened_events
+
     @staticmethod
     def from_iterable_request_times(
         requests: list[tuple[float, float]],