<jupyter_start><jupyter_text>build-metro-applicationBuilds metro appliction that accepts Kafka messages from the edge(s). Aggregates/Analyze messages - push to storage- generate events - send notifcations- deep analysis- host views to monitor the 'goings on'.  Install and import required libraries<jupyter_code>!pip install streamsx.eventstreams
import urllib3, time, json, os, sys, collections, warnings, ast
import numpy as np

from streamsx.topology.topology import Topology
from streamsx.topology.schema import CommonSchema
from streamsx.topology.context import submit, ContextTypes
import streamsx.eventstreams as eventstreams
from streamsx.rest_primitives import Instance
from streamsx import rest
from streamsx.topology import context

if 'scripts' not in sys.path:
    sys.path.insert(0, 'scripts')
import digit_support<jupyter_output><empty_output><jupyter_text>Add credentials for the IBM Streams service<jupyter_code>urllib3.disable_warnings()
# Cell to grab Streams instance config object and REST reference
from icpd_core import icpd_util
STREAMS_INSTANCE_NAME = "<INSERT-STREAMS-INSTANCE-NAME-HERE>"
streams_cfg=icpd_util.get_service_instance_details(name=STREAMS_INSTANCE_NAME)
streams_cfg[context.ConfigParams.SSL_VERIFY] = False
streams_instance = Instance.of_service(streams_cfg)<jupyter_output><empty_output><jupyter_text>Define Kafka topic<jupyter_code>## Define the topic to access 
## Make sure the topic matches the name specified in the Streams Flow app
ALL_METRICS_TOPIC = '<TOPIC-FOR-ALL-METRICS>'<jupyter_output><empty_output><jupyter_text>Add IBM Event Streams credentials  If you are using Kafka instead, you can skip to the "Create the application" cell<jupyter_code>import getpass<jupyter_output><empty_output><jupyter_text>To add the credentials, go to your Event Streams service on IBM cloud, and under the "Service Credentials" tab, click on the "Copy to clipboard" button, run the following cell and paste the credentials into the textbook that appears underneath<jupyter_code>## Eventstreams for all metrics
metrics_eventstreams_credentials_json = getpass.getpass('Your Event Streams credentials:')
metrics_eventstreams_credentials = eventstreams.configure_connection(streams_instance, name='METRICS', credentials=metrics_eventstreams_credentials_json)
metrics_eventstreams_credentials = 'METRICS'<jupyter_output><empty_output><jupyter_text>Create the application<jupyter_code>class SlideWindow(object):
    """ Window with slide_length elements. 
    
    Window fills, intial output will have less than slide_length.
    
    Args:
        slide_length: maximum number of elements in window.
        
    Returns:
        list of up to 25 of the last tups input.
    """
    def __init__(self, slide_length:int=25):
        self.slide_length = slide_length

    def __enter__(self):
        self.chunk = collections.deque(maxlen=self.slide_length)
        
    def __exit__(self, exc_type, exc_value, traceback):
        # __enter__ and __exit__ must both be defined.
        pass
    
    def __call__(self, tup) -> list:
        self.chunk.append(tup)
        return list(self.chunk)
class get_metrics_tuples(object):
    
    def __call__(self, tup):
        # Unbox the stream from streams flow
        if 'metrics_json' in tup:
            string_tup = tup['metrics_json']
            
            return str(string_tup)
class get_uncertain_metrics_tuples(object):
    
    def __call__(self, tup):
        # Unbox the stream from streams flow
        if 'uncertain_metrics_json' in tup:
            string_tup = tup['uncertain_metrics_json']            
            string_tup = json.loads(string_tup)
        
            # Convert from streams flow types into python types 
            string_tup['result_class'] = int(string_tup['result_class']) # float -> int
            string_tup['predictions'] = ast.literal_eval(string_tup['predictions']) # String -> List
            string_tup['image'] = ast.literal_eval(string_tup['image']) # String -> List
            string_tup['prepared_image'] = np.reshape(string_tup['image'], (28, 28)).tolist() # reshape 784 len List into 28 x 28
    
            return json.dumps(string_tup)<jupyter_output><empty_output><jupyter_text>If you wish to use Kafka instead of IBM Event Streams, uncomment and fill in the following cell<jupyter_code>## To learn more, go to https://streamsxkafka.readthedocs.io/en/latest/ 

