# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3.6
#     language: python
#     name: python3
# ---

# + [markdown] id="e0973593-fcb6-4b3e-bf76-61b78f5994b7"
# <table style="border: none" align="left">
#    <tr style="border: none">
#       <th style="border: none"><font face="verdana" size="5" color="black"><b>Use WML online scoring on CPD to predict Iris species on streaming data</b></th>
# </table>

# + [markdown] id="52c363b0-cf3a-43ff-8921-1a33c9a1590f"
# ![how-it-works.jpg](attachment:how-it-works.jpg)
# This notebook is based on the [WML sample notebook](https://github.com/pmservice/wml-sample-notebooks/blob/master/Use%20PMML%20to%20predict%20Iris%20species.ipynb) which shows storing, deploying and online scoring of the Iris PMML model.
#
# In this notebook, we extend the example to create a Streaming Analytics application that uses the deployment to score the model on streaming data.
#
#
# Some familiarity with Python is helpful. This notebook uses **Python 3.6**.
#
# You will use the **Iris** data set to predict the species of an iris flower. This data set contains measurements of the iris perianth flower. 
#
# ## Learning goals
#
# You will learn how to:
#
# -  Work with the Watson Machine Learning (WML) instance
#   -  Store a PMML model in WML repository
#   -  Deploy the stored model as Online Deployment 
#   -  Score single data with the deployed model to test the deployment
# -  Test the deployed model and its performance in the notebook with the streamsx.wml Python package
# -  Create a Streaming Application with streamsx.wml Python package and deploy it
#
#
# ## Contents
#
# 1.	[Work with the WML service instance](#instance) 
#   1.  [Store the model](#store)
#   2.  [Create a deployment](#deploy)
#   3.  [Single score test](#singlescore)
#   
# 2.  [Test the deployment and its performance for streaming data](#test)
#
# 3.  [Create and run a Streaming Analytics application using this WML online scoring deployment](#application)
#
# 4.	[Summary and next steps](#summary)
#

# + [markdown] id="3deff235-c8ad-4400-b3ac-cb334fdb08ec"
# <a id="instance"></a>
# # 1. Working with the WML service instance

# + [markdown] id="a9b1e86d-cd26-4663-9884-dc0aec3258be"
# In this section you will learn which information is necessary to connect to WML in CPD and which steps are needed to use the WML API client.
#
# You will create and use some WML artifacts (space, model, deployment) which need to have own name. The notebook defines these names in next cells, they will be used in the code of the notebook later. If you want to use your own name you can change them there.

# + id="c3e44eaf-eca7-44a4-a5fe-470dfeb16393"
SPACE_NAME = "STREAMS_WML_SAMPLE_SPACE"
MODEL_NAME = "STREAMS_WML_SAMPLE_MODEL"
DEPLOYMENT_NAME = "STREAMS_WML_SAMPLE_DEPLOYMENT"

# + [markdown] id="d2f1b521-e1df-4c17-83a9-b13e<PASSWORD>"
# **Authenticate to the WML service on CPD**
#
# Running in a Jupyter notebook in CPD you can use the token from the Jupyter environment. This token identifies you in CPD.
# Below you find the WML credential informations needed to connect to WML in CPD.
# The WML service URL is addressed by the CPD internal URL (same for all CPD instances), so you don't take care for this.
# The instance id is a fix value and the version also.

# + id="ec49b6a5-690b-4ef8-b0e8-c5e9444c8d00"
import os

userToken = os.environ["USER_ACCESS_TOKEN"]

wml_credentials = {
                   "url": "https://internal-nginx-svc:12443", # access the main CP4D proxy/the CP4D cluster root from internal side
                   "token": <PASSWORD>Token,
                   "instance_id": "openshift",
                   "version" : "3.5"
                  }

# + [markdown] id="f5951dac-ed98-435c-9e3d-77479d1b5676"
# **Connect to WML in CPD using the WML API client**
#
# Create a WML client by using your credentials.
#
# To work with the client you need also to set a **SPACE** your client should use.
# A **SPACE** is where you can train, store and deploy your models. 
# You can have different **SPACES** depending on your needs (e.g. one for TEST and one for PRODUCTION).
#
# This notebook creates the space for you if it doesn't exist.

