def run()

in optimum_benchmark/scenarios/energy_star/scenario.py [0:0]
61 lines of code
11 McCabe index (conditional complexity)

    def run(self, backend: Backend[BackendConfigT]) -> BenchmarkReport:
        self.backend = backend

        if self.backend.config.task in TEXT_GENERATION_TASKS:
            self.logger.info("\t+ Updating Text Generation kwargs with default values")
            self.config.generate_kwargs = {**TEXT_GENERATION_DEFAULT_KWARGS, **self.config.generate_kwargs}
            self.logger.info("\t+ Initializing Text Generation report")
            self.report = BenchmarkReport.from_list(
                targets=["load_dataset", "preprocess_dataset", "load_model", "prefill", "decode"]
            )
        elif self.backend.config.task in IMAGE_DIFFUSION_TASKS:
            self.logger.info("\t+ Updating Image Diffusion kwargs with default values")
            self.config.call_kwargs = {**IMAGE_DIFFUSION_DEFAULT_KWARGS, **self.config.call_kwargs}
            self.logger.info("\t+ Initializing Image Diffusion report")
            self.report = BenchmarkReport.from_list(
                targets=["load_dataset", "preprocess_dataset", "load_model", "call"]
            )
        else:
            self.logger.info("\t+ Initializing Inference report")
            self.report = BenchmarkReport.from_list(
                targets=["load_dataset", "preprocess_dataset", "load_model", "forward"]
            )

        if self.config.latency:
            self.logger.info("\t+ Initializing Latency tracker")
            self.latency_tracker = LatencyTracker(
                backend=self.backend.config.name,
                device=self.backend.config.device,
            )
        if self.config.memory:
            self.logger.info("\t+ Initializing Memory tracker")
            self.memory_tracker = MemoryTracker(
                backend=self.backend.config.name,
                device=self.backend.config.device,
                device_ids=self.backend.config.device_ids,
            )
        if self.config.energy:
            self.logger.info("\t+ Initializing Energy tracker")
            self.energy_tracker = EnergyTracker(
                backend=self.backend.config.name,
                device=self.backend.config.device,
                device_ids=self.backend.config.device_ids,
            )

        # we start with loading/preprocessing the dataset as it takes no vram
        self.run_dataset_loading_tracking()
        self.run_dataset_preprocessing_tracking()
        self.run_model_loading_tracking()

        if self.config.warmup_runs > 0:
            self.logger.info("\t+ Preparing sample inputs for warmup")
            self.sample_inputs = self.dataset[: self.config.input_shapes["batch_size"]]
            self.sample_inputs = self.backend.prepare_inputs(self.sample_inputs)

            if self.backend.config.task in TEXT_GENERATION_TASKS:
                self.warmup_text_generation()
            elif self.backend.config.task in IMAGE_DIFFUSION_TASKS:
                self.warmup_image_diffusion()
            else:
                self.warmup_inference()

        if self.backend.config.task in TEXT_GENERATION_TASKS:
            self.run_text_generation_tracking()
        elif self.backend.config.task in IMAGE_DIFFUSION_TASKS:
            self.run_image_diffusion_tracking()
        else:
            self.run_inference_tracking()

        return self.report