fix typing issues

Author: DaltheCow

src/guidellm/backend/openai.py

@@ -130,7 +130,7 @@ def available_models(self) -> List[str]:
         :rtype: List[str]
         """
 
-        models: list[str] = [
+        models: List[str] = [
             model.id for model in self.openai_client.models.list().data
         ]
         logger.info(f"Available models: {models}")

src/guidellm/core/distribution.py

@@ -135,7 +135,7 @@ def percentile(self, percentile: float) -> float:
             logger.warning("No data points available to calculate percentile.")
             return 0.0
 
-        percentile_value = np.percentile(self._data, percentile)
+        percentile_value = np.percentile(self._data, percentile).item()
         logger.debug(f"Calculated {percentile}th percentile: {percentile_value}")
         return percentile_value
 

src/guidellm/core/result.py

@@ -80,7 +80,7 @@ def __eq__(self, other: object) -> bool:
         :rtype: bool
         """
 
-        if not isinstance(other, "TextGenerationResult"):
+        if not isinstance(other, TextGenerationResult):
             raise NotImplementedError(
                 "Only TextGenerationResult type could be used in that operation"
             )
@@ -232,6 +232,7 @@ def output_token(self, token: str):
 
     def end(
         self,
+        output: str,
         prompt_token_count: Optional[int] = None,
         output_token_count: Optional[int] = None,
     ):
@@ -253,7 +254,7 @@ def end(
         self._output_token_count = output_token_count or self._output_word_count
         self._prompt_token_count = prompt_token_count or self._prompt_word_count
 
-        logger.info(f"Text generation ended with output: '{self.output}'")
+        logger.info(f"Text generation ended with output: '{output}'")
 
 
 class TextGenerationError:
@@ -347,6 +348,8 @@ def __init__(self, mode: str, rate: Optional[float]):
         self._results: List[TextGenerationResult] = []
         self._errors: List[TextGenerationError] = []
         self._concurrencies: List[RequestConcurrencyMeasurement] = []
+        self._overloaded = False
+        self._args_rate: Optional[float] = None
 
         logger.debug(
             f"Initialized TextGenerationBenchmark with mode={mode} and rate={rate}"
@@ -400,6 +403,16 @@ def __iter__(self):
         """
         return iter(self._results)
 
+    @property
+    def overloaded(self) -> bool:
+        """
+        Get the overloaded state of the result.
+
+        :return: The overloaded state.
+        :rtype: bool
+        """
+        return self._overloaded
+
     @property
     def mode(self) -> str:
         """
@@ -410,6 +423,16 @@ def mode(self) -> str:
         """
         return self._mode
 
+    @property
+    def args_rate(self) -> Optional[float]:
+        """
+        Get the args rate of the result.
+
+        :return: The args rate.
+        :rtype: Optional[float]
+        """
+        return self._args_rate
+
     @property
     def rate(self) -> Optional[float]:
         """
@@ -631,3 +654,6 @@ def add_benchmark(self, benchmark: TextGenerationBenchmark):
         """
         self._benchmarks.append(benchmark)
         logger.debug(f"Added result: {benchmark}")
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {}

src/guidellm/executor/profile_generator.py

@@ -1,7 +1,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from enum import Enum
-from typing import Optional, Union
+from typing import Dict, Optional, Type, Union
 
 import numpy
 
@@ -29,7 +29,7 @@ class Profile:
 
 
 class ProfileGenerator(ABC):
-    _registry = {}
+    _registry: Dict[ProfileGenerationModes, "Type[ProfileGenerator]"] = {}
 
     @staticmethod
     def register_generator(mode: ProfileGenerationModes):

src/guidellm/main.py

@@ -10,6 +10,7 @@
     FileRequestGenerator,
     TransformersDatasetRequestGenerator,
 )
+from guidellm.request.base import RequestGenerator
 
 
 @click.command()
@@ -95,7 +96,7 @@ def main(
         )
 
     if data_type == "emulated":
-        request_generator = EmulatedRequestGenerator(config=data, tokenizer=tokenizer)
+        request_generator: RequestGenerator = EmulatedRequestGenerator(config=data, tokenizer=tokenizer)
     elif data_type == "file":
         request_generator = FileRequestGenerator(file_path=data, tokenizer=tokenizer)
     elif data_type == "transformers":

src/guidellm/request/emulated.py

@@ -2,7 +2,7 @@
 import re
 import unicodedata
 from dataclasses import dataclass
-from typing import Dict, List, Optional, Union
+from typing import Dict, List, Optional, Tuple, Union
 
 import numpy as np
 import requests
@@ -26,7 +26,7 @@ class EmulatedConfig:
     prompt_tokens_min: Optional[int] = None
     prompt_tokens_max: Optional[int] = None
 
-    generated_tokens: int = None
+    generated_tokens: Optional[int] = None
     generated_tokens_variance: Optional[int] = None
     generated_tokens_min: Optional[int] = None
     generated_tokens_max: Optional[int] = None
@@ -151,7 +151,7 @@ def _token_count(self, text: str) -> int:
             len(self.tokenizer.tokenize(text)) if self.tokenizer else len(text.split())
         )
 
-    def _sample_prompt(self) -> (str, int):
+    def _sample_prompt(self) -> Tuple[str, int]:
         prompt_tokens = self._config.prompt_tokens
         prompt_tokens_variance = self._config.prompt_tokens_variance or 0
         prompt_tokens_min = self._config.prompt_tokens_min or 1

src/guidellm/request/file.py

@@ -85,6 +85,8 @@ def _load_csv_file(self) -> List[str]:
         with open(self._file_path, "r", encoding="utf-8") as file:
             reader = csv.DictReader(file)
             columns = reader.fieldnames
+            if not columns:
+                raise ValueError("Invalid empty value for columns")
             for row in reader:
                 # convert the row to a dictionary
                 obj = {col: row[col] for col in columns}

src/guidellm/scheduler/scheduler.py

@@ -71,6 +71,8 @@ async def _run_async(self) -> TextGenerationBenchmark:
         result_set = TextGenerationBenchmark(
             mode=self._load_gen_mode.value, rate=self._load_gen_rate
         )
+        if (not self._load_gen_rate):
+            raise ValueError("Invalid empty value for self._load_gen_rate")
         load_gen = LoadGenerator(self._load_gen_mode, self._load_gen_rate)
 
         tasks = []