# + id="5dac0a183c5643c1821b5f5e0121956e"
from ibm_watson_machine_learning import APIClient
wml_client = APIClient(wml_credentials)
wml_client.version

# + [markdown] id="1db21f0608804b8d857345d083efe8e8"
# **1.0.1 Create new space**
#
# If you have a space already, please skip the cell below and continue with **1.0.2** 

# + id="2de52af460bc4859b46f862b4d79b7eb"
space_metadata = {
    'name': SPACE_NAME,
    'description': DEPLOYMENT_NAME,
}
space_details = wml_client.spaces.store(space_metadata)
print(space_details)

# + id="35e3b2eafc8141cd8790a8112e65f52f"
space_id = wml_client.spaces.get_id(space_details)
print(space_id)

# + [markdown] id="24c0497bdaf54c6289df1c30ead6b56c"
# In order to check if the space creation is completed succesfully change next cell format to code and execute it. It should return 'active'.

# + id="cc6089c9b98f4db1806de31f2ec9da09"
wml_client.spaces.get_details(space_id)['entity']['status']['state']

# + [markdown] id="49f44785ce3b4ed1a88e150faead038c"
# **1.0.2 Use existing space**
#
# If you didn't create new space in this notebook by ibm_watson_machine_learning, please assign space ID below.

# + id="5bd9f9144525444386d4ca56e721446b"
wml_client.spaces.list()

# + id="e56d32ce8dd74cf4b8cb4968d6d201a0"
#space_id = 'PASTE YOUR SPACE ID HERE'

# + [markdown] id="e015435933b44c2396e5cbef298f1a3d"
# **Set default space:** To be able to interact with all resources available in Watson Machine Learning, you need to set space which you will be using

# + id="f299459eedb44448830082e615b2495b"
wml_client.set.default_space(space_id)

# + [markdown] id="857f6c9f-39f8-4b44-a8fc-ab3b7c9efd10"
# Get information about the stored models for your WML instance.

# + id="ce4581d8-e79f-47bf-85cb-547f2c4bfa0d"
# List information about the stored models.
wml_client.repository.list_models()

# + [markdown] id="81351f20-b00d-4f00-85ce-032dd1b46ece"
# Get a list of the deployed models for your WML service instance.

# + id="15d2895d-c7b2-4907-9234-9ca76226ab1d"
# List the deployed models.
wml_client.deployments.list()

# + [markdown] id="f6def848-13f6-476b-a601-98992dbc3ea0"
# ### Delete existing model and deployment
#
# As WML doesn't prevent creation of models or deployments with same name the notebook deletes both if they exist.
#

# + id="9004925b-65fe-41a6-bce6-758c3b8c9030"
deployment_details = wml_client.deployments.get_details()
for deployment in deployment_details['resources']:
    deployment_id = deployment['metadata']['guid']
    model_id = deployment['entity']['asset']['href'].split('/')[3].split('?')[0]
    if deployment['entity']['name'] == DEPLOYMENT_NAME:
        print('Deleting deployment id', deployment_id)
        wml_client.deployments.delete(deployment_id)
        print('Deleting model id', model_id)
        wml_client.repository.delete(model_id)
wml_client.repository.list_models()

# + [markdown] id="e450dbb8-cb13-4f0e-81dc-43aa00ad0a76"
# ### 1.1. Store the model<a id="store"></a>

# + [markdown] id="2315ae3e-68b8-44db-bbdf-c80d49ce3c14"
# **Action**: Use `wget` to download the sample PMML model, `iris_chaid.xml` from the Git project.

# + id="22838bd42df9438984e1fca4af9cdc48"
import os
from wget import download

sample_dir = 'pmml_sample_model'
if not os.path.isdir(sample_dir):
    os.mkdir(sample_dir)
    
filename=os.path.join(sample_dir, 'iris_chaid.xml')
if not os.path.isfile(filename):
    filename = download('https://raw.githubusercontent.com/IBM/watson-machine-learning-samples/master/cpd/models/pmml/iris-species/model/iris_chaid.xml', out=sample_dir)

# + id="6dcd1a6604434c9abd90a99d06ea908c"
sw_spec_uid = wml_client.software_specifications.get_uid_by_name("spark-mllib_2.4")

meta_props = {
    wml_client.repository.ModelMetaNames.NAME: "pmmlmodel",
    wml_client.repository.ModelMetaNames.SOFTWARE_SPEC_UID: sw_spec_uid,
    wml_client.repository.ModelMetaNames.TYPE: 'pmml_4.2.1'}

