Initial benchmark utility

.gitignore

@@ -1,4 +1,5 @@
 /target
+output/
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

benchmark/__init__.py

@@ -0,0 +1,5 @@
+from pathlib import Path
+
+PACKAGE_ROOT = Path(__file__).parent.resolve()
+PROJECT_ROOT = PACKAGE_ROOT.parent
+OUT_DIR = PROJECT_ROOT / "output"

benchmark/__main__.py

@@ -0,0 +1,42 @@
+import argparse
+import sys
+from datetime import datetime
+
+from benchmark.manager import TestManager
+
+
+def unload_arcade():
+    to_uncache = []
+    for mod in sys.modules:
+        if mod.startsWith("arcade."):
+            to_uncache.append(mod)
+
+    for mod in to_uncache:
+        del sys.modules[mod]
+
+
+def main():
+    args = parse_args(sys.argv[1:])
+    print(f"Session Name: '{args.session}'")
+    manager = TestManager(args.session, debug=True)
+    manager.find_test_classes(args.type, args.name)
+    manager.create_test_instances()
+    manager.run()
+
+
+def parse_args(args):
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "-s",
+        "--session",
+        help="Session Name",
+        type=str,
+        default=datetime.now().strftime("%Y-%m-%dT%H-%M-%S"),
+    )
+    parser.add_argument("-t", "--type", help="Test Type", type=str)
+    parser.add_argument("-n", "--name", help="Test Name", type=str)
+    return parser.parse_args(args)
+
+
+if __name__ == "__main__":
+    main()

benchmark/graph.py

@@ -0,0 +1,87 @@
+import csv
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+import seaborn as sns
+
+sns.set_style("whitegrid")
+
+FPS = 1
+SPRITE_COUNT = 2
+DRAWING_TIME = 3
+PROCESSING_TIME = 4
+
+
+class DataSeries:
+    def __init__(self, name: str, path: Path) -> None:
+        self.name = name
+        self.path = path
+        # Data
+        self.count = []
+        self.processing_time = []
+        self.draw_time = []
+        self.fps = []
+        # Process data
+        self._process_data()
+
+    def _process_data(self):
+        rows = self._read_file(self.path)
+        for row in rows:
+            self.count.append(row[SPRITE_COUNT])
+            self.fps.append(row[FPS])
+            self.processing_time.append(row[PROCESSING_TIME])
+            self.draw_time.append(row[DRAWING_TIME])
+
+    def _read_file(self, path: Path):
+        results = []
+        with open(path) as csv_file:
+            csv_reader = csv.reader(csv_file, delimiter=",")
+            first_row = True
+            for row in csv_reader:
+                if first_row:
+                    first_row = False
+                else:
+                    results.append([float(cell) for cell in row])
+
+            return results
+
+
+class PerfGraph:
+    def __init__(self, title: str, label_x: str, label_y: str) -> None:
+        self.title = title
+        self.label_x = label_x
+        self.label_y = label_y
+        self.series = []
+
+    def add_series(self, series: DataSeries):
+        self.series.append(series)
+
+    def create(self, output_path: Path):
+        plt.title(self.title)
+
+        for series in self.series:
+            plt.plot(series.count, series.processing_time, label=series.name)
+
+        plt.legend(loc="upper left", shadow=True, fontsize="large")
+        plt.xlabel(self.label_x)
+        plt.ylabel(self.label_y)
+
+        plt.savefig(output_path)
+        plt.clf()
+
+
+if __name__ == "__main__":
+    from benchmark import OUT_DIR
+
+    OUTPUT_ROOT = OUT_DIR / "test" / "graphs"
+    OUTPUT_ROOT.mkdir(parents=True, exist_ok=True)
+    path = OUT_DIR / "test" / "data"
+
+    graph = PerfGraph(
+        "Time To Detect Collisions", label_x="Sprite Count", label_y="Time"
+    )
+    graph.add_series(DataSeries("Arcade 0", path / "arcade_collision-0.csv"))
+    graph.add_series(DataSeries("Arcade 1", path / "arcade_collision-1.csv"))
+    graph.add_series(DataSeries("Arcade 2", path / "arcade_collision-2.csv"))
+    graph.add_series(DataSeries("Arcade 3", path / "arcade_collision-3.csv"))
+    graph.create(OUTPUT_ROOT / "arcade_collision.png")

benchmark/manager.py

