# Copyright (c) 2021 Intel Corporation.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM,OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
"""Accepts the frame and publishes it to the EII JupyterNotebook service which
   processes the frame and publishes it back to the jupyter_connector UDF.
"""

import logging
import eii.msgbus as mb
import cfgmgr.config_manager as cfg
import threading
import cv2
import numpy as np
import time


class Udf:
    """Jupyter Notebook connector UDF object
    """
    def __init__(self):
        """Constructor
        """
        self.log = logging.getLogger('JupyterNotebookConnector')
        self.log.debug(f"In {__name__}...")

        # Initializing Etcd to fetch UDF config
        ctx = cfg.ConfigMgr()
        app_cfg = ctx.get_app_config()
        app_cfg = app_cfg.get_dict()
        if "udfs" in app_cfg:
            for udf in app_cfg["udfs"]:
                if udf["name"] == "jupyter_connector":
                    self.udf_config = udf

        # Initializing msgbus publisher
        self.msgbus_cfg = {
                            "type": "zmq_ipc",
                            "socket_dir": "/EII/sockets"
                          }
        self.msgbus = mb.MsgbusContext(self.msgbus_cfg)
        self.publisher = self.msgbus.new_publisher("jupyter_publisher")

        # Initializing msgbus subscriber
        self.subscriber = self.msgbus.new_subscriber("jupyter_subscriber")

        # Starting thread to send UDF config
        conf_thread = threading.Thread(target=self._send_config)
        conf_thread.start()

    def _send_config(self):
        # Send UDF config to jupyter notebook service
        self.config_publisher = self.msgbus.new_publisher("jupyter_config")
        while True:
            self.config_publisher.publish(self.udf_config)
            time.sleep(3)

    def decode_frame(self, results, blob):
        """Identify the defects on the frames

        :param results: Metadata of frame received from message bus.
        :type: dict
        :param blob: Actual frame received from message bus.
        :type: bytes
        :return: Return decoded frame
        :rtype: numpy array
        """
        height = int(results['height'])
        width = int(results['width'])
        channels = int(results['channels'])
        encoding = None

        if 'encoding_type' and 'encoding_level' in results:
            encoding = {"type": results['encoding_type'],
                        "level": results['encoding_level']}
        # Convert to Numpy array and reshape to frame
        if isinstance(blob, list):
            # If multiple frames, select first frame for
            # visualization
            blob = blob[0]
        frame = np.frombuffer(blob, dtype=np.uint8)
        if encoding is not None:
            frame = np.reshape(frame, (frame.shape))
            try:
                frame = cv2.imdecode(frame, 1)
            except cv2.error as ex:
                self.log.error("frame: {}, exception: {}".format(frame, ex))
        else:
            self.log.debug("Encoding not enabled...")
            frame = np.reshape(frame, (height, width, channels))

        return frame

    def process(self, frame, metadata):
        """Publishes every frame received to Jupyter Notebook service
           and returns the respective processed & subscribed frame

        :param frame: frame blob
        :type frame: numpy.ndarray
        :param metadata: frame's metadata
        :type metadata: str
        :return:  (should the frame be dropped, has the frame been updated,
                   new metadata for the frame if any)
        :rtype: (bool, numpy.ndarray, str)
        """
        self.log.debug(f"In process() method...")

        # Publishing metadata & frames
        self.publisher.publish((metadata, frame.tobytes(),))
        metadata, frame = self.subscriber.recv()

        if "jpnb_frame_drop" in metadata:
            del metadata["jpnb_frame_drop"]
            return True, None, None
        elif "jpnb_frame_updated" in metadata and "jpnb_metadata_updated" in metadata:
            del metadata["jpnb_frame_updated"]
            del metadata["jpnb_metadata_updated"]
            return False, frame, metadata
        elif "jpnb_frame_updated" in metadata or "jpnb_metadata_updated" in metadata:
            if "jpnb_frame_updated" in metadata:
                # Decode frame to retain text
                frame = self.decode_frame(metadata, frame)
                del metadata["jpnb_frame_updated"]
                return False, frame, None
            if "jpnb_metadata_updated" in metadata:
                del metadata["jpnb_metadata_updated"]
                return False, None, metadata
        else:
            return False, None, None