published_model = wml_client.repository.store_model(model=filename, meta_props=meta_props)

# + id="9bbf42aa298f48f295ccce46be0e19d1"
wml_client.repository.list_models()

# + [markdown] id="67e9c6fe-5eb0-47ba-9d60-cb569ee661db"
# **Note:** From the list of downloaded files, you can see that model is successfully stored in the WML service repository.

# + [markdown] id="61ceaa76-2a1d-47f9-930c-2bedb7153e9c"
# ### 1.2. Create the online deployment<a id="deploy"></a>

# + [markdown] id="d9f8750c-726c-41df-97f9-67fb9caf7b12"
# Now create the online deployment, *Iris species prediction*, for the stored model, then list all the online deployments for the model.
#
# **Hint**
#
# You give models as well as deployments a name. But the name is not checked if it already exists. The name not not the unique identifier, this is only the **guid** of a model and a deployment.
# If you take care with your naming the name may be an identifier for you but you need for all API functions the **guid**.

# + id="5ed3a356-82e2-4064-ae52-e04a6c88cb64"
model_uid = wml_client.repository.get_model_uid(published_model)
deployment = wml_client.deployments.create(
    artifact_uid=model_uid,
    meta_props={
        wml_client.deployments.ConfigurationMetaNames.NAME: DEPLOYMENT_NAME,
        wml_client.deployments.ConfigurationMetaNames.ONLINE:{}}
)

wml_client.deployments.list()
deployment_id = wml_client.deployments.get_id(deployment)

# + [markdown] id="ce34b7c8-3ef4-43c0-8ac5-b231f087ae5c"
# **Note:** From the list of deployed models, you can see that model was  successfully created.

# + [markdown] id="4d3618a8-9628-4e7a-a05f-ed26478a4dcb"
# ### 1.3. Score data<a id="singlescore"></a>

# + [markdown] id="0a8cdc38-a4a1-4b4a-bcde-3f03a6ebc915"
# Send a sample record using the WML Python API client to check that the online scoring endpoint is available. Use here the **guid** of the just created deployment.
#
# The data you will send for scoring defines the fields and the field values.
# You get back the prediction as well as a set of probabilities.

# + id="508d1d63-5dde-497a-b9fa-cf99bc22b918"
# Score data and predict species of flower.
scoring_data = [{'fields': ['Sepal.Length', 'Sepal.Width', 'Petal.Length', 'Petal.Width'], 'values': [[5.1, 3.5, 1.4, 0.2]]}]

predictions = wml_client.deployments.score(deployment_id, meta_props={'input_data':scoring_data})

print(predictions)

# + [markdown] id="ee6e4c58-9f90-4e5b-953b-3fb212103bf0"
# As we can see from the prediction, this is the Iris Setosa flower.

# + [markdown] id="fb71c07c-b4db-47b3-a0d8-819fe0ca58c0"
# <a id="test"></a>
# # 2. Test the performance of your online scoring deployment
#
# The streamsx.wml Python package allows you to use each WML online deployment to score streaming data. 
# WML online scoring is based on a REST API. As such, it has poor performance when each input data is scored in single REST request.
#
# To imporve the performance of WML online scoring, the streamsx.wml Python package uses these methods:
#
# -  Sending multiple input data in one request (mini batch paradigm)
# -  Having multiple request senders
# -  Deploying multple nodes
#
#
# The below performance test code can be used to determine the influence of different settings for number of input data in a request and request senders (Python threads).
# If performance is sufficient, the appropriate settings can be taken as parameters for the Streaming Analytics application which will score your real time data.
#
# There are some points to have in mind:
# -  having a failure in a request with multiple input data will lead to fail the complete request 
# -  so, the higher the number of input data in one request the higher the number to be processed afterwards one by one in case of failure, which takes longer time
# -  so, input data quality is of high importance to use this mini batch paradigm
# -  increasing thread number will work only to certain level, delays caused by Python thread handling will get important
# -  actual input data sequence is kept, which may lead to postpone sending threads which got results earlier than the the ones with earlier data (especially when using multiple deployment nodes)
# -  most important for throughput performance is the parameter __bundle_size__ which configures the maximal amount of tuples being bundled in one REST scoring request
# -  The __bundle_size__ is only the upper limit. Each sending  thread having completet its work will grab from a queue as much tuples as are available upt to __bundle_size__. If there is only one in queue, which means that there is only low input rate, the thread will take this one for its next cycle.
#
# The below code can be taken for any WML online deployment when __field_mapping_dict__ (which assigns streaming data attributes to the mining fields the model expects) and __class TestSource__ are changed according to the model which is deployed.
#
# __space_guid__ and __deployment_guid__ are set by the code cells just ran before.
#
#
# <div class="alert alert-block alert-info">
# The test performance seen when running the test below will be higher as the one you will get when you use same parameters for the Streaming Application later. 
# The reason is simple:
# <ul>    
#     <li> There is no resource intensive upstream or down stream processing.</li>
# <li>At streaming application runtime each tuple has to be converted from its transport representation to Python object when received and from Python object/dictionary to transport representation when it is sent.</li>
# </ul>
# <p/>    
# <b>You can assume that you'll get in your streaming application (created in chapter 3) approximately 1/2 of the performance you will see in this test chapter below.</b>
# <p/>
# <p> The performance varies on the used ML frame work as well as on the used model. You should test the deployment with different parameters for <b>bundle_size</b> and <b>threads_per_node</b>. The parameter <b>node_count</b> has to be changed only when you configured your WML deployment to use more than 1 node.  
# </div>

