Source code for fury.stream.client

from functools import partial
import logging
import platform
import time

import numpy as np
import vtk

from fury.decorators import warn_on_args_to_kwargs
from fury.stream.constants import PY_VERSION_8, _CQUEUE
from fury.stream.tools import (
    ArrayCircularQueue,
    IntervalTimer,
    IntervalTimerThreading,
    RawArrayImageBufferManager,
    SharedMemCircularQueue,
    SharedMemImageBufferManager,
)



[docs]
def callback_stream_client(stream_client):
    """This callback is used to update the image inside of
    the ImageManager instance

    Parameters
    ----------
    stream_client : StreamClient

    """
    if stream_client.in_request:
        return
    stream_client.in_request = True
    stream_client.window2image_filter.Update()
    stream_client.window2image_filter.Modified()
    vtk_image = stream_client.window2image_filter.GetOutput()
    vtk_array = vtk_image.GetPointData().GetScalars()

    w, h, _ = vtk_image.GetDimensions()
    np_arr = np.frombuffer(vtk_array, dtype="uint8")
    if np_arr is None:
        stream_client.in_request = False
        return

    stream_client.img_manager.write_into(w, h, np_arr)
    stream_client.in_request = False




[docs]
class FuryStreamClient:
    """This obj is responsible to create a StreamClient."""

    @warn_on_args_to_kwargs()
    def __init__(
        self,
        showm,
        *,
        max_window_size=None,
        use_raw_array=True,
        whithout_iren_start=False,
        num_buffers=2,
    ):
        """A StreamClient extracts a framebuffer from the OpenGL context
        and writes into a shared memory resource.

        Parameters
        ----------
        showm : ShowManager
        max_window_size : tuple of ints, optional
            This allows resize events inside of the FURY window instance.
            Should be greater than the window size.
        use_raw_array : bool, optional
            If False then FuryStreamClient will use SharedMemory
            instead of RawArrays. Notice that Python >=3.8 it's a
            requirement
            to use SharedMemory)
        whithout_iren_start : bool, optional
            Sometimes you can't initiate the vtkInteractor instance.
        num_buffers : int, optional
            Number of buffers to be used in the n-buffering
            technique.

        """
        self._whithout_iren_start = whithout_iren_start
        self.showm = showm
        self.window2image_filter = vtk.vtkWindowToImageFilter()
        self.window2image_filter.SetInput(self.showm.window)
        self.image_buffers = []
        self.image_buffer_names = []
        self.info_buffer_name = None
        self.image_reprs = []
        self.num_buffers = num_buffers
        if max_window_size is None:
            max_window_size = (
                int(self.showm.size[0] * (1 + 0.1)),
                int(self.showm.size[1] * (1 + 0.1)),
            )

        self.max_size = max_window_size[0] * max_window_size[1]
        self.max_window_size = max_window_size
        if self.max_size < self.showm.size[0] * self.showm.size[1]:
            raise ValueError("max_window_size must be greater than window_size")

        if not PY_VERSION_8 and not use_raw_array:
            raise ValueError(
                """
               SharedMemory works only in python 3.8 or higher"""
            )

        if use_raw_array:
            self.img_manager = RawArrayImageBufferManager(
                max_window_size=max_window_size, num_buffers=num_buffers
            )
        else:
            self.img_manager = SharedMemImageBufferManager(
                max_window_size=max_window_size, num_buffers=num_buffers
            )

        self._id_timer = None
        self._id_observer = None
        self._interval_timer = None
        self.in_request = False
        self.update = True
        self.use_raw_array = use_raw_array
        self._started = False


