gevserver/QmlGevServer

# ---------------------------------------------------------------------------
# The QmlGevServer example starts a QML GevServer which can load a device
# and stream the device image.
# ---------------------------------------------------------------------------
 
import os, sys
 
import cvb
import cvb.gevserver as gs
import cvb.ui
 
from PySide2.QtWidgets import QApplication
from PySide2.QtQuick import QQuickView
from PySide2.QtCore import QObject, Signal, Slot, Property, QUrl
from PySide2.QtGui import QIcon
 
class StreamProcessingHandler(cvb.SingleStreamHandler):
    def __init__(self, stream):
        super().__init__(stream)
 
    def handle_async_wait_result(self, image, status):
        if status is cvb.WaitStatus.Ok:
            print("New image arrived")
            # display image or
            if BackEnd.grab_image or BackEnd.single_shot:
                BackEnd.single_shot = False
                BackEnd.image_controller.refresh(image)
 
                if BackEnd.has_ring_buffer:
                    image.unlock()
            # send image
            elif BackEnd.server_started and (not BackEnd.server_single_shot_send_enabled or BackEnd.server_single_shot_send) and BackEnd.server.stream.is_running:
                try:
                    print('Sending image')
                    BackEnd.server_single_shot_send = False
                    if BackEnd.has_ring_buffer:
                        BackEnd.buffer_index = image.buffer_index
                        BackEnd.img_list[image.buffer_index] = image
 
                        def image_release(img):
                            img.unlock()
                            BackEnd.img_list[BackEnd.buffer_index].unlock()
                            print('Unlocked buffer at index ' + str(img.buffer_index))
 
                        BackEnd.server.stream.send(image, image_release)
                        BackEnd.image_controller.refresh(BackEnd.img_list[BackEnd.buffer_index])
                    else:
                        BackEnd.server.stream.send(image)
                        BackEnd.image_controller.refresh(image)
                except:
                    print('Failed to send image')
            elif BackEnd.has_ring_buffer:
                print("Unlock unused image")
                image.unlock()
 
 