# + id="85babcb3-e3ce-4966-95bc-0d727921c4ef"
import streamsx.wml
streamsx.wml.__version__

# + id="fab6cd11-5167-4cb9-8cff-e487d16c111c"
from streamsx.wml.bundleresthandler.wmlbundlecontroller import WmlBundleController
import streamsx.wml.utils as wml_utils
import threading
import json
import time


###################################################################################
# Parameters
#
# Collection all the parameters needed for using WML online scoring 
# resp. the underlying Rest Bundlehandler classes 
# 
# deployment_guid and space_guid are set by the code ran before.
# If you want to run this code w/o running the steps before you
# need to set the appropriate guids below and uncomment the 2 lines.
###################################################################################
#deployment_guid = <your_guid>
#space_guid = <your_guid>

field_mapping_dict =[{"model_field":"Sepal.Length",
                      "tuple_field":"sepal_length"},
                     {"model_field":"Sepal.Width",
                      "tuple_field":"sepal_width"},
                     {"model_field":"Petal.Length",
                      "tuple_field":"petal_length"},
                     {"model_field":"Petal.Width",
                      "tuple_field":"petal_width"}]


# streamsx.wml functions support field mapping as JSON 
field_mapping = json.dumps(field_mapping_dict)
# streamsx.wml functions support WML credentials as JSON 
wml_credentials = json.dumps(wml_utils.get_wml_credentials(version = '3.5'))  #token,url,instance_id,version

######################################################
# Callable class used by the REST handler threads.
# It's being called with the results of processed
# records bundle.
######################################################
class OutputClass:
    def __init__(self, interval):
        self._timer_interval = interval
        self._record_counter = 0
        self._result_1_counter = 0
        self._result_2_counter = 0
        self._time_a = time.time()
        self._last_count = 0
        self._timer = threading.Timer(interval,self._timer_action)
    
    def _timer_action(self):
        load = (self._record_counter - self._last_count) / (time.time() - self._time_a)
        print('Load(1/s): ', load, '   Records: ', self._record_counter, '   Records on output 1: ',self._result_1_counter,'   Failed records on output 2: ',self._result_2_counter)
         
    def __call__(self, results):
        if not self._timer.is_alive():
            self._timer = threading.Timer(self._timer_interval,self._timer_action)
            self._time_a = time.time()
            self._last_count = self._record_counter
            self._timer.start()
        for index,result_list in enumerate(results):
            # first returned list are the successful scored record 
            #(and ... depending on parameter single_output ... the failed)
            if index == 0:
                for list_element in result_list:
                    self._record_counter +=1
                    self._result_1_counter += 1
            # second returned list are the failed records 
            # if 'single_output = False'
            elif index == 1:
                for list_element in result_list:
                    self._record_counter +=1
                    self._result_2_counter += 1
            else:
                print("Internal error: More result lists generated than supported. ")
        


            
