In this tutorial, you are writing a custom function that you package and store in a GitHub repository. You are testing the function in your local environment before you register it with the catalog and use it in your calculations.
In this tutorial, you are writing a custom function, MultiplyByFactor
, that multiples an input item by a factor value. As part of the function code, define how the UI controls for input and output items are represented on the configuration
UI in Analytics Service. Later, specify the input item and the factor value through the configuration user interface.
Use the execute_local_test
method to test the function. You use this method to test any function that is derived from base classes in IoT Functions. The method generates sample data for a function locally. It writes the results to a file
named df_test_entitity_for_function_name
in the working directory. If the function works, register it with Analytics Service.
Finally, create a sample robot device type in Analytics Service and apply the new MultiplyByFactor
function to the device type.
This tutorial uses the Pycharm IDE to create the Python package for the custom function. If you use another IDE or Watson Studio, make sure to use Python 3.x with it.
MultiplyByFactor
, use a unique name for the function in your own code, such as MultiplyByFactor<YourInitials>
.
For example, you might name your function MutlipleByFactorLD
.Alternatively, separate from this tutorial, a custom function starter package is available in IoT functions to help you get started with building your own custom functions.
You can clone a starter package and use it as a sample package in your local environment. Cloning a starter package provides you with an easy way to install the required packages, including IoT Functions, for your custom function. The starter package provides an example of the required directory structure for your custom package.
mkdir project
.cd project
.git clone --branch starter_package https://github.com/ibm-watson-iot/functions.git
cd functions
git status
pwd
and copy the path. pip3 install git+https://github.com/ibm-watson-iot/functions.git@production --upgrade
pip3 install git+https://github.com/ibm-watson-iot/functions.git@beta --upgrade
Make sure that you are always using the latest version of IoT Functions. Check the version of each Python module that is supported by Analytics Service in the Python module dependencies topic. If you do not use the exact version that is supported, your pipeline might break.
Complete these steps to apply a calculation to sample data by using a custom function.
Acme is a company that develops robotic arms. The company is introducing a new robotic arm and is testing that it performs as well as the existing robotic. In early testing, the team at Acme discovered that for some of the robots, the distance traveled during testing was too high. After investigation, the team discovered that the tools used to do the testing were adding a delay to the speed and travel times of the robot. The operations manager wants to adjust both values by a factor of 2 but wants the flexibility to change that value later.
The analyst identifies that a similar calculation might be useful in other calculations and generalizes the function by writing a MultiplyByFactor
class.
distance * factor
test.md
and click Commit new file
.functions
|
|__ setup.py
|
|__ scripts
|
|_ local_test_of_function.py
|
|__ custom
|
|_ functions.py
|_ __init__.py
custom
, to a unique name, such as customyourinitials
. For example, customld
.Verify that the contents of the setup.py
file includes the following code: Remember: Change the module name to match your module name.
from setuptools import setup, find_packages
setup(name='customld', version='0.0.1', packages=find_packages(),
install_requires=['iotfunctions@git+https://github.com/ibm-watson-iot/functions.git@production'])
Create a multiplybyfactor<yourinitials>.py
function module in custom<yourinitials>
and paste the following code. Update the variables in <>
to match your environment. For example, you might
set PACKAGE_URL
to 'git+https://<XXXXXX>@github.com/github.com/jones/starter@starter_package'
. Note: In this function, you implement two methods. You add your calculation to the execute
method. You set the inputs and outputs arguments for the function in the build_ui
method. These arguments are configured through the configuration user interface.
import inspect
import logging
import datetime as dt
import math
from sqlalchemy.sql.sqltypes import TIMESTAMP,VARCHAR
import numpy as np
import pandas as pd
from iotfunctions.base import BaseTransformer
from iotfunctions import ui
logger = logging.getLogger(__name__)
# Specify the URL to your package here.
# This URL must be accessible via pip install.
# Example assumes the repository is private.
# Replace XXXXXX with your personal access token.
# After @ you must specify a branch.
PACKAGE_URL = 'git+https://XXXXXX@github.com/<user_id><path_to_repository>@starter_package'
class MultiplyByFactor<YourInitials>(BaseTransformer):
def __init__(self, input_items, factor, output_items):
self.input_items = input_items
self.output_items = output_items
self.factor = float(factor)
super().__init__()
def execute(self, df):
df = df.copy()
for i,input_item in enumerate(self.input_items):
df[self.output_items[i]] = df[input_item] * self.factor
return df
@classmethod
def build_ui(cls):
#define arguments that behave as function inputs
inputs = []
inputs.append(ui.UIMultiItem(
name = 'input_items',
datatype=float,
description = "Data items adjust",
output_item = 'output_items',
is_output_datatype_derived = True)
)
inputs.append(ui.UISingle(
name = 'factor',
datatype=float)
)
outputs = []
return (inputs,outputs)
Set credentials to connect to Analytics Service.
credentials_as.json
file in the scripts
folder in your working directory. On the user interface, go to the Services tab. credentials_as.json
file.The credentials file is used to run or test the function locally. Do not push this file to your external repository in step 4.
Note:
Push your function code to your external repository in GitHub.
git remote -v
. The remote repository is still pointing to https://github.com/ibm-watson-iot/functions
git remote set-url origin URL_To_YOUR_GITHUB_REPOSITORY
Confirm that the Fetch and Push URLs are pointing to your repository. git remote add upstream https://github.com/ibm-watson-iot/functions.git
custom<your_initials>
directory.starter_package
branch in your repository.Install the custom function in your local environment. Update the variables in <>
to match your environment.
pip3 install git+https://<XXXXXX>@github.com/<user_id><path_to_repository>@starter_package --upgrade
For example:
pip3 install git+https://<XXXXXX>@github.com/jones/starter@starter_package --upgrade
Replace
test_my_custom_function.py
in the scripts
folder. In the script, import Python libraries and packages:import datetime as dt
import json
import pandas as pd
import numpy as np
from sqlalchemy import Column, Integer, String, Float, DateTime, Boolean, func
from iotfunctions.base import BaseTransformer
from iotfunctions.metadata import EntityType
from iotfunctions.db import Database
from iotfunctions import ui
test_my_custom_function.py
script, add:with open('credentials_as.json', encoding='utf-8') as F:
credentials = json.loads(F.read())
db_schema = None
db = Database(credentials=credentials)
test_my_custom_function.py
script, add the following code. Update the variables in <>
to match your environment.from custom<yourinitials>.multiplybyfactor<yourinitials> import MultiplyByFactor<YourInitials>
fn = MultiplyByFactor<YourInitials>(
input_items = ['speed', 'travel_time'],
factor = '2',
output_items = ['adjusted_speed', 'adjusted_travel_time']
)
df = fn.execute_local_test(db=db, db_schema=db_schema, generate_days=1,to_csv=True)
print(df)
df_test_entity_for_multiplybyfactor<your_initials>.csv
in the scripts directory. Enter:python3 test_my_custom_function.py
register_my_custom_function.py
in the scripts
folder. In the script, import Python libraries and packages:import datetime as dt
import json
import pandas as pd
import numpy as np
from sqlalchemy import Column, Integer, String, Float, DateTime, Boolean, func
from iotfunctions.base import BaseTransformer
from iotfunctions.metadata import EntityType
from iotfunctions.db import Database
from iotfunctions import ui
register_my_custom_function.py
script, add:with open('credentials_as.json', encoding='utf-8') as F:
credentials = json.loads(F.read())
db_schema = None
db = Database(credentials=credentials)
In the register_my_custom_function.py
script, to register the function, add the following code. Update the variables in <>
to match your environment.
from custom<yourinitials>.multiplybyfactor<yourinitials> import MultiplyByFactor<YourInitials>
db.register_functions([MultiplyByFactor<YourInitials>])
python3 register_my_custom_function.py
KPI data takes up to 5 minutes to generate.
distance
metric, add it. Follow the steps in Tutorial: Adding expressionsMultiplyByFactor<YourInitials>
function from the catalog. input_item
field, select distance
.adjusted_distance
. Wait up to 5 minutes for Analytics Service to evaluate the custom function against the sample data.