138d72222074634f9c6fb2cf534e08cbcf98ab69
<mask token>


class Udf:
    <mask token>

    def __init__(self):
        """Constructor
        """
        self.log = logging.getLogger('JupyterNotebookConnector')
        self.log.debug(f'In {__name__}...')
        ctx = cfg.ConfigMgr()
        app_cfg = ctx.get_app_config()
        app_cfg = app_cfg.get_dict()
        if 'udfs' in app_cfg:
            for udf in app_cfg['udfs']:
                if udf['name'] == 'jupyter_connector':
                    self.udf_config = udf
        self.msgbus_cfg = {'type': 'zmq_ipc', 'socket_dir': '/EII/sockets'}
        self.msgbus = mb.MsgbusContext(self.msgbus_cfg)
        self.publisher = self.msgbus.new_publisher('jupyter_publisher')
        self.subscriber = self.msgbus.new_subscriber('jupyter_subscriber')
        conf_thread = threading.Thread(target=self._send_config)
        conf_thread.start()
    <mask token>

    def decode_frame(self, results, blob):
        """Identify the defects on the frames

        :param results: Metadata of frame received from message bus.
        :type: dict
        :param blob: Actual frame received from message bus.
        :type: bytes
        :return: Return decoded frame
        :rtype: numpy array
        """
        height = int(results['height'])
        width = int(results['width'])
        channels = int(results['channels'])
        encoding = None
        if 'encoding_type' and 'encoding_level' in results:
            encoding = {'type': results['encoding_type'], 'level': results[
                'encoding_level']}
        if isinstance(blob, list):
            blob = blob[0]
        frame = np.frombuffer(blob, dtype=np.uint8)
        if encoding is not None:
            frame = np.reshape(frame, frame.shape)
            try:
                frame = cv2.imdecode(frame, 1)
            except cv2.error as ex:
                self.log.error('frame: {}, exception: {}'.format(frame, ex))
        else:
            self.log.debug('Encoding not enabled...')
            frame = np.reshape(frame, (height, width, channels))
        return frame
    <mask token>

<mask token>


class Udf:
    <mask token>

    def __init__(self):
        """Constructor
        """
        self.log = logging.getLogger('JupyterNotebookConnector')
        self.log.debug(f'In {__name__}...')
        ctx = cfg.ConfigMgr()
        app_cfg = ctx.get_app_config()
        app_cfg = app_cfg.get_dict()
        if 'udfs' in app_cfg:
            for udf in app_cfg['udfs']:
                if udf['name'] == 'jupyter_connector':
                    self.udf_config = udf
        self.msgbus_cfg = {'type': 'zmq_ipc', 'socket_dir': '/EII/sockets'}
        self.msgbus = mb.MsgbusContext(self.msgbus_cfg)
        self.publisher = self.msgbus.new_publisher('jupyter_publisher')
        self.subscriber = self.msgbus.new_subscriber('jupyter_subscriber')
        conf_thread = threading.Thread(target=self._send_config)
        conf_thread.start()

    def _send_config(self):
        self.config_publisher = self.msgbus.new_publisher('jupyter_config')
        while True:
            self.config_publisher.publish(self.udf_config)
            time.sleep(3)

    def decode_frame(self, results, blob):
        """Identify the defects on the frames

        :param results: Metadata of frame received from message bus.
        :type: dict
        :param blob: Actual frame received from message bus.
        :type: bytes
        :return: Return decoded frame
        :rtype: numpy array
        """
        height = int(results['height'])
        width = int(results['width'])
        channels = int(results['channels'])
        encoding = None
        if 'encoding_type' and 'encoding_level' in results:
            encoding = {'type': results['encoding_type'], 'level': results[
                'encoding_level']}
        if isinstance(blob, list):
            blob = blob[0]
        frame = np.frombuffer(blob, dtype=np.uint8)
        if encoding is not None:
            frame = np.reshape(frame, frame.shape)
            try:
                frame = cv2.imdecode(frame, 1)
            except cv2.error as ex:
                self.log.error('frame: {}, exception: {}'.format(frame, ex))
        else:
            self.log.debug('Encoding not enabled...')
            frame = np.reshape(frame, (height, width, channels))
        return frame

    def process(self, frame, metadata):
        """Publishes every frame received to Jupyter Notebook service
           and returns the respective processed & subscribed frame

        :param frame: frame blob
        :type frame: numpy.ndarray
        :param metadata: frame's metadata
        :type metadata: str
        :return:  (should the frame be dropped, has the frame been updated,
                   new metadata for the frame if any)
        :rtype: (bool, numpy.ndarray, str)
        """
        self.log.debug(f'In process() method...')
        self.publisher.publish((metadata, frame.tobytes()))
        metadata, frame = self.subscriber.recv()
        if 'jpnb_frame_drop' in metadata:
            del metadata['jpnb_frame_drop']
            return True, None, None
        elif 'jpnb_frame_updated' in metadata and 'jpnb_metadata_updated' in metadata:
            del metadata['jpnb_frame_updated']
            del metadata['jpnb_metadata_updated']
            return False, frame, metadata
        elif 'jpnb_frame_updated' in metadata or 'jpnb_metadata_updated' in metadata:
            if 'jpnb_frame_updated' in metadata:
                frame = self.decode_frame(metadata, frame)
                del metadata['jpnb_frame_updated']
                return False, frame, None
            if 'jpnb_metadata_updated' in metadata:
                del metadata['jpnb_metadata_updated']
                return False, None, metadata
        else:
            return False, None, None

