Add GitHub Actions CI/CD workflows and fix code formatting

Co-authored-by: DimaBir <28827735+DimaBir@users.noreply.github.com>

Author: Copilot

.github/workflows/lint.yml

@@ -0,0 +1,32 @@
+name: Code Quality
+
+on:
+  push:
+    branches: [ main, copilot/* ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.12'
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install ruff
+
+    - name: Lint with ruff
+      run: |
+        ruff check src/ common/ tests/ --output-format=github
+
+    - name: Check formatting with ruff
+      run: |
+        ruff format --check src/ common/ tests/

.github/workflows/tests.yml

@@ -0,0 +1,40 @@
+name: Tests
+
+on:
+  push:
+    branches: [ main, copilot/* ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ['3.10', '3.11', '3.12']
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+
+    - name: Run tests with coverage
+      run: |
+        pytest tests/ --cov=src --cov=common --cov-report=term-missing --cov-report=xml
+
+    - name: Upload coverage reports to Codecov
+      uses: codecov/codecov-action@v4
+      if: matrix.python-version == '3.12'
+      with:
+        file: ./coverage.xml
+        flags: unittests
+        name: codecov-umbrella
+        fail_ci_if_error: false

common/utils.py

@@ -1,5 +1,4 @@
 import argparse
-from typing import Dict, Tuple
 
 import matplotlib.pyplot as plt
 import pandas as pd
@@ -13,24 +12,34 @@
 INFERENCE_MODES = ["onnx", "ov", "cpu", "cuda", "tensorrt", "all"]
 
 
-def _create_sorted_dataframe(data: Dict[str, float], column_name: str, ascending: bool) -> pd.DataFrame:
+def _create_sorted_dataframe(
+    data: dict[str, float], column_name: str, ascending: bool
+) -> pd.DataFrame:
     df = pd.DataFrame(list(data.items()), columns=["Model", column_name])
     return df.sort_values(column_name, ascending=ascending)
 
 
-def _plot_bar_chart(ax, data: pd.DataFrame, x_col: str, y_col: str, 
-                    xlabel: str, ylabel: str, title: str, palette: str, value_format: str):
-    sns.barplot(x=data[x_col], y=data[y_col], hue=data[y_col], palette=palette, 
-                ax=ax, legend=False)
+def _plot_bar_chart(
+    ax,
+    data: pd.DataFrame,
+    x_col: str,
+    y_col: str,
+    xlabel: str,
+    ylabel: str,
+    title: str,
+    palette: str,
+    value_format: str,
+):
+    sns.barplot(x=data[x_col], y=data[y_col], hue=data[y_col], palette=palette, ax=ax, legend=False)
     ax.set_xlabel(xlabel)
     ax.set_ylabel(ylabel)
     ax.set_title(title)
-    
+
     for index, value in enumerate(data[x_col]):
         ax.text(value, index, value_format.format(value), color="black", ha="left", va="center")
 
 
-def plot_benchmark_results(results: Dict[str, Tuple[float, float]]):
+def plot_benchmark_results(results: dict[str, tuple[float, float]]):
     models = list(results.keys())
     times = {model: results[model][0] for model in models}
     throughputs = {model: results[model][1] for model in models}
@@ -40,13 +49,29 @@ def plot_benchmark_results(results: Dict[str, Tuple[float, float]]):
 
     fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(20, 6))
 
-    _plot_bar_chart(ax1, time_data, "Time", "Model", 
-                    "Average Inference Time (ms)", "Model Type",
-                    "ResNet50 - Inference Benchmark Results", "rocket", "{:.2f} ms")
+    _plot_bar_chart(
+        ax1,
+        time_data,
+        "Time",
+        "Model",
+        "Average Inference Time (ms)",
+        "Model Type",
+        "ResNet50 - Inference Benchmark Results",
+        "rocket",
+        "{:.2f} ms",
+    )
 
-    _plot_bar_chart(ax2, throughput_data, "Throughput", "Model",
-                    "Throughput (samples/sec)", "",
-                    "ResNet50 - Throughput Benchmark Results", "viridis", "{:.2f}")
+    _plot_bar_chart(
+        ax2,
+        throughput_data,
+        "Throughput",
+        "Model",
+        "Throughput (samples/sec)",
+        "",
+        "ResNet50 - Throughput Benchmark Results",
+        "viridis",
+        "{:.2f}",
+    )
 
     plt.tight_layout()
     plt.savefig(PLOT_OUTPUT_PATH, bbox_inches="tight")
@@ -66,10 +91,7 @@ def parse_arguments():
     )
 
     parser.add_argument(
-        "--topk", 
-        type=int, 
-        default=DEFAULT_TOPK, 
-        help="Number of top predictions to show"
+        "--topk", type=int, default=DEFAULT_TOPK, help="Number of top predictions to show"
     )
 
     parser.add_argument(

main.py

@@ -1,6 +1,5 @@
 import logging
 import warnings
-from typing import Dict, Tuple
 
 import torch
 
@@ -18,53 +17,58 @@
 CUDA_AVAILABLE = False
 if torch.cuda.is_available():
     try:
-        import torch_tensorrt
+        import torch_tensorrt  # noqa: F401
+
         CUDA_AVAILABLE = True
     except ImportError:
         print("torch-tensorrt not installed. Running in CPU mode only.")
 
 
-def _run_onnx_inference(args, model_loader, img_batch) -> Dict[str, Tuple[float, float]]:
+def _run_onnx_inference(args, model_loader, img_batch) -> dict[str, tuple[float, float]]:
     onnx_inference = ONNXInference(model_loader, args.onnx_path, debug_mode=args.DEBUG)
     benchmark_result = onnx_inference.benchmark(img_batch)
     onnx_inference.predict(img_batch)
     return {"ONNX (CPU)": benchmark_result}
 
 
-def _run_openvino_inference(args, model_loader, img_batch) -> Dict[str, Tuple[float, float]]:
+def _run_openvino_inference(args, model_loader, img_batch) -> dict[str, tuple[float, float]]:
     ov_inference = OVInference(model_loader, args.ov_path, debug_mode=args.DEBUG)
     benchmark_result = ov_inference.benchmark(img_batch)
     ov_inference.predict(img_batch)
     return {"OpenVINO (CPU)": benchmark_result}
 
 
-def _run_pytorch_cpu_inference(args, model_loader, img_batch) -> Dict[str, Tuple[float, float]]:
+def _run_pytorch_cpu_inference(args, model_loader, img_batch) -> dict[str, tuple[float, float]]:
     pytorch_cpu_inference = PyTorchInference(model_loader, device="cpu", debug_mode=args.DEBUG)
     benchmark_result = pytorch_cpu_inference.benchmark(img_batch)
     pytorch_cpu_inference.predict(img_batch)
     return {"PyTorch (CPU)": benchmark_result}
 
 
-def _run_pytorch_cuda_inference(args, model_loader, device, img_batch) -> Dict[str, Tuple[float, float]]:
+def _run_pytorch_cuda_inference(
+    args, model_loader, device, img_batch
+) -> dict[str, tuple[float, float]]:
     print("Running CUDA inference...")
     pytorch_cuda_inference = PyTorchInference(model_loader, device=device, debug_mode=args.DEBUG)
     benchmark_result = pytorch_cuda_inference.benchmark(img_batch)
     pytorch_cuda_inference.predict(img_batch)
     return {"PyTorch (CUDA)": benchmark_result}
 
 
-def _run_tensorrt_inference(args, model_loader, device, img_batch) -> Dict[str, Tuple[float, float]]:
+def _run_tensorrt_inference(
+    args, model_loader, device, img_batch
+) -> dict[str, tuple[float, float]]:
     results = {}
     precisions = [torch.float16, torch.float32]
-    
+
     for precision in precisions:
         tensorrt_inference = TensorRTInference(
             model_loader, device=device, precision=precision, debug_mode=args.DEBUG
         )
         benchmark_result = tensorrt_inference.benchmark(img_batch)
         tensorrt_inference.predict(img_batch)
         results[f"TRT_{precision}"] = benchmark_result
-    
+
     return results
 
 
@@ -76,7 +80,7 @@ def main():
 
     benchmark_results = {}
     device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-    
+
     model_loader = ModelLoader(device=device)
     img_processor = ImageProcessor(img_path=args.image_path, device=device)
     img_batch = img_processor.process_image()
@@ -92,7 +96,9 @@ def main():
 
     if torch.cuda.is_available():
         if args.mode in ["cuda", "all"]:
-            benchmark_results.update(_run_pytorch_cuda_inference(args, model_loader, device, img_batch))
+            benchmark_results.update(
+                _run_pytorch_cuda_inference(args, model_loader, device, img_batch)
+            )
 
         if args.mode in ["tensorrt", "all"]:
             benchmark_results.update(_run_tensorrt_inference(args, model_loader, device, img_batch))

src/image_processor.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 import torch
 from PIL import Image
 from torchvision import transforms
@@ -11,19 +9,21 @@
 
 
 class ImageProcessor:
-    def __init__(self, img_path: str, device: Union[str, torch.device] = "cuda") -> None:
+    def __init__(self, img_path: str, device: str | torch.device = "cuda") -> None:
         self.img_path = img_path
         self.device = device if isinstance(device, torch.device) else torch.device(device)
         self.transform = self._create_transform()
 
     @staticmethod
     def _create_transform() -> transforms.Compose:
-        return transforms.Compose([
-            transforms.Resize(IMAGE_SIZE),
-            transforms.CenterCrop(CROP_SIZE),
-            transforms.ToTensor(),
-            transforms.Normalize(IMAGENET_MEAN, IMAGENET_STD),
-        ])
+        return transforms.Compose(
+            [
+                transforms.Resize(IMAGE_SIZE),
+                transforms.CenterCrop(CROP_SIZE),
+                transforms.ToTensor(),
+                transforms.Normalize(IMAGENET_MEAN, IMAGENET_STD),
+            ]
+        )
 
     def process_image(self) -> torch.Tensor:
         img = Image.open(self.img_path)

src/inference_base.py

@@ -1,6 +1,5 @@
 import logging
 import time
-from typing import Optional, Tuple
 
 import numpy as np
 import torch
@@ -17,8 +16,8 @@ class InferenceBase:
     def __init__(
         self,
         model_loader,
-        onnx_path: Optional[str] = None,
-        ov_path: Optional[str] = None,
+        onnx_path: str | None = None,
+        ov_path: str | None = None,
         topk: int = DEFAULT_TOPK,
         debug_mode: bool = False,
         batch_size: int = DEFAULT_BATCH_SIZE,
@@ -52,46 +51,48 @@ def _warmup(self, input_batch: torch.Tensor, warmup_runs: int):
             for img in input_batch:
                 self.predict(img.unsqueeze(0), is_benchmark=True)
 
-    def _run_benchmark(self, input_batch: torch.Tensor, num_runs: int) -> Tuple[float, int]:
+    def _run_benchmark(self, input_batch: torch.Tensor, num_runs: int) -> tuple[float, int]:
         logging.info(f"Starting benchmark for {self.__class__.__name__} inference...")
         start_time = time.time()
         for _ in range(num_runs):
             for img in input_batch:
                 self.predict(img.unsqueeze(0), is_benchmark=True)
-        
+
         elapsed_time = time.time() - start_time
         total_samples = self.batch_size * num_runs
         return elapsed_time, total_samples
 
-    def _calculate_metrics(self, elapsed_time: float, total_samples: int) -> Tuple[float, float]:
+    def _calculate_metrics(self, elapsed_time: float, total_samples: int) -> tuple[float, float]:
         avg_time = (elapsed_time / total_samples) * MS_PER_SECOND
         throughput = total_samples / elapsed_time
-        
+
         logging.info(f"Average inference time: {avg_time:.4f} ms")
         logging.info(f"Throughput: {throughput:.2f} samples/sec")
-        
+
         if self.debug_mode:
             print(f"Average inference time for {self.__class__.__name__}: {avg_time:.4f} ms")
             print(f"Throughput for {self.__class__.__name__}: {throughput:.2f} samples/sec")
-        
+
         return avg_time, throughput
 
     def benchmark(
-        self, 
-        input_data: torch.Tensor, 
-        num_runs: int = DEFAULT_NUM_RUNS, 
-        warmup_runs: int = DEFAULT_WARMUP_RUNS
-    ) -> Tuple[float, float]:
+        self,
+        input_data: torch.Tensor,
+        num_runs: int = DEFAULT_NUM_RUNS,
+        warmup_runs: int = DEFAULT_WARMUP_RUNS,
+    ) -> tuple[float, float]:
         input_batch = self._prepare_batch(input_data)
         self._warmup(input_batch, warmup_runs)
         elapsed_time, total_samples = self._run_benchmark(input_batch, num_runs)
         return self._calculate_metrics(elapsed_time, total_samples)
 
-    def get_top_predictions(self, prob: np.ndarray, is_benchmark: bool = False) -> Optional[np.ndarray]:
+    def get_top_predictions(
+        self, prob: np.ndarray, is_benchmark: bool = False
+    ) -> np.ndarray | None:
         if is_benchmark:
             return None
 
-        top_indices = prob.argsort()[-self.topk:][::-1]
+        top_indices = prob.argsort()[-self.topk :][::-1]
         top_probs = prob[top_indices]
 
         for i in range(self.topk):
@@ -100,5 +101,5 @@ def get_top_predictions(self, prob: np.ndarray, is_benchmark: bool = False) -> O
             logging.info(f"#{i + 1}: {int(probability * 100)}% {class_label}")
             if self.debug_mode:
                 print(f"#{i + 1}: {int(probability * 100)}% {class_label}")
-        
+
         return prob