[docs]
    @warn_on_args_to_kwargs()
    def start(self, *, ms=0, use_asyncio=False):
        """Start the stream client.

        Parameters
        ----------
        ms : float, optional
            positive number. This update the image buffer using a interval
            of ms milliseconds. If ms is 0 then the stream client
            will update the buffer after every Render event.
        use_asyncio : bool, optional
            If False then the stream client will update the image
            using a threading technique.

        """

        @warn_on_args_to_kwargs()
        def callback_for_vtk(caller, event, *args, **kwargs):
            callback_stream_client(**{"stream_client": kwargs["stream_client"]})

        use_asyncio = platform.system() == "Windows" or use_asyncio
        if self._started:
            self.stop()
        if ms > 0:
            if self._whithout_iren_start:
                Interval = IntervalTimer if use_asyncio else IntervalTimerThreading
                self._interval_timer = Interval(
                    ms / 1000, callback_stream_client, **{"stream_client": self}
                )
            else:
                self._id_observer = self.showm.iren.AddObserver(
                    "TimerEvent", partial(callback_for_vtk, **{"stream_client": self})
                )
                self._id_timer = self.showm.iren.CreateRepeatingTimer(ms)
        else:
            self._id_observer = self.showm.iren.AddObserver(
                "RenderEvent", partial(callback_for_vtk, **{"stream_client": self})
            )
        self.showm.window.Render()
        self.showm.iren.Render()
        self._started = True
        callback_stream_client(**{"stream_client": self})



[docs]
    def stop(self):
        """Stop the stream client."""
        if not self._started:
            return False

        if self._interval_timer is not None:
            self._interval_timer.stop()
        if self._id_timer is not None:
            # self.showm.destroy_timer(self._id_timer)
            self.showm.iren.DestroyTimer(self._id_timer)
            self._id_timer = None

        if self._id_observer is not None:
            self.showm.iren.RemoveObserver(self._id_observer)
            self._id_observer = None

        self._started = False



[docs]
    def cleanup(self):
        """Release the shared memory resources if necessary."""
        if self.use_raw_array:
            return

        self.img_manager.info_buffer.close()
        # this it's due the python core issues
        # https://bugs.python.org/issue38119
        # https://bugs.python.org/issue39959
        # https://github.com/luizalabs/shared-memory-dict/issues/13
        try:
            self.img_manager.info_buffer.unlink()
        except FileNotFoundError:
            print(
                f"Shared Memory {self.img_manager.info_buffer_name}\
                    (info_buffer) File not found"
            )
        for buffer, name in zip(
            self.img_manager.image_buffers,
            self.img_manager.image_buffer_names,
        ):
            buffer.close()
            try:
                buffer.unlink()
            except FileNotFoundError:
                print(f"Shared Memory {name}(buffer image) File not found")





[docs]
def interaction_callback(circular_queue, showm, iren, render_after):
    """This callback is used to invoke vtk interaction events
    reading those events from the provided circular_queue instance

    Parameters
    ----------
    circular_queue : CircularQueue
    showm : ShowmManager
    iren : vtkInteractor
    render_after : bool, optional
        If the render method should be called after an
        dequeue

    """
    ts = time.time() * 1000
    data = circular_queue.dequeue()
    if data is None:
        return

    user_event_id = data[0]
    user_timestamp = data[_CQUEUE.index_info.user_timestamp]

    ts = time.time() * 1000
    newX = int(showm.size[0] * data[_CQUEUE.index_info.x])
    newY = int(showm.size[1] * data[_CQUEUE.index_info.y])
    ctrl_key = int(data[_CQUEUE.index_info.ctrl])
    shift_key = int(data[_CQUEUE.index_info.shift])
    newY = showm.size[1] - newY
    event_ids = _CQUEUE.event_ids
    if user_event_id == event_ids.mouse_weel:
        zoomFactor = 1.0 - data[_CQUEUE.index_info.weel] / 1000.0
        camera = showm.scene.GetActiveCamera()
        fp = camera.GetFocalPoint()
        pos = camera.GetPosition()
        delta = [fp[i] - pos[i] for i in range(3)]

        pos2 = camera.GetPosition()
        camera.SetFocalPoint([pos2[i] + delta[i] for i in range(3)])
        camera.Zoom(zoomFactor)
    elif user_event_id == event_ids.mouse_move:
        iren.SetEventInformation(newX, newY, ctrl_key, shift_key, chr(0), 0, None)

        iren.MouseMoveEvent()

    elif event_ids.mouse_ids:
        iren.SetEventInformation(newX, newY, ctrl_key, shift_key, chr(0), 0, None)
        mouse_actions = {
            event_ids.left_btn_press: iren.LeftButtonPressEvent,
            event_ids.left_btn_release: iren.LeftButtonReleaseEvent,
            event_ids.middle_btn_press: iren.MiddleButtonPressEvent,
            event_ids.middle_btn_release: iren.MiddleButtonReleaseEvent,
            event_ids.right_btn_press: iren.RightButtonPressEvent,
            event_ids.right_btn_release: iren.RightButtonReleaseEvent,
        }
        mouse_actions[user_event_id]()
    logging.info("Interaction: time to perform event " + f"{ts-user_timestamp:.2f} ms")
    if render_after:
        showm.window.Render()
        showm.iren.Render()