class BackEnd(QObject):
    connection_list_changed = Signal()
    server_started = False
    server_single_shot_send_enabled = False
    server_single_shot_send = False
    single_shot = False
    server = None
    grab_image = False
    has_ring_buffer = False
    img_list = list()
    buffer_index = 0
    # create an image controller to interact with the UI
    image_controller = cvb.ui.ImageController()
 
    def __init__(self):
        QObject.__init__(self)
        self.start_in_progress = False
        self.window_state_reg_node = None
        self.event_cookie = None
 
        # load the device
        self.device = cvb.DeviceFactory.open(
            os.path.join(cvb.install_path(), 'drivers', 'CVMock.vin'),
            cvb.AcquisitionStack.Vin)
        self.stream = self.device.stream()
 
        BackEnd.has_ring_buffer = bool(self.stream.ring_buffer)
        if BackEnd.has_ring_buffer:
            self.stream.ring_buffer.lock_mode = cvb.RingBufferLockMode.On
            BackEnd.img_list = [None] * self.stream.ring_buffer.count
 
        # use a single stream handler to setup an acquisition thread and acquire images
        self.handler = StreamProcessingHandler(self.stream)
        self.view = view
        self.view.rootContext().setContextProperty('mainImage', BackEnd.image_controller)
        # register the device image with UI controller to trigger automatic refreshes
        self.stream.get_snapshot()
        BackEnd.image_controller.refresh(self.device.device_image)
        self.handler.run()
 
        # fill combo box with all available connections
        self.connections = gs.LogicalNetworkInterface.get_all_available()
        self.m_connection_list = list()
        for connection in self.connections:
            self.m_connection_list.append(connection.ip_address() + ' (' + connection.ipv4_mask() + ')')
        self.connection_list_changed.emit()
 
    def __enter__(self):
        return self
 
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.handler.is_active:
            self.handler.finish()
        if BackEnd.server:
            BackEnd.server = None
 
    @Property('QVariantList', notify=connection_list_changed)
    def connection_list(self):
        return self.m_connection_list
 
    @Property('QUrl')
    def driver_path(self):
        return QUrl.fromLocalFile(cvb.expand_path('%CVB%drivers'))
 
    @Property('QUrl')
    def desktop_path(self):
        return QUrl.fromLocalFile(os.path.expanduser("~/Desktop"))
 
    @Slot(bool)
    def switch_grab(self, switched):
        if switched and not self.handler.is_active:
          self.handler.run()
        BackEnd.grab_image = switched
 
    def reset_ui(self):
        print("Resetting ui")
        if BackEnd.grab_image:
            print('Resetting grab switch')
            BackEnd.grab_image = False
            switch = self.view.findChild(QObject, 'switchGrab')
            switch.setProperty('checked', False)
 
    def stop_grab(self):
        print('Stopping grab')
        self.reset_ui()
        if self.stream.is_running:
            self.handler.try_finish()
        print('Grab stopped')
 
    @Slot(str)
    def load_device(self, path):
        print('Loading device from ' + path)
        self.stop_grab()
        try:
            tmp_device = cvb.DeviceFactory.open(path, cvb.AcquisitionStack.Vin)
            self.device = tmp_device
            self.stream = self.device.stream()
            self.handler = StreamProcessingHandler(self.stream)
 
            self.stream.get_snapshot()
            BackEnd.has_ring_buffer = bool(self.stream.ring_buffer)
            if BackEnd.has_ring_buffer:
                self.stream.ring_buffer.lock_mode = cvb.RingBufferLockMode.On
                BackEnd.img_list = [None] * self.stream.ring_buffer.count
            BackEnd.image_controller.refresh(self.device.device_image)
            self.handler.run()
        except:
            print('Failed to load new device')
 
    @Slot(str)
    def save_image(self, path):
        print('Saving image to ' + path)
        self.device.device_image.save(path)
 
    @Slot()
    def btn_single_shot(self):
        print('Single Shot')
        self.reset_ui()
        if not self.handler.is_active:
            self.handler.run()
        BackEnd.single_shot = True
 
    @Slot(bool)
    def switch_enable_single_shot_send(self, enabled):
        BackEnd.server_single_shot_send_enabled = enabled
 
    @Slot()
    def btn_single_shot_send(self):
        if BackEnd.server.stream.is_running:
            print('Send single shot')
            BackEnd.server_single_shot_send = True
        else:
            print('Acquisition not enabled on server')
 
    @Slot(int, int)
    def start_server(self, connection_index, driver_type_index):
        print('connection_index', connection_index, 'driver_type_index', driver_type_index)
        # check for recursions
        if self.start_in_progress:
            return
 
        self.start_in_progress = True
 
        try:
            if BackEnd.server_started is False:
                self.reset_ui()
 
                # create server
                BackEnd.server = gs.Server.create_with_const_size(self.device.device_image.size,
                                                                  self.device.device_image.color_model,
                                                                  self.device.device_image.planes[0].data_type,
                                                                  driver_type_index)
 
                self.add_genicam_features()
 
                BackEnd.server.user_version = 'Python GevServer'
                BackEnd.server.node_map.xml_file_version = cvb.GenApiVersion(1, 0, 0)
 
                if(BackEnd.has_ring_buffer):
                    BackEnd.server.stream.resend_buffers_count = 2
                else:
                    BackEnd.server.stream.resend_buffers_count = 0
 
                # start server
                BackEnd.server.start(self.connections[connection_index].ip_address())
 
                # start acquisition
                BackEnd.server_started = True
                print('Started server')
            else:
                BackEnd.server.stop()
                BackEnd.server_started = False
                print('Server stopped')
 
        except:
            print('Failed to start/stop the server')
        self.start_in_progress = False
 
    def on_window_size_changed(self, valueNode):
        print('Window size change callback called')
        val = self.window_state_reg_node.value
        if val is 0:
            self.view.showMinimized()
        elif val is 1:
            self.view.showNormal()
        elif val is 2:
            view.showMaximized()
 
 
    def add_genicam_features(self):
        cat_node = gs.CategoryNode.create('Cust::CustomFeatures')
        BackEnd.server.node_map.add_node(cat_node)
        cat_node.display_name = 'Custom Features'
        cat_node.tool_tip = 'Contains all application defined features'
        root_node = BackEnd.server.node_map['Root']
        root_node.add(cat_node, gs.NodeList.Child)
 
        self.window_state_reg_node = gs.Int32RegNode.create('Cust::WindowStateReg')
        BackEnd.server.node_map.add_node(self.window_state_reg_node)
        self.window_state_reg_node.visibility = cvb.Visibility.Invisible
        # no cache because of polling
        self.window_state_reg_node.cache_mode = cvb.CacheMode.NoCache
        # polling time in ms
        self.window_state_reg_node.polling_time = 333
        self.window_state_reg_node.value = 1 # view in normal mode
        self.event_cookie = self.window_state_reg_node.register_event_written_updated(self.on_window_size_changed)
 
        enumeration_node = gs.EnumerationNode.create('Cust::WindowState')
        BackEnd.server.node_map.add_node(enumeration_node)
        enumeration_node.display_name = 'Window State'
        enumeration_node.tool_tip = 'Current window state of server application'
        enumeration_node.value_config_as_node = self.window_state_reg_node
 
        minimized_node = gs.EnumEntryNode.create('Cust::Minimized')
        minimized_node.numeric_value = 0
        enumeration_node.add(minimized_node, gs.NodeList.Child)
 
        normal_node = gs.EnumEntryNode.create('Cust::Normal')
        normal_node.numeric_value = 1 # must be ascending entry
        enumeration_node.add(normal_node, gs.NodeList.Child)
 
        max_node = gs.EnumEntryNode.create('Cust::Maximized')
        max_node.numeric_value = 2
        enumeration_node.add(max_node, gs.NodeList.Child)
 
        cat_node.add(enumeration_node, gs.NodeList.Child)
 
 