<mask token>


class Udf:
    """Jupyter Notebook connector UDF object
    """

    def __init__(self):
        """Constructor
        """
        self.log = logging.getLogger('JupyterNotebookConnector')
        self.log.debug(f'In {__name__}...')
        ctx = cfg.ConfigMgr()
        app_cfg = ctx.get_app_config()
        app_cfg = app_cfg.get_dict()
        if 'udfs' in app_cfg:
            for udf in app_cfg['udfs']:
                if udf['name'] == 'jupyter_connector':
                    self.udf_config = udf
        self.msgbus_cfg = {'type': 'zmq_ipc', 'socket_dir': '/EII/sockets'}
        self.msgbus = mb.MsgbusContext(self.msgbus_cfg)
        self.publisher = self.msgbus.new_publisher('jupyter_publisher')
        self.subscriber = self.msgbus.new_subscriber('jupyter_subscriber')
        conf_thread = threading.Thread(target=self._send_config)
        conf_thread.start()

    def _send_config(self):
        self.config_publisher = self.msgbus.new_publisher('jupyter_config')
        while True:
            self.config_publisher.publish(self.udf_config)
            time.sleep(3)

    def decode_frame(self, results, blob):
        """Identify the defects on the frames

        :param results: Metadata of frame received from message bus.
        :type: dict
        :param blob: Actual frame received from message bus.
        :type: bytes
        :return: Return decoded frame
        :rtype: numpy array
        """
        height = int(results['height'])
        width = int(results['width'])
        channels = int(results['channels'])
        encoding = None
        if 'encoding_type' and 'encoding_level' in results:
            encoding = {'type': results['encoding_type'], 'level': results[
                'encoding_level']}
        if isinstance(blob, list):
            blob = blob[0]
        frame = np.frombuffer(blob, dtype=np.uint8)
        if encoding is not None:
            frame = np.reshape(frame, frame.shape)
            try:
                frame = cv2.imdecode(frame, 1)
            except cv2.error as ex:
                self.log.error('frame: {}, exception: {}'.format(frame, ex))
        else:
            self.log.debug('Encoding not enabled...')
            frame = np.reshape(frame, (height, width, channels))
        return frame

    def process(self, frame, metadata):
        """Publishes every frame received to Jupyter Notebook service
           and returns the respective processed & subscribed frame

        :param frame: frame blob
        :type frame: numpy.ndarray
        :param metadata: frame's metadata
        :type metadata: str
        :return:  (should the frame be dropped, has the frame been updated,
                   new metadata for the frame if any)
        :rtype: (bool, numpy.ndarray, str)
        """
        self.log.debug(f'In process() method...')
        self.publisher.publish((metadata, frame.tobytes()))
        metadata, frame = self.subscriber.recv()
        if 'jpnb_frame_drop' in metadata:
            del metadata['jpnb_frame_drop']
            return True, None, None
        elif 'jpnb_frame_updated' in metadata and 'jpnb_metadata_updated' in metadata:
            del metadata['jpnb_frame_updated']
            del metadata['jpnb_metadata_updated']
            return False, frame, metadata
        elif 'jpnb_frame_updated' in metadata or 'jpnb_metadata_updated' in metadata:
            if 'jpnb_frame_updated' in metadata:
                frame = self.decode_frame(metadata, frame)
                del metadata['jpnb_frame_updated']
                return False, frame, None
            if 'jpnb_metadata_updated' in metadata:
                del metadata['jpnb_metadata_updated']
                return False, None, metadata
        else:
            return False, None, None