###################################################################################
# Test source
#
# Class with callable generating data as needed for the model which should be used.
# In streaming application the connect stream is the source.
###################################################################################
class TestSource:
    def __init__(self, ):
        pass
    def __call__(self):
            # run this indefinitely so that there will always be data for the view
        counter = 0
        while True:
            counter += 1
            record = {"petal_length":1.4,
                       "petal_width":0.2,
                       "sepal_length":5.1,
                       "sepal_width":3.5,
                       "number" : counter}
            # generate errors
            #if counter % 20 == 0:
            #    record.pop("petal_length",None)
            #    record.pop("sepal_length",None)

            #yield everytime same values, doesn't matter for test
            yield record


###################################################################################
# Test class
#
# The test application which is using the streamsx.wml WmlBundlerController, 
# same as the one used in function for Streaming Analytics applications provided by streamsx.wml
###################################################################################
class PerformanceTest:
    def __init__(self,source, **params):
        self._params = params
        self._source_generator = source()

    def _generate_data(self):
        while self._run:
            #for data in TestSource():
            self._testcontroller.process_data(next(self._source_generator))

    def start(self):
        self._testcontroller = WmlBundleController(**self._params)
        self._testcontroller.prepare()
        self._testcontroller.run()
        self._run = True
        self._source_thread = threading.Thread(target=self._generate_data)
        self._source_thread.start()

    def stop(self):
        self._run = False
        self._source_thread.join()
        self._testcontroller.stop()
        self._testcontroller.finish()



# + [markdown] id="bf6f3f98-9f62-47bd-b524-9b96bef0658f"
# ### Create a test instance with your parameters

# + id="75ce25ff-340d-4102-b35f-ca24223da24f"
####################################################
# Dictionary of all parameters
#
# as used also in Streaming application
# except the 'output_function' parameter which is 
# added by the streaming operator as the output is
# special for any implementation using the 
# WmlBundleController
####################################################            
params= {
    'deployment_guid':deployment_id,
    'wml_credentials':wml_credentials,
    'space_guid':space_id,
    'bundle_size':  1000,
    'queue_size': 10000, 
    'threads_per_node': 2,
    'node_count':1,
    'field_mapping':field_mapping, 
    'output_function':OutputClass(interval=5.0 ),
    'single_output': False}

test = PerformanceTest(TestSource(), **params)

# + [markdown] id="cec12712-7bad-4271-bc9b-c096acf3dd14"
# ### Start the test
#
# It will print every 5 secods a metrics as following:
#
# `Load(1/s):  10194.683301388393    Records:  51000    Records on output 1:  51000    Failed records on output 2:  0`
#
# It shows the actual determined load for the 5s intervall the records/input data processed in this time and the records/scoring results which were successful (output 1) and failed (output 2). Failed will be seen only when you set your __class TestSource__ to generate some error input data.

# + id="5d7682d2-3e33-46d5-a120-cd07e02d912d"
test.start()

# + [markdown] id="7b45c6a2-0b91-459f-b98d-4fe9c49f196c"
# ### Stop the test
#
# __!!!Please stop the test before you change the parameters and start again!!!__
#
# There are threads running which need to be stopped!

# + id="2b2d6da2-26b5-4bef-a699-a19e81885796"
test.stop()

# + [markdown] id="84f00826-7a8b-4a97-8317-0cc6dcaff6c7"
# <a id="application"></a>
# # 3. Create and run a Streaming Analytics application using this WML online scoring deployment

# + [markdown] id="2a8c6968-d4c4-441d-8831-542b39c7a2d3"
#
# <a id="setup"></a>
#
# ## 3.1. Setup
#
#   
#   
# <div class="alert alert-block alert-info">
# <b>What do we need to set up for our sample application?</b>  
# <ul>    
# <li>A connection to a running <b>IBM Streams instance</b> <br> 
#    This notebook describes using a Streams instance in IBM Cloud Pak for Data.</li>
# <li><b>streamsx</b> - the Python package providing the IBM Streams standard/basic function set</li>
# <li><b>streamsx.wml</b> - the Python package providing the IBM Streams WML online scoring functionality</li>
# <li><b>Data set</b> with sample Iris data</li>
# <li><b>WML deployment</b> of the PMML Iris species prediction model (done above) </li>
# </ul>
# </div>
#
#
#
# <a id="setupStreams"></a>
#
# ### 3.1.1 Add credentials for the IBM Streams service
#
# In order to submit a Streams application you need to provide the name of the Streams instance.
#
# 1. From the navigation menu, click **Services > Instances**.
# 2. Update the value of `streams_instance_name` in the cell below according to your Streams instance name.

