path: root/tests/performance/api/config.py

# Apache License, Version 2.0

import fnmatch
import json
import pathlib
import sys

from dataclasses import dataclass, field
from typing import Dict, List

from .test import TestCollection


def get_build_hash(args: None) -> str:
    import bpy
    import sys
    build_hash = bpy.app.build_hash.decode('utf-8')
    return '' if build_hash == 'Unknown' else build_hash


@dataclass
class TestEntry:
    """Test to run, a combination of revision, test and device."""
    test: str = ''
    category: str = ''
    revision: str = ''
    git_hash: str = ''
    executable: str = ''
    date: int = 0
    device_type: str = 'CPU'
    device_id: str = 'CPU'
    device_name: str = 'Unknown CPU'
    status: str = 'queued'
    output: Dict = field(default_factory=dict)
    benchmark_type: str = 'comparison'

    def to_json(self) -> Dict:
        json_dict = {}
        for field in self.__dataclass_fields__:
            json_dict[field] = getattr(self, field)
        return json_dict

    def from_json(self, json_dict):
        for field in self.__dataclass_fields__:
            setattr(self, field, json_dict[field])


class TestQueue:
    """Queue of tests to be run or inspected. Matches JSON file on disk."""

    def __init__(self, filepath: pathlib.Path):
        self.filepath = filepath
        self.has_multiple_revisions_to_build = False
        self.has_multiple_categories = False
        self.entries = []

        if self.filepath.is_file():
            with open(self.filepath, 'r') as f:
                json_entries = json.load(f)

            for json_entry in json_entries:
                entry = TestEntry()
                entry.from_json(json_entry)
                self.entries.append(entry)

    def rows(self, use_revision_columns: bool) -> List:
        # Generate rows of entries for printing and running.
        entries = sorted(self.entries, key=lambda entry:
            (entry.revision,
             entry.device_id,
             entry.category,
             entry.test))

        if not use_revision_columns:
            # One entry per row.
            return [[entry] for entry in entries]
        else:
            # Multiple revisions per row.
            rows = {}

            for entry in entries:
                key = (entry.device_id, entry.category, entry.test)
                if key in rows:
                    rows[key].append(entry)
                else:
                    rows[key] = [entry]

            return [value for _, value in sorted(rows.items())]

    def find(self, revision: str, test: str, category: str, device_id: str) -> Dict:
        for entry in self.entries:
            if entry.revision == revision and \
               entry.test == test and \
               entry.category == category and \
               entry.device_id == device_id:
                return entry

        return None

    def write(self) -> None:
        json_entries = [entry.to_json() for entry in self.entries]
        with open(self.filepath, 'w') as f:
            json.dump(json_entries, f, indent=2)


class TestConfig:
    """Test configuration, containing a subset of revisions, tests and devices."""

    def __init__(self, env, name: str):
        # Init configuration from config.py file.
        self.name = name
        self.base_dir = env.base_dir / name
        self.logs_dir = self.base_dir / 'logs'

        config = self._read_config_module()
        self.tests = TestCollection(env,
                                    getattr(config, 'tests', ['*']),
                                    getattr(config, 'categories', ['*']))
        self.revisions = getattr(config, 'revisions', {})
        self.builds = getattr(config, 'builds', {})
        self.queue = TestQueue(self.base_dir / 'results.json')
        self.benchmark_type = getattr(config, 'benchmark_type', 'comparison')

        self.devices = []
        self._update_devices(env, getattr(config, 'devices', ['CPU']))

        self._update_queue(env)

    def revision_names(self) -> List:
        return sorted(list(self.revisions.keys()) + list(self.builds.keys()))

    def device_name(self, device_id: str) -> str:
        for device in self.devices:
            if device.id == device_id:
                return device.name

        return "Unknown"

    @staticmethod
    def write_default_config(env, config_dir: pathlib.Path) -> None:
        config_dir.mkdir(parents=True, exist_ok=True)

        default_config = """devices = ['CPU']\n"""
        default_config += """tests = ['*']\n"""
        default_config += """categories = ['*']\n"""
        default_config += """builds = {\n"""
        default_config += """    'master': '/home/user/blender-git/build/bin/blender',"""
        default_config += """    '2.93': '/home/user/blender-2.93/blender',"""
        default_config += """}\n"""
        default_config += """revisions = {\n"""
        default_config += """}\n"""

        config_file = config_dir / 'config.py'
        with open(config_file, 'w') as f:
            f.write(default_config)

    def _read_config_module(self) -> None:
        # Import config.py as a module.
        import importlib.util
        spec = importlib.util.spec_from_file_location("testconfig", self.base_dir / 'config.py')
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        return mod

    def _update_devices(self, env, device_filters: List) -> None:
        # Find devices matching the filters.
        need_gpus = device_filters != ['CPU']
        machine = env.get_machine(need_gpus)

        self.devices = []
        for device in machine.devices:
            for device_filter in device_filters:
                if fnmatch.fnmatch(device.id, device_filter):
                    self.devices.append(device)
                    break

    def _update_queue(self, env) -> None:
        # Update queue to match configuration, adding and removing entries
        # so that there is one entry for each revision, device and test
        # combination.
        entries = []

        # Get entries for specified commits, tags and branches.
        for revision_name, revision_commit in self.revisions.items():
            git_hash = env.resolve_git_hash(revision_commit)
            date = env.git_hash_date(git_hash)
            entries += self._get_entries(revision_name, git_hash, '', date)

        # Optimization to avoid rebuilds.
        revisions_to_build = set()
        for entry in entries:
            if entry.status in ('queued', 'outdated'):
                revisions_to_build.add(entry.git_hash)
        self.queue.has_multiple_revisions_to_build = len(revisions_to_build) > 1

        # Get entries for revisions based on existing builds.
        for revision_name, executable in self.builds.items():
            executable_path = pathlib.Path(executable)
            if not executable_path.exists():
                sys.stderr.write(f'Error: build {executable} not found\n')
                sys.exit(1)

            env.set_blender_executable(executable_path)
            git_hash, _ = env.run_in_blender(get_build_hash, {})
            env.unset_blender_executable()

            mtime = executable_path.stat().st_mtime
            entries += self._get_entries(revision_name, git_hash, executable, mtime)

        # Detect number of categories for more compact printing.
        categories = set()
        for entry in entries:
            categories.add(entry.category)
        self.queue.has_multiple_categories = len(categories) > 1

        # Replace actual entries.
        self.queue.entries = entries

    def _get_entries(self,
                     revision_name: str,
                     git_hash: str,
                     executable: pathlib.Path,
                     date: int) -> None:
        entries = []
        for test in self.tests.tests:
            test_name = test.name()
            test_category = test.category()

            for device in self.devices:
                entry = self.queue.find(revision_name, test_name, test_category, device.id)
                if entry:
                    # Test if revision hash or executable changed.
                    if entry.git_hash != git_hash or \
                       entry.executable != executable or \
                       entry.benchmark_type != self.benchmark_type or \
                       entry.date != date:
                        # Update existing entry.
                        entry.git_hash = git_hash
                        entry.executable = executable
                        entry.benchmark_type = self.benchmark_type
                        entry.date = date
                        if entry.status in ('done', 'failed'):
                            entry.status = 'outdated'
                else:
                    # Add new entry if it did not exist yet.
                    entry = TestEntry(
                        revision=revision_name,
                        git_hash=git_hash,
                        executable=executable,
                        date=date,
                        test=test_name,
                        category=test_category,
                        device_type=device.type,
                        device_id=device.id,
                        device_name=device.name,
                        benchmark_type=self.benchmark_type)
                entries.append(entry)

        return entries