import logging
import threading
import time
from os import path as osp
from transitions.extensions import LockedMachine as Machine
from transitions import State as State
from blinker import signal
import defaults
import errors
import vs
import utils
import glfw
from algorithm.algorithm_manager import AlgorithmManager
from clientmessenger import CLIENT_MESSENGER
from utils.settings_manager import SETTINGS
from utils import async
logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())
FAIL_SLEEP = 0.25
class Stitcher(object):
""" Stitcher automaton
- Responsible for the stitching process
- Holds references to streams (preview, broadcast)
- Holds references to recorders (input, output)
"""
def __init__(self, server_instance, stitcher_executor, display_):
"""Init
"""
self.server = server_instance
self.stitcher_executor = stitcher_executor
self.algorithm_manager = None
self.has_audio = None
self.project_manager = self.server.project_manager
self.display = display_
self.old_submit = self.stitcher_executor.submit
self.overlay_window = None
self.overlay_config = None
self.overlay = None
# State Machine
states = [
# The stitcher is waiting for the camera to connect
State(name="WaitingForCamera", ignore_invalid_triggers=True),
# Stitcher running, no errors
State(name="Running", ignore_invalid_triggers=True),
# Stitcher running, but producing errors (EOF, etc..)
State(name="Failing", ignore_invalid_triggers=True),
]
transitions = [
{"source": "WaitingForCamera",
"trigger": "t_camera_connected",
"dest": "Failing"},
{"source": "Running",
"trigger": "t_fail",
"dest": "Failing"},
{"source": "Failing",
"trigger": "t_resume",
"dest": "Running"},
{"source": ["Running", "Failing"],
"trigger": "t_camera_connected",
"conditions": "_ignore",
"dest": "Failing"},
{"source": ["Running", "Failing"],
"trigger": "t_camera_disconnected",
"before": "terminate",
"dest": "WaitingForCamera"},
{"source": ["Running", "Failing"],
"trigger": "t_terminate",
"dest": "WaitingForCamera"},
]
self.machine = Machine(
name="Stitcher", model=self, states=states,
transitions=transitions, initial="WaitingForCamera", async=True)
# Signals
self._connect_signals()
self._load_plugins()
self.run_loop = None
self.native_loop = None
def _connect_signals(self):
signal("camera_connected").connect(self.t_camera_connected)
signal("camera_disconnected").connect(self.t_camera_disconnected)
signal("camera_video_fail").connect(self.t_camera_disconnected)
signal("algorithm_running_success").connect(self._reset_executor)
def _submit(self, fn, *args, **kwargs):
self.native_loop.Lock()
result = self.old_submit(fn, *args, **kwargs)
self.native_loop.Unlock()
return result
def _reset_executor(self, sender, panorama, output):
self.stitcher_executor.submit(self.reset_panorama, sender, panorama, output)
# ------------------------------------------------------------------------
#
# ------------------------------------------------------------------------
def _ignore(self, sender=None):
return False
def on_enter_WaitingForCamera(self, sender=None):
self._cleanup()
def on_exit_WaitingForCamera(self, sender=None):
""" Create stitcher"""
self._create()
def start_streams(self):
signal("stitching_running").send()
def on_enter_Running(self, sender=None):
self.stitcher_executor.submit(self.start_streams).result()
if SETTINGS.enable_ev_compensation:
# Note, if stitcher fails and algorithm won't be completed - it won't be rescheduled
# So we want to start it again when stitcher is running
self.algorithm_manager.start_exposure_compensation()
def on_exit_Running(self, sender=None):
signal("stitching_failing").send(preserve=SETTINGS.output_recovery_enabled)
# ------------------------------------------------------------------------
#
# ------------------------------------------------------------------------
def _load_plugins(self):
if vs.loadPlugins(osp.join(SETTINGS.lib_path, "plugins")) == 0:
raise errors.StitcherError("Parsing: could not load any plugin")
def _create(self):
""" Creates the controller and the stitcher
"""
status = vs.checkDefaultBackendDeviceInitialization()
if not status.ok():
logger.error("CUDA device Initialization failed. Reason {}".format(status.getErrorMessage()))
raise errors.StitcherError("CUDA device Initialization failed")
try:
self.project_manager.create_controller()
except (errors.StitcherError, errors.AudioError, errors.ParserError) as error:
logger.warn("current configuration seems corrupted : {} "
"Switching to default configuration.".format(error))
CLIENT_MESSENGER.send_event(defaults.RESET_TO_DEFAULT_MESSAGE)
utils.async.defer(self.server.reset, True)
return
status = self.project_manager.controller.createStitcher()
if not status.ok():
raise errors.StitcherError("Cannot create stitcher: " + str(status.getErrorMessage()))
self._start_stitching_thread()
self.algorithm_manager = AlgorithmManager(self.project_manager.controller)
self.renderer = None
if self.display.is_active():
self.display.reset_pano()
self.display.open_output_window()
surfaces = self.display.get_surfaces()
if surfaces is None:
raise errors.StitcherError("Error creating destination surface")
self.renderer = self.display.get_renderer()
self.renderer.thisown = 0
self.stitcher_executor.submit = self._submit
#overlay create
self.overlay_config = self.project_manager.project_config.has("overlay")
if self.overlay_config :
glfw.default_window_hints()
glfw.window_hint(glfw.VISIBLE, False)
glfw.window_hint(glfw.CONTEXT_ROBUSTNESS, glfw.NO_RESET_NOTIFICATION)
glfw.window_hint(glfw.CONTEXT_RELEASE_BEHAVIOR, glfw.RELEASE_BEHAVIOR_FLUSH)
self.overlay_window = glfw.create_window(16, 16, "", None, self.display.offscreen_window)
glfw.make_context_current(self.overlay_window)
self.overlay = vs.Compositor(self.display._to_swig(self.overlay_window), self.project_manager.panorama, self.project_manager.controller.getFrameRateFromInputController())
glfw.make_context_current(None)
else:
surfaces = vs.PanoSurfaceVec()
surf = vs.OffscreenAllocator_createPanoSurface(self.project_manager.panorama.width,
self.project_manager.panorama.height,
"StitchOutput")
if not surf.ok():
raise errors.StitcherError("Error creating destination surface " + \
str(surf.status().getErrorMessage()))
dontmemleak = vs.panoSurfaceSharedPtr(surf.release())
# SWIG create a proxy object with an empty deleter
# when passing directly the pointer to the vector object :(
# DON'T TRY TO FACTORIZE THE PREVIOUS LINE OR MEMLEAK
surfaces.push_back(dontmemleak)
self.stitcher_executor.submit = self.old_submit
self.stitch_output = self.project_manager.controller.createAsyncStitchOutput(
surfaces, vs.PanoRendererPtrVector(), vs.OutputWriterPtrVector())
if self.display.is_active():
self.stitch_output.addRenderer(vs.panoRendererSharedPtr(self.renderer))
self.native_loop = vs.StitchLoop(self.display._to_swig(self.display.offscreen_window), self.project_manager.controller, self.stitch_output.object())
if self.overlay:
dontmemleak = vs.overlayerSharedPtr(self.overlay)
# SWIG create a proxy object with an empty deleter
# when passing directly the pointer to the setCompositor function :(
# DON'T TRY TO FACTORIZE THE PREVIOUS LINE OR MEMLEAK
self.stitch_output.setCompositor(dontmemleak)
self.run_loop.set()
# To be deprecated
self.has_audio = False
def _start_stitching_thread(self):
self.run_loop = threading.Event()
self.thread = threading.Thread(target=self._stitch_loop, name="Stitcher")
self.thread.start()
# Stitching process management
def _cleanup(self):
"""Makes sure that all the lib objects are released to prevent assert at exit time
"""
if self.display.is_active():
self.display.close_output_window()
self.stitch_output.removeRenderer("OpenGLRenderer")
self.stitch_output = None
self.algorithm_manager = None
def _check_status(self, status):
current_state = self.state
if status is not None:
if current_state == "WaitingForCamera":
time.sleep(FAIL_SLEEP)
return None
if status.ok():
if current_state == "Failing":
async.defer(self.t_resume)
return status
else:
msg = status.getErrorMessage()
if current_state == "Running":
logger.error("Stitching failed : " + msg)
async.defer(self.t_fail)
return status
# unrecoverable GPU failure. Server need to be resetted
elif self.state == "Failing" and status.getOriginString() == "GPU":
logger.error("Stitching critical failed : " + msg)
CLIENT_MESSENGER.send_error(errors.StitcherError("Internal GPU error: " +
status.getErrorMessage()))
utils.async.defer(self.server.reset)
return status
# No state change
return None
def _stitch_loop(self):
"""Stitcher runner job (blocking while stitcher is starved)
"""
logger.info("starting loop...")
self.run_loop.wait()
if self.display.is_active():
""" When displaying on HDMI, the stitch loop runs in C++.
This loop checks the status of the native loop on a regular basis
"""
self.native_loop.Start()
old_status = None
while self.run_loop.is_set():
algodata = self.algorithm_manager.get_next_algorithm(self.project_manager.panorama)
self.native_loop.setAlgorithm(algodata[0], algodata[1])
status = self.native_loop.StitchStatus()
if old_status is None or old_status.ok() != status.ok():
old_status = self._check_status(status)
time.sleep(1 / 1000.)
self.native_loop.Stop()
else:
""" When not ouptuting to HDMI, the stitcher_executor fires the stitch command
"""
while self.run_loop.is_set():
status = self.stitcher_executor.submit(self._stitch).result()
self._check_status(status)
time.sleep(1 / 1000.)
logger.info("leaving loop...")
def _stitch(self):
"""Stitch the input frames into the output. No python thread blocking.
"""
algodata = self.algorithm_manager.get_next_algorithm(self.project_manager.panorama)
status = None
try:
status = vs.stitchAndExtractNoGIL(
self.project_manager.controller,
self.stitch_output.object(),
algodata[1],
algodata[0],
True)
except Exception as e:
logger.warning(e)
return status
def reset_panorama(self, sender, updater, output):
# Output is passed so that we keep a reference to the underlying object. Please do not remove this unused parameter.
# Get updater function if it's updater, otherwise - pass panorama
updater_param = updater.getCloneUpdater() if isinstance(updater, vs.PanoramaDefinitionUpdater) else updater
status = self.project_manager.controller.updatePanorama(updater_param)
if not status.ok():
CLIENT_MESSENGER.send_error(errors.StitcherError("Cannot update the panorama. Reason: " +
status.getErrorMessage()))
return
self.project_manager.update("pano", self.project_manager.get_save_panorama().serialize())
def terminate(self, sender=None):
if self.run_loop and self.run_loop.is_set():
self.run_loop.clear()
logger.info("Joining run loop...")
self.thread.join()
logger.info("Done")
if self.algorithm_manager is not None:
self.algorithm_manager.cancel_running_algorithms()
self.t_terminate()