# !pip install --user streamsx.kafka>=1.9.0
# import streamsx.kafka as kafka

# consumerProperties = dict()
# consumerProperties['bootstrap.servers'] = ...,
# consumerProperties['fetch.min.bytes'] = ...,
# consumerProperties['max.partition.fetch.bytes'] = ...,
# consumerProperties['security.protocol'] = ...,
# consumerProperties['sasl.jaas.config'] = ...

# # Metrics
# consumer_all_metrics = kafka.KafkaConsumer(config=consumerProperties,
#                          topic=ALL_METRICS_TOPIC,
#                          schema=CommonSchema.Json)<jupyter_output><empty_output><jupyter_text>If you wish to use Kafka instead of IBM Event Streams, uncomment the relevant lines in the following cell<jupyter_code>urllib3.disable_warnings()
def build_metro() -> Topology:
    """ metro application subscribing to two topics 
    
    * Subscribed topics are reflected out view.
    * Metrics are windowed
    
    Returns:
        Topology of the application. 
    """
    topo = Topology('EdgeMetroSubscribe')

    all_metrics = eventstreams.subscribe(topo, schema=CommonSchema.Json, topic=ALL_METRICS_TOPIC, credentials=metrics_eventstreams_credentials)
    ## --- If using kafka, comment out the above line and use the following line instead ---
    # all_metrics = topo.source(consumer_metrics)
    
    # COLLECT METRICS (Includes both certain and uncertain data)
    metrics = all_metrics.map(get_metrics_tuples())
    metrics_view = metrics.view(name="ClassificationMetrics")
    # from_evstr1_clean.print(name="ClassificationMetrics")
    
    # WINDOW METRICS (Includes both certain and uncertain data)
    windowSlide = metrics.map(SlideWindow())
    metrics_windowSlide_view = windowSlide.view(name="WindowClassificationMetrics")
    # windowSlide.print(name="windowPrint")
    
    # COLLECT UNCERTAIN METRICS    
    uncertain_metrics = all_metrics.map(get_uncertain_metrics_tuples())
    uncertrain_metrics_view = uncertain_metrics.view(name="UncertainPredictions")
    # from_evstr2_clean.print(name="uncertainPrint")

    return metrics_view, metrics_windowSlide_view, uncertrain_metrics_view, topo

metrics_view, metrics_windowSlide_view, uncertrain_metrics_view, topo = build_metro()
result = digit_support.submitToStreams(topo, streams_cfg)
result['return_code']
# show the views that we've got up...
time.sleep(15)   # give the applications time to wake up collect some images.....
digit_support.display_views(streams_instance, job_name="EdgeMetroSubscribe")
# Connect to the view and display sample data
queue = uncertrain_metrics_view.start_data_fetch()
try:
    json_val = queue.get()
    val = json.loads(json_val)
    print('Predicted digit is {}'.format(val['result_class']))
    print('Predicted probability is {}\n'.format(val['result_probability']))
    print(val)
finally:
    uncertrain_metrics_view.stop_data_fetch()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08893470790378007, 'fraction_numerical': 0.0035738831615120276, 'mean_word_length': 4.433905899925318, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 41, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '557979', 'license': 'Apache-2.0', 'n_tokens_mistral': 2205, 'n_tokens_neox': 2163, 'n_words': 546, 'path': 'build-metro-app.ipynb', 'repo_name': 'IBMStreams/sample.edge-mnist-flows', 'size': '7275'}