[docs]
class FuryStreamInteraction:
    """This obj. is responsible to manage the user interaction"""

    @warn_on_args_to_kwargs()
    def __init__(
        self, showm, *, max_queue_size=50, use_raw_array=True, whithout_iren_start=False
    ):
        """Initialize the StreamInteraction obj.

        Parameters
        ----------
        showm : ShowmManager
        max_queue_size : int, optional
            maximum number of events to be stored.
        use_raw_array : bool, optional
            If False then a CircularQueue will be created using
            SharedMemory instead of RawArrays. Notice that
            Python >=3.8 it's requirement to use SharedMemory.
        whithout_iren_start : bool, optional
            Set that to True if you can't initiate the vtkInteractor
            instance.

        """
        self.showm = showm
        self.iren = self.showm.iren
        if use_raw_array:
            self.circular_queue = ArrayCircularQueue(
                max_size=max_queue_size, dimension=_CQUEUE.dimension
            )
        else:
            self.circular_queue = SharedMemCircularQueue(
                max_size=max_queue_size, dimension=_CQUEUE.dimension
            )

        self._id_timer = None
        self._id_observer = None
        self._interval_timer = None
        self._whithout_iren_start = whithout_iren_start
        self._started = False


[docs]
    @warn_on_args_to_kwargs()
    def start(self, *, ms=3, use_asyncio=False):
        """Start the stream interaction client.

        Parameters
        ----------
        ms : float, optional
            positive number greater than zero.
        use_asyncio : bool, optional
            If False then the interaction will be performed in a
            separate thread.

        """
        use_asyncio = platform.system() == "Windows" or use_asyncio
        if ms <= 0:
            raise ValueError("ms must be greater than zero")

        if self._started:
            self.stop()
        if self._whithout_iren_start:
            Interval = IntervalTimer if use_asyncio else IntervalTimerThreading
            self._interval_timer = Interval(
                ms / 1000,
                interaction_callback,
                *[self.circular_queue, self.showm, self.iren, True],
            )
        else:

            def callback(caller, event, *args, **kwargs):
                interaction_callback(self.circular_queue, self.showm, self.iren, True)

            self._id_observer = self.showm.iren.AddObserver("TimerEvent", callback)
            self._id_timer = self.showm.iren.CreateRepeatingTimer(ms)

        self._started = True



[docs]
    def stop(self):
        """Stop the stream interaction client."""
        if not self._started:
            return False

        if self._id_timer is not None:
            # self.showm.destroy_timer(self._id_timer)
            self.showm.iren.DestroyTimer(self._id_timer)
            self._id_timer = None

        if self._id_observer is not None:
            self.showm.iren.RemoveObserver(self._id_observer)
            self._id_observer = None

        if self._interval_timer is not None:
            self._interval_timer.stop()
            self._interval_timer = None

        self._started = False



[docs]
    def cleanup(self):
        """Release the shared memory resources if necessary."""
        self.circular_queue.cleanup()