@@ -0,0 +1,160 @@
+import importlib
+import pkgutil
+from typing import List, Optional, Type
+
+from benchmark import OUT_DIR
+from benchmark.graph import DataSeries, PerfGraph
+from benchmark.tests.base import PerfTest
+
+
+def find_test_classes(path: str) -> List[Type[PerfTest]]:
+    """Find all test classes in submodules"""
+    target_module = importlib.import_module(f"benchmark.tests.{path}")
+
+    classes = []
+    for v in pkgutil.iter_modules(target_module.__path__):
+        module = importlib.import_module(f"benchmark.tests.{path}.{v.name}")
+        if hasattr(module, "Test"):
+            classes.append(module.Test)
+        else:
+            print(
+                (
+                    "WARNING: "
+                    f"Module '{module.__name__}' does not have a Test class. "
+                    "Please add a test class or rename the class to 'Test'."
+                )
+            )
+
+    return classes
+
+
+class TestManager:
+    """
+    Finds and executes tests
+
+    :param str session: The session name.
+    :param bool debug: If True, print debug messages.
+    """
+
+    def __init__(self, session: str, debug: bool = True):
+        self.debug = debug
+        self.session = session
+        self.session_dir = OUT_DIR / session
+        self.session_dir.mkdir(parents=True, exist_ok=True)
+        self.data_dir = self.session_dir / "data"
+
+        self.test_classes: List[Type[PerfTest]] = []
+        self.test_instances: List[PerfTest] = []
+
+    @property
+    def num_test_classes(self) -> int:
+        return len(self.test_classes)
+
+    @property
+    def num_test_instances(self) -> int:
+        return len(self.test_instances)
+
+    def find_test_classes(
+        self,
+        type: Optional[str] = None,
+        name: Optional[str] = None,
+    ):
+        """
+        Find test classes based on type and name.
+
+        :param str type: The type of test to run.
+        :param str name: The name of the test to run.
+        :return: The number of test classes found.
+        """
+        all_classes = find_test_classes("arcade")
+        all_classes += find_test_classes("arcade_accelerate")
+
+        for cls in all_classes:
+            if type is not None and cls.type != type:
+                continue
+            if name is not None and cls.name != name:
+                continue
+            self.test_classes.append(cls)
+
+        if self.debug:
+            num_classes = len(self.test_classes)
+            print(f"Found {num_classes} test classes")
+            for cls in self.test_classes:
+                print(f" -> {cls.type}.{cls.name}")
+
+    def create_test_instances(self):
+        """
+        Create test instances based on each test's instances attribute.
+        """
+        for cls in self.test_classes:
+            # If a test have multiple instances, create one instance for each
+            if cls.instances:
+                for params, _ in cls.instances:
+                    self.add_test_instance(cls(**params))
+            else:
+                self.add_test_instance(cls())
+
+        if self.debug:
+            num_instances = len(self.test_instances)
+            print(f"Created {num_instances} test instances")
+            for instance in self.test_instances:
+                print(f" -> {instance.type}.{instance.name}")
+
+    def add_test_instance(self, instance: PerfTest):
+        """Validate instance"""
+        if instance.name == "default":
+            raise ValueError(
+                (
+                    "Test name cannot be 'default'."
+                    "Please add a class attribute 'name' to your test class."
+                    f"Class: {instance}"
+                )
+            )
+        self.test_instances.append(instance)
+
+    def get_test_instance(self, name: str) -> Optional[PerfTest]:
+        for instance in self.test_instances:
+            if instance.instance_name == name:
+                return instance
+
+    def run(self):
+        """Run all tests"""
+        for instance in self.test_instances:
+            instance.run(self.session_dir)
+
+    def create_graph(
+        self,
+        file_name: str,
+        title: str,
+        x_label: str,
+        y_label: str,
+        series_names=[],
+    ):
+        """Create a graph using matplotlib"""
+        print("Creating graph : {title}} [{x_label}, {y_label}]}]")
+        series = []
+        skip = False
+        for _series in series_names:
+            # Check if we have a test instance with this name
+            instance = self.get_test_instance(_series)
+            if instance is None:
+                print(f" -> No test instance found for series '{_series}'")
+                skip = True
+
+            path = self.data_dir / f"{_series}.csv"
+            if not path.exists():
+                print(
+                    f"No data found for series '{_series}' in session '{self.session}'"
+                )
+                skip = True
+
+            if skip:
+                continue
+
+            series.append(DataSeries(instance.name, path))
+
+        out_path = self.session_dir / "graphs"
+        out_path.mkdir(parents=True, exist_ok=True)
+        out_path = out_path / f"{file_name}.png"
+        graph = PerfGraph(title, x_label, y_label, series)
+        graph.create(out_path)

benchmark/tests/arcade/__init__.py

@@ -0,0 +1 @@
+from . import collision