Spaces:
Running
Running
File size: 1,729 Bytes
a083fd4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 |
# Copyright (C) 2024 Intel Corporation
# SPDX-License-Identifier: Apache-2.0
#
"""Synchronous Executor based on ModelAPI."""
from __future__ import annotations
import time
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from demo_package.model_wrapper import ModelWrapper
from demo_package.visualizers import BaseVisualizer
from demo_package.streamer.streamer import get_streamer
from demo_package.visualizers import dump_frames
class SyncExecutor:
"""Synchronous executor for model inference.
Args:
model (ModelContainer): model for inference
visualizer (Visualizer): visualizer of inference results. Defaults to None.
"""
def __init__(self, model: ModelWrapper, visualizer: BaseVisualizer) -> None:
self.model = model
self.visualizer = visualizer
def run(self, input_stream: int | str, loop: bool = False) -> None:
"""Run demo using input stream (image, video stream, camera)."""
streamer = get_streamer(input_stream, loop)
saved_frames = []
for frame in streamer:
# getting result include preprocessing, infer, postprocessing for sync infer
start_time = time.perf_counter()
predictions, _ = self.model(frame)
output = self.visualizer.draw(frame, predictions)
self.visualizer.show(output)
if output is not None:
saved_frames.append(output)
if self.visualizer.is_quit():
break
# visualize video not faster than the original FPS
self.visualizer.video_delay(time.perf_counter() - start_time, streamer)
dump_frames(saved_frames, self.visualizer.output, input_stream, streamer)
|