# + id="e5c3b3b2-d9c5-4c63-98a1-82bcc73fa55a"
from icpd_core import icpd_util
streams_instance_name = "tooling-55-cpd"#-streams" ## Change this to your Streams instance

try:
    cfg=icpd_util.get_service_instance_details(name=streams_instance_name, instance_type="streams")
except TypeError:
    cfg=icpd_util.get_service_instance_details(name=streams_instance_name)

# + [markdown] id="f06ad84d-9c10-4a2d-b957-611365b54dff"
# <a id="setupPackages"></a>
#
# ### 3.1.2 Import  the `streamsx` and `streamsx.wml` package and verify the package version

# + id="15dad156-c998-49b7-baed-647c094d736e"

import streamsx.topology.context
import streamsx.wml
print("INFO: streamsx package version: " + streamsx.topology.context.__version__)
print("INFO: streamsx.wml package version: " + streamsx.wml.__version__)

# + [markdown] id="c13dce09-209f-4d0c-8ccd-d53cf6019339"
# <a id="setupDataModel"></a>
#
# ### 3.1.3 Add test data your project
#

# + id="8639fd96-ece4-4424-a7b9-2b9c8fb1d22c"
# Download sample data from Git.
import wget, os

sample_dir = 'pmml_sample_model'
if not os.path.isdir(sample_dir):
    os.mkdir(sample_dir)
    
filename_data=os.path.join(sample_dir, 'iris.csv')
if not os.path.isfile(filename_data):
    filename_data = wget.download('https://ibm.box.com/shared/static/nnxx7ozfvpdkjv17x4katwu385cm6k5d.csv', out=sample_dir)

# + [markdown] id="1ffd1aa5-37e0-41f2-b56d-ddf1e62e3ffe"
# <a id="create"></a>
#
# ## 3.2. Create the application
#
#
# All Streams applications start with  a `Topology` object, so we start by creating one.<br>Additionally we add our data set to the topology as a dependency, which means that is bundled within the built application and so available at the environment where the application will run.

# + id="fa918b06-8066-4933-a51e-51a9ef8b25ca"
from streamsx.topology.topology import Topology
import os

topo = Topology(name="WMLScoring", namespace="sample")

# add files to be contained in the archive which is deployed to the node running the application
# in this sample we need the Iris `dataset` with sample data to be present at the worker node
topo.add_file_dependency(filename_data, 'etc')

# + [markdown] id="fb8857d4-1bb5-4a70-b430-86186c6f64c0"
# <a id="defineInput"></a>
#
# ### 3.2.1 Define Data Input 
#
# Let's take a look at the sample data, a set of measured data records:

# + id="48c081fc-ebcd-4858-b56c-e6a293645042"
sample_data = filename_data
# !head $sample_data

# + [markdown] id="4941a251-8a8e-4596-ad34-383e098b6320"
# We will use this CSV file as data source for this sample. To simulate an endless stream of data we will read the file repeatedly and submit each line in the file as a tuple. The class below reads the data set and produces a stream of Python `dict` objects. The produced `Stream` is a stream of Python `dict`.
#
# `Stream data : {"sepal_length" : <sepal_length_input>,
#                "sepal_width"  : <sepal_width_input>, 
#                "petal_length" : <petal_length_input>,
#                "petal_width"  : <petal_width_input>,
#                "referenceSpecies" : <species_target>,
#                "Prediction" : ""}`
#
# "Prediction" will be the field which contains the result from scoring. Depending on model and model framework the "Prediction" may be single value or a dictionary of multiple values.

# + id="a484c0e6-7a8e-4382-882b-2f9871f2e8dd"
import csv
from streamsx.ec import get_application_directory
import time 