<mask token>
import logging
import eii.msgbus as mb
import cfgmgr.config_manager as cfg
import threading
import cv2
import numpy as np
import time


class Udf:
    """Jupyter Notebook connector UDF object
    """

    def __init__(self):
        """Constructor
        """
        self.log = logging.getLogger('JupyterNotebookConnector')
        self.log.debug(f'In {__name__}...')
        ctx = cfg.ConfigMgr()
        app_cfg = ctx.get_app_config()
        app_cfg = app_cfg.get_dict()
        if 'udfs' in app_cfg:
            for udf in app_cfg['udfs']:
                if udf['name'] == 'jupyter_connector':
                    self.udf_config = udf
        self.msgbus_cfg = {'type': 'zmq_ipc', 'socket_dir': '/EII/sockets'}
        self.msgbus = mb.MsgbusContext(self.msgbus_cfg)
        self.publisher = self.msgbus.new_publisher('jupyter_publisher')
        self.subscriber = self.msgbus.new_subscriber('jupyter_subscriber')
        conf_thread = threading.Thread(target=self._send_config)
        conf_thread.start()

    def _send_config(self):
        self.config_publisher = self.msgbus.new_publisher('jupyter_config')
        while True:
            self.config_publisher.publish(self.udf_config)
            time.sleep(3)

    def decode_frame(self, results, blob):
        """Identify the defects on the frames

        :param results: Metadata of frame received from message bus.
        :type: dict
        :param blob: Actual frame received from message bus.
        :type: bytes
        :return: Return decoded frame
        :rtype: numpy array
        """
        height = int(results['height'])
        width = int(results['width'])
        channels = int(results['channels'])
        encoding = None
        if 'encoding_type' and 'encoding_level' in results:
            encoding = {'type': results['encoding_type'], 'level': results[
                'encoding_level']}
        if isinstance(blob, list):
            blob = blob[0]
        frame = np.frombuffer(blob, dtype=np.uint8)
        if encoding is not None:
            frame = np.reshape(frame, frame.shape)
            try:
                frame = cv2.imdecode(frame, 1)
            except cv2.error as ex:
                self.log.error('frame: {}, exception: {}'.format(frame, ex))
        else:
            self.log.debug('Encoding not enabled...')
            frame = np.reshape(frame, (height, width, channels))
        return frame

    def process(self, frame, metadata):
        """Publishes every frame received to Jupyter Notebook service
           and returns the respective processed & subscribed frame

        :param frame: frame blob
        :type frame: numpy.ndarray
        :param metadata: frame's metadata
        :type metadata: str
        :return:  (should the frame be dropped, has the frame been updated,
                   new metadata for the frame if any)
        :rtype: (bool, numpy.ndarray, str)
        """
        self.log.debug(f'In process() method...')
        self.publisher.publish((metadata, frame.tobytes()))
        metadata, frame = self.subscriber.recv()
        if 'jpnb_frame_drop' in metadata:
            del metadata['jpnb_frame_drop']
            return True, None, None
        elif 'jpnb_frame_updated' in metadata and 'jpnb_metadata_updated' in metadata:
            del metadata['jpnb_frame_updated']
            del metadata['jpnb_metadata_updated']
            return False, frame, metadata
        elif 'jpnb_frame_updated' in metadata or 'jpnb_metadata_updated' in metadata:
            if 'jpnb_frame_updated' in metadata:
                frame = self.decode_frame(metadata, frame)
                del metadata['jpnb_frame_updated']
                return False, frame, None
            if 'jpnb_metadata_updated' in metadata:
                del metadata['jpnb_metadata_updated']
                return False, None, metadata
        else:
            return False, None, None

# Copyright (c) 2021 Intel Corporation.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM,OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
"""Accepts the frame and publishes it to the EII JupyterNotebook service which
   processes the frame and publishes it back to the jupyter_connector UDF.
"""

import logging
import eii.msgbus as mb
import cfgmgr.config_manager as cfg
import threading
import cv2
import numpy as np
import time


class Udf:
    """Jupyter Notebook connector UDF object
    """
    def __init__(self):
        """Constructor
        """
        self.log = logging.getLogger('JupyterNotebookConnector')
        self.log.debug(f"In {__name__}...")

        # Initializing Etcd to fetch UDF config
        ctx = cfg.ConfigMgr()
        app_cfg = ctx.get_app_config()
        app_cfg = app_cfg.get_dict()
        if "udfs" in app_cfg:
            for udf in app_cfg["udfs"]:
                if udf["name"] == "jupyter_connector":
                    self.udf_config = udf

        # Initializing msgbus publisher
        self.msgbus_cfg = {
                            "type": "zmq_ipc",
                            "socket_dir": "/EII/sockets"
                          }
        self.msgbus = mb.MsgbusContext(self.msgbus_cfg)
        self.publisher = self.msgbus.new_publisher("jupyter_publisher")

        # Initializing msgbus subscriber
        self.subscriber = self.msgbus.new_subscriber("jupyter_subscriber")

        # Starting thread to send UDF config
        conf_thread = threading.Thread(target=self._send_config)
        conf_thread.start()

    def _send_config(self):
        # Send UDF config to jupyter notebook service
        self.config_publisher = self.msgbus.new_publisher("jupyter_config")
        while True:
            self.config_publisher.publish(self.udf_config)
            time.sleep(3)

    def decode_frame(self, results, blob):
        """Identify the defects on the frames

        :param results: Metadata of frame received from message bus.
        :type: dict
        :param blob: Actual frame received from message bus.
        :type: bytes
        :return: Return decoded frame
        :rtype: numpy array
        """
        height = int(results['height'])
        width = int(results['width'])
        channels = int(results['channels'])
        encoding = None

        if 'encoding_type' and 'encoding_level' in results:
            encoding = {"type": results['encoding_type'],
                        "level": results['encoding_level']}
        # Convert to Numpy array and reshape to frame
        if isinstance(blob, list):
            # If multiple frames, select first frame for
            # visualization
            blob = blob[0]
        frame = np.frombuffer(blob, dtype=np.uint8)
        if encoding is not None:
            frame = np.reshape(frame, (frame.shape))
            try:
                frame = cv2.imdecode(frame, 1)
            except cv2.error as ex:
                self.log.error("frame: {}, exception: {}".format(frame, ex))
        else:
            self.log.debug("Encoding not enabled...")
            frame = np.reshape(frame, (height, width, channels))

        return frame

    def process(self, frame, metadata):
        """Publishes every frame received to Jupyter Notebook service
           and returns the respective processed & subscribed frame

        :param frame: frame blob
        :type frame: numpy.ndarray
        :param metadata: frame's metadata
        :type metadata: str
        :return:  (should the frame be dropped, has the frame been updated,
                   new metadata for the frame if any)
        :rtype: (bool, numpy.ndarray, str)
        """
        self.log.debug(f"In process() method...")

        # Publishing metadata & frames
        self.publisher.publish((metadata, frame.tobytes(),))
        metadata, frame = self.subscriber.recv()

        if "jpnb_frame_drop" in metadata:
            del metadata["jpnb_frame_drop"]
            return True, None, None
        elif "jpnb_frame_updated" in metadata and "jpnb_metadata_updated" in metadata:
            del metadata["jpnb_frame_updated"]
            del metadata["jpnb_metadata_updated"]
            return False, frame, metadata
        elif "jpnb_frame_updated" in metadata or "jpnb_metadata_updated" in metadata:
            if "jpnb_frame_updated" in metadata:
                # Decode frame to retain text
                frame = self.decode_frame(metadata, frame)
                del metadata["jpnb_frame_updated"]
                return False, frame, None
            if "jpnb_metadata_updated" in metadata:
                del metadata["jpnb_metadata_updated"]
                return False, None, metadata
        else:
            return False, None, None