if __name__ == '__main__':
    sys.argv += ['--style', 'Windows']
    app = QApplication(sys.argv)
    app.setOrganizationName('STEMMER IMAGING')
    app.setOrganizationDomain('https://www.stemmer-imaging.com/')
    app.setApplicationName('GevServer Python tutorial')
 
    # tell Windows the correct AppUserModelID for this process (shows icon in the taskbar)
    if sys.platform == 'win32':
      import ctypes
      myappid = u'stemmerimaging.commonvisionblox.pygevserver.0'
      ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(myappid)
 
    app.setWindowIcon(QIcon('Tutorial-Python_32x32.png'))
 
    # register the display component with QML
    cvb.ui.ImageViewItem.register()
 
    # setup the QML UI
    view = QQuickView()
    view.setResizeMode(QQuickView.SizeRootObjectToView)
 
    # create backend object
    with BackEnd() as backEnd:
 
        # register it to the context of QML
        view.rootContext().setContextProperty('backEnd', backEnd)
        # load the qml file into the engine
        view.setSource(QUrl('main.qml'))
        # call single shot to update the UI
        backEnd.btn_single_shot()
 
        view.resize(640, 390)
        view.show()
 
        # start the UI event handler
        app.exec_()
 

import QtQuick 2.0
import CvbQuick 1.0 as CvbQuick
import QtQuick.Controls 2.12
import QtQuick.Layouts 1.12
import QtQuick.Dialogs 1.3
import QtQuick.Controls.Universal 2.4
import QtQuick.Window 2.3
 
 
GroupBox {
    id: gridBox
    width: 640
    height: 390
    Layout.fillWidth: true
    Layout.fillHeight: true
 
    property var loadFile: true
    property var enableSingleShotSend: false
 
    Universal.theme: Universal.System
 
    background: Rectangle {
            width: parent.width
            height: parent.height
            color: Universal.background
        }
 
    FileDialog {
        id: fileDialog
        modality: Qt.NonModal
        title: loadFile ? "Select an image or driver to load" : "Save image"
        selectExisting: loadFile
        selectFolder: false
        nameFilters: ["Drivers (*.vin *.emu)", "Image files (*png *.bmp)", "All files (*)"]
        selectedNameFilter: loadFile ? "Drivers" : "Image files"
        sidebarVisible: true
        onAccepted: {
        var path = fileUrl.toString();
        // remove prefixed "file:///"
        path = path.replace(/^(file:\/{3})/,"");
        // unescape html codes like '%23' for '#'
        var cleanPath = decodeURIComponent(path);
 
        loadFile ? backEnd.load_device(path) : backEnd.save_image(path)
        }
    }
 
    GridLayout {
        id: gridLayout
        rows: 4
        flow: GridLayout.TopToBottom
        anchors.fill: parent
 
        ComboBox
        {
            id: connectionCombo
            objectName: "connectionCombo"
            enabled: btnStartServer.m_server_disabled
            Layout.fillWidth: true
            Layout.columnSpan: 2
            model: backEnd.connection_list
        }
 
    Rectangle {
        id: page
        Layout.rowSpan: 3
        Layout.columnSpan: 3
        Layout.fillHeight: true
        Layout.fillWidth: true
        CvbQuick.ImageView
           {
               id: view
               image: mainImage
               anchors.fill: parent
           }
           }
 
        ComboBox
        {
            id: driverTypeCombo
            enabled: btnStartServer.m_server_disabled
            Layout.fillWidth: true
            model:
            (Qt.platform.os == "windows") ?
            [
                qsTr("Socket Driver (loopback)"),
                qsTr("Filter Driver (default)")
            ]
            :
            [
                qsTr("Socket Driver")
            ]
        }
 
        Button
        {
            objectName: "btnStartServer"
            id: btnStartServer
            implicitWidth: btnLoadDevice.width
            property var m_server_disabled: true
            text: m_server_disabled ? qsTr("Start Server") : qsTr("Stop Server")
             onClicked: {
                 m_server_disabled = m_server_disabled ? false : true
                 backEnd.start_server(connectionCombo.currentIndex, driverTypeCombo.currentIndex)
             }
        }
 
        Column
        {
        spacing: 5
 
        Button
        {
            objectName: "btnLoadDevice"
            id: btnLoadDevice
            enabled: btnStartServer.m_server_disabled
            text: qsTr("Load Device")
            onClicked:
            {
                loadFile = true
                fileDialog.folder = backEnd.driver_path
                fileDialog.open()
            }
 
        }
 
        Button
        {
            objectName: "btnSaveImg"
            id: btnSaveImg
            implicitWidth: btnLoadDevice.width
            text: qsTr("Save Image")
            onClicked:
            {
                loadFile = false
                fileDialog.folder = backEnd.desktop_path
                fileDialog.open()
            }
        }
 
        Button
        {
            objectName: "btnSingleShot"
            id: btnSingleShot
            implicitWidth: btnLoadDevice.width
            width: btnSaveImg.width
            enabled: btnStartServer.m_server_disabled
            text: qsTr("Single Shot")
            onClicked: backEnd.btn_single_shot()
        }
 
        Switch
        {
            objectName: "switchEnableSingleShotSend"
            id: switchEnableSingleShotSend
            text: qsTr("Enable Single Shot Send")
            checked: false
            enabled: !btnStartServer.m_server_disabled
            onCheckedChanged:
            {
                enableSingleShotSend = enableSingleShotSend ? false : true
                backEnd.switch_enable_single_shot_send(checked)
            }
        }
 
        Button
        {
            objectName: "btnSingleShotSend"
            id: btnSingleShotSend
            implicitWidth: btnLoadDevice.width
            enabled: !btnStartServer.m_server_disabled && enableSingleShotSend
            text: qsTr("Single Shot Send")
            onClicked: backEnd.btn_single_shot_send()
        }
        }
 
        Switch
        {
            objectName: "switchGrab"
            id: switchGrab
            text: qsTr("Grab")
            checked: false
            enabled: btnStartServer.m_server_disabled
            onCheckedChanged: backEnd.switch_grab(checked)
        }
    }
}