# a class which can be used in topology as a source
class CSVFileReader:
    def __init__(self, file_name):
        self.file_name = file_name
    def __call__(self):
        # Convert each row in the file to a dict
        drug_file =  os.path.join(get_application_directory(), "etc", self.file_name)
        col_names = ["sepal_length", "sepal_width", "petal_length", "petal_width", "referenceSpecies"]
       
        # run this indefinitely so that there will always be data for the view
        while True:
            with open(drug_file) as handle:
                reader = csv.DictReader(handle, delimiter=',',
                                                fieldnames=col_names)
                #Use this to skip the header line if your file has one
                next(reader)
                #yield the lines in the file one at a time
                #throttle
                throttle_count = 0
                for row in reader:
                    throttle_count += 1
                    if throttle_count % 4 is 0:
                        throttle_count = 0
                        time.sleep(0.0001)
                    yield dict(sepal_length= float(row["sepal_length"]),
                                     sepal_width = float(row["sepal_width"]), 
                                    petal_length= float(row["petal_length"]),
                                    petal_width=float(row["petal_width"]) ,
                                    referenceSpecies=row["referenceSpecies"],
                                    predictedSpecies="")

# let the csv file reader the source/edge node in our topology, producing the 'records' stream                    
records = topo.source(CSVFileReader("iris_initial.csv"))

# Let the source run in separate Python interpreter instance.
# streamsx.wml.wml_online_scoring() is a Python function and for bestt performance it will need all available ressources of an own Python interpreter instance.
records.isolate()


# + [markdown] id="dcfb8b01-af17-4884-b427-771862e052d1"
# <a id="analyzeData"></a>
# ### 3.2.2 Score data and run the application
#
#
# Now that we have created a stream of Iris data records, we need to define the analytics we want to perform on our data. Our WML deployment (PMML model) is predicting the Iris species from the measured data.  
#
# <div class="alert alert-block alert-info">
# <p>These are the core steps you need for using WML model scoring in your application:</p>
# <ul>
#     <li><b>Data preparation</b><br>Make sure your input stream meets your deployments (models) expected input features
#     </li>
#     <li><b>Score using <code>streamsx.wml.wml_online_scoring()</code>, specifying: </b><br>
#         - the input stream which should be processed <b>mandatory</b><br>
#         - the deployment_guid of the deployment we want to use <b>mandatory</b><br>
#         - the field_mapping from input stream fields to model input features <b>mandatory</b><br>
#         - the credentials to get access to the deployment <b>mandatory</b><br>
#         - the space_guid of the deployment <b>mandatory</b><br>
#         - the queue_size <br>
#         - the bundle_size to configure the maximum tuples in one scoring request<br>
#         - the threads_per_node to configure the number of threads used to send request to WML REST endpoint<br>
#         - the node_count if you configured you WML deployment to use more than 1 node<br>
#     </li>
# </ul>
#
# <p>In the code below we use values which were seen to give a good performance. It is assumed that there is one deployment node created. You may later play around with the values to see if they match the expectations you have from the performance test before</p>
#     
# </div>

# + id="c753d638-f7d5-4cd1-9068-4431e14c4e1b"
from streamsx.topology import context
import streamsx.wml.utils as wml_utils
import json

#####################################################################################
# Define the parameter for the scoring function
#####################################################################################
field_mapping_dict =[{"model_field":"Sepal.Length",
                      "tuple_field":"sepal_length"},
                     {"model_field":"Sepal.Width",
                      "tuple_field":"sepal_width"},
                     {"model_field":"Petal.Length",
                      "tuple_field":"petal_length"},
                     {"model_field":"Petal.Width",
                      "tuple_field":"petal_width"}]

# streamsx.wml.wml_online_scoring() support field mapping as JSON or dict
field_mapping = json.dumps(field_mapping_dict)

# streamsx.wml.wml_online_scoring() functions support WML credentials as JSON or dict
wml_credentials = json.dumps(wml_utils.get_wml_credentials(version = '3.5'))  #token,url,instance_id,version

# function call parameter
params= {
    'deployment_guid':deployment_id, #the one from the cells run above
    'field_mapping':field_mapping, 
    'credentials':wml_credentials,
    'space_guid':space_id,           #the one from the cells run above
    'bundle_size':  1000,
    'queue_size': 10000, 
    'threads_per_node': 2,
    'node_count':1,
    'single_output': False}


#####################################################################################
# Add the scoring function to your topology
#
# 2 result streams are generated: 
#        1st for successful scorings, 
#        2nd for failed scorings or invalid input
####################################################################################
scorings,invalids = streamsx.wml.wml_online_scoring(records, 
                                                  **params)
# for best load results in this test application
# separate the scoring from down stream functionality of publish() and view()
scorings.isolate()
scorings_mapped = scorings.map(lambda x: x)

# publish results as JSON
scorings_mapped.publish(topic="ScoredRecords",schema=json,name="PublishScores")
score_view = scorings_mapped.view(name="ScoredRecords", description="Sample of scored records")



# -

# # <a name="submit"> </a> 3. Submit the application
#
# A running Streams application is called a *job*. This next cell submits the application for execution and prints the resulting job id.

# + id="c67287117014471c821f7965465ce21a"


# Disable SSL certificate verification if necessary
cfg[context.ConfigParams.SSL_VERIFY] = False


#####################################################################################
# Build and submit your topology
#####################################################################################
submission_result = context.submit('DISTRIBUTED', 
                                   topo, 
                                   cfg)


# The submission_result object contains information about the running application, or job
if submission_result.job:
    streams_job = submission_result.job
    print ("JobId: ", streams_job.id , "\nJob name: ", streams_job.name)

# + [markdown] id="f3af538e-a391-473d-abd3-6491c89ee9a6"
# ## 3.1 Use the View to access data from the job
#
# Now that the job is started, use the View object you have already created to start retrieving successfully scored data.

# + id="cc1a89f3-6884-4aa8-8af9-cc4558141f10"
# Connect to the view and display the selected data
queue = score_view.start_data_fetch()
try:
    for val in range(20):
        print(queue.get(timeout=60))    
finally:
    score_view.stop_data_fetch()

# + [markdown] id="53fd09d9-006b-4042-8ba3-3aed23794a92"
#
# ## 3.2 See job status
#
# The tools available to monitor the running application depend on the version of Streams and your development environment.
#
# - **If you are using a Cloud Pak for Data 3.5 project:**  When you submit the `Topology`, you create a new <i>job run</i>. The job represents the application and the job run represents a single instance of the running application. 
#
#     1. Open your project and click on the **Jobs** tab. This will show a list of the project's jobs.
#     1. Under the **Job name** column, find your job based on the `Job Name` [printed when you submitted the job](#submit). This will list all of the job runs for that job.
#     1. Click the **Run name** to open the job run. The run name will be the same as the `Job Name` printed above.
#     
#
# - **For all other development environments and versions of Streams**, [see this page for more information](http://ibmstreams.github.io/streamsx.documentation/docs/spl/quick-start/qs-4).
#
# # 4. Cancel the application
# The Streams job is running in the Streams service. You can cancel it within the notebook or delete it from **Projects > Jobs**.

# + id="d4d82964-7107-4e5c-8b21-bffe7f448e23"
# cancel the job in the IBM Streams service
# submission_result.cancel_job_button()
# Or, 
# cancel the job directly using the Job object
streams_job.cancel()

# + [markdown] id="0c2a2468048c49999b3d0e6e3109d284"
# ## 4.1  Delete existing space
# You can use the command below to delete existing space. You need to provide space_id of the space you want to delete.

# + id="2860335ba0174ec6948bc1080be5cdf8"
wml_client.spaces.delete(space_id)

# + [markdown] id="6bfdb07b-1859-4c64-b471-a73cfde75152"
# <a id="summary"></a>
# # 5. Summary and next steps     

# + [markdown] id="23dff00c-3100-4c15-8b1d-ab2c76836987"
# You successfully completed this notebook! 
# <p>You learned how to use Watson Machine Learning for PMML model storage, deployment and scoring.</p>
# <p>You have further used the deployment to run a performance test in this notebook to see how many scorings per second can be reached with the WML provided <b>mini-batch</b> approach.</p>
# <p>Last you have created a Streaming Analytics application which is using the WML deployment for online scoring of its streaming data. You saw that you can use same parameters set in performance test.</p>
#

# + [markdown] id="83e06c73-a0d0-46fe-a1b9-4b4ab91dc177"
# ### Citations
#
# <NAME>. and <NAME>. (2017). [UCI Machine Learning Repository](http://archive.ics.uci.edu/ml). Irvine, CA: University of California, School of Information and Computer Science.
#

# + [markdown] id="ffb9f95b-9624-444a-b519-41620ea2c62b"
# Copyright © 2020 IBM. This notebook and its source code are released under the terms of the MIT License.

StarcoderdataJupyterScriptsDedupFiltered