<jupyter_start><jupyter_text>Streaming Analysis of Video FramesThis is the image processing portion of the [wikiImageAnalysis](wikiImageAnalysis.jupyter-py36.ipynb), deploying only the image collection/analysis portion. The wikiImageAnalysis project walks through the processes of deriving images from the wiki feed, a multistep process. In the instances that you have a supply of images, ,video stream or mpeg files, their is no derivation process. The notebook composes and submits three applications that- Receive image/video frames via Kafka.- Uses a Face Detection model locate faces in the frames. - Uses a Object Detection mode to locates objects (person, car, bicycle, boat...)A rendering accompanies each of the models. The [videoToAnalysis](videoToAnalysis.jupyter-py36.ipynb) notebook publishes video frames to the Kafka to complete the example.   OVERVIEWThe image below show the entire application, encoded images flow into the VideoFrame operator from Kafka, the frame is decode in 'kafka_frame' and published to the other Streams via 'pubImageA..'![graph of application](images/imageOnlyApplications.png) Documentation- [Streams Python development guide](https://ibmstreams.github.io/streamsx.documentation/docs/latest/python/)- [Streams Python API](https://streamsxtopology.readthedocs.io/)- [Topology](https://streamsxtopology.readthedocs.io/en/latest/index.html) Streams Topology documentation- [Widgets](https://ipywidgets.readthedocs.io/en/stable/examples/Widget%20Basics.html) Notebook Widgets documentation.  Setup Setup Imports<jupyter_code># Install components
#!pip --user install SSEClient===0.0.22 --upgrade
#!pip install --user --upgrade streamsx<jupyter_output><empty_output><jupyter_text>Additional installs. - pip install streamsx.eventstreams Initial imports are in support of accessing Streams with Wikipedia data, subsequent are in support of rendering the results as they flow back from Streams.<jupyter_code>import json
import datetime 
import os
import sys
import time
import threading


from  functools import lru_cache
from statistics import mean
import collections
from collections import deque
from collections import Counter

import pandas as pd

%matplotlib inline
import matplotlib.pyplot as plt
import ipywidgets as widgets
from ipywidgets import Button, HBox, VBox, Layout
from IPython.core.debugger import set_trace
from IPython.display import display, clear_output

#from bs4 import BeautifulSoup
#from sseclient import SSEClient as EventSource
import requests

import streamsx
from streamsx.topology.topology import *
import streamsx.rest as rest
from streamsx.topology import context
if '../scripts' not in sys.path:
    sys.path.insert(0, '../scripts')
    
#import ipynb

import streams_aid as aid
import streams_render as render
import credential
import cvsupport
print("streamsx package version: " + streamsx.topology.context.__version__)


import cv2
cv2.__version__<jupyter_output><empty_output><jupyter_text>Helper functions used in interfacing  to Streams  Their are number of helper functions to make it aid in the development of Streams applicatons, refer to scripts/streams_aid.py or utilize.For details on the avaliable aid functions, use the help command.  Setup CredentialsThe service that you use defines what you need to setup.- Describe CP4D- Describe Cloud  Add credentials for the IBM Streams service ICP4D setupWith the cell below selected, click the "Connect to instance" button in the toolbar to insert the credentials for the service.See an example.  Cloud setupTo use Streams instance running in the cloud setup a [credential.py](setup_credential.ipynb)<jupyter_code># If your using the cloud set the name of the Cloud Streams instance you will be using. 
# Credential data 
SERVICE_NAME='Streaming3Turbine'
instance,cfg = aid.get_instance(service_name=SERVICE_NAME)<jupyter_output>Outside ICP4D<jupyter_text>Connect to the server :  ICP4D or Cloud instance -<jupyter_code># The code is setup to run in the Cloud and CP4D. 
# If in the Cloud, you'll see a message.
# If in CP4D, you'll need the service credential, which what the above link explains. 
#    Delete this cell and use the above instructions if you only using CP4D.

try:
    from icpd_core import icpd_util
except ModuleNotFoundError as e:  # get all exceptions
    instance,cfg = aid.get_instance(service_name=SERVICE_NAME)

else:   # runs when no exception occurs
    cfg=icpd_util.get_service_instance_details(name='zen-sample-icp1-blitz-env')<jupyter_output>Outside ICP4D<jupyter_text>ConnectedWhen you're up to this point you have established connection to the Streams instance. The cell below shows, if any, the applications that are running. For the applications that are running thier corresponding rendering codes could be executed without going through the process of submitting the application.<jupyter_code>aid.list_jobs(instance, cancel=True)<jupyter_output><empty_output><jupyter_text>Next compose and submit the first appication or back to the [Directory](DIRECTORY)  Thier may not be any jobs currently running. Continue on to compose an application or [back to Directory](DIRECTORY)   View into the live streamThe [view](https://streamsxtopology.readthedocs.io/en/latest/streamsx.topology.topology.htmlstreamsx.topology.topology.Stream.view) enables access to live stream at runtime. We spread them liberaly throughout the application to observe how the processing is procedeing  Viewing live data:- constructing the view object : https://streamsxtopology.readthedocs.io/en/latest/streamsx.topology.topology.html?highlight=viewstreamsx.topology.topology.Stream.view- methods on the view object : https://streamsxtopology.readthedocs.io/en/latest/streamsx.topology.topology.html?highlight=viewstreamsx.topology.topology.View<jupyter_code># Notebook support

def render_image(image_url=None, output_region=None):
    """Write the image into a output region.
    
    Args::
        url: image
        output_region: output region
        
    .. note:: The creation of the output 'stage', if this is not done the image is rendered in the page and
        the output region. 
        
    """
    
    try:
        response = requests.get(image_url)
        stage = widgets.Output(layout={'border': '1px solid green'})
    except:
        print("Error on request : ", image_url)
    else:
        if response.status_code == 200:
            with output_region:
                stage.append_display_data(widgets.Image(
                    value=response.content,
                    #format='jpg',
                    width=300,
                    height=400,
                ))
            output_region.clear_output(wait=True) 

ana_stage = list()
def display_image(tup, image_region=None, title_region=None, url_region=None):
    if tup['img_desc'] is not None and len(tup['img_desc']) > 0:
        display_desc = tup['img_desc'][0]
        ana_stage.append(display_desc)
        title_region.value = "Img Title:{}".format(display_desc['title'] )
        url_region.value = "{}".format(display_desc['img'])
        render_image(image_url=display_desc['img'], output_region=image_region)<jupyter_output><empty_output><jupyter_text>Recieve Video Frames via Kafka - VideoRcvKafka- subscribe to kafka 'image_active' - publish to 'image_active' topic<jupyter_code>import streamsx.eventstreams as eventstreams
from streamsx.topology.schema import CommonSchema

def VideoRcvKafka():
    """Recieve video frame on topic and publish to 'image_string'. 
    
    Notes:
        - The script VideoSndKafka.py pushed video frame onto the topic.
    
    
    """
    topo = Topology("VideoRcvKafka")
    topo.add_pip_package('opencv-contrib-python')

    video_chunk = eventstreams.subscribe(topo, schema=CommonSchema.Json, 
                                         credentials=json.loads(credential.magsEventStream),
                                         topic='VideoFrame' )
    kafka_frame = video_chunk.map(cvsupport.BuildVideoFrame(), name="kafka_frame")
    kafka_frame.view(name="frame_kafka", description="frame from kafka image")
    kafka_frame.publish(topic="image_active", name="pubImageActive")
    return topo


aid.cloudSubmit(instance, SERVICE_NAME, VideoRcvKafka(), credential)<jupyter_output><empty_output><jupyter_text>[Directory](DIRECTORY)  Compose and submit 'FaceAnalysis' application- subscribe to 'image_active' topic- Find faces in images- Less sophisticated Model![graph of application](images/faceAnalysis.png)<jupyter_code>## compose the findImageElements application
import cvsupport

def FaceAnalysis():
    """Analyse the images with various tools. 
       - Subscribe to 'image_active' with encoded images in 'image_string'
       - process 'image_string' with FaceRegions into 'face_regions' 
       
    """
    topo = Topology("FaceAnalysis")
    topo.add_file_dependency('../datasets/haarcascade_frontalface_default.xml', 'etc')
    topo.add_pip_package('opencv-contrib-python')   
    
    image_active = topo.subscribe(topic="image_active", name="subImageActive")
    
    # Find faces analysis ....
    face_regions = image_active.map(cvsupport.FaceRegions(), name="face_regions")
    face_trimmed = face_regions.map(lambda t: {
        'url':t['img_desc'][0]['img'], 
        'face_regions':t['face_regions'],
        'image_string':t['image_string']
    }, name="faces_trimmed")
    face_trimmed.view(name="faces_view", description="faces regions")
    return topo

aid.cloudSubmit(instance, SERVICE_NAME, FaceAnalysis(), credential)<jupyter_output><empty_output><jupyter_text>[Directory](DIRECTORY)  Compose and submit 'ObjectAnalysis' application- subscribe to 'image_active' topic- Analysis operator that does analysis: findObject.- More sophisticated model- Bigger longer to submit![graph of application](images/objectAnalysis.png)<jupyter_code>def ObjectAnalysis():
    """Analyse the images with various tools. 
       - Subscribe to 'image_active' with encoded images in 'image_string'
       - process 'image_string' with ObjectRegions into 'object_regions'
       - 
       
    """
    topo = Topology("ObjectAnalysis")
    topo.add_file_dependency('../datasets/haarcascade_frontalface_default.xml', 'etc')
    topo.add_file_dependency('../datasets/yolov3.weights', 'etc')
    topo.add_file_dependency('../datasets/yolov3.cfg', 'etc')
    topo.add_pip_package('opencv-contrib-python')    
    image_active = topo.subscribe(topic="image_active", name="subImageActive")
    
    # Find objects analysis ...
    object_regions = image_active.map(cvsupport.ObjectRegions(classes="../datasets/yolov3.txt"), name="object_fetch")
    object_trimmed = object_regions.map(lambda t: {
        'url':t['img_desc'][0]['img'], 
        'object_regions':t['object_regions'],
        'image_string':t['image_string']
    }, name="objects_trimmed")
    object_trimmed.view(name="objects_view", description="object regions")
    return topo


## Compose and submit the findElements Application 


aid.cloudSubmit(instance, SERVICE_NAME, ObjectAnalysis(), credential)<jupyter_output><empty_output><jupyter_text>[Back to Directory](DIRECTORY) View the ImageAnalysis application's FacesRegions results -<jupyter_code># SHOW Found Faces
face_dash = render.faceDashboard(instance, sleep=2)
display(face_dash.dashboard)
face_dash.ignition(True)<jupyter_output><empty_output><jupyter_text>[Back to Directory](DIRECTORY) View the ImageAnalysis application's ObjectRegions resultsDisplay the images and the objects they found. Run in a thread to make is easier to stop.On the Streams side it's processing the images to find objects, if not no objects are found they are not pushed to the view and will not be rendered here.The object detection maybe CPU intensive (you don't say) you may want to distribute this across hardware, whichis left as an exercise to the user.<jupyter_code>object_dash = render.objectDashboard(instance, sleep=.2)
display(object_dash.dashboard)
object_dash.ignition(True)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08650720893407784, 'fraction_numerical': 0.0035836319693307774, 'mean_word_length': 4.599626691553897, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 44, 'https://': 7, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '649274', 'license': 'Apache-2.0', 'n_tokens_mistral': 3487, 'n_tokens_neox': 3364, 'n_words': 1042, 'path': 'jupyter/videoImageAnalysis.jupyter-py36.ipynb', 'repo_name': 'IBMStreams/sample.wikirecent', 'size': '11999'}