Python API Use Guide¶

Sedna Python SDK¶

The Sedna Python Software Development Kit (SDK) aims to provide developers with a convenient yet flexible tool to write the Sedna applications.

This document introduces how to obtain and call Sedna Python SDK.

Introduction¶

Expose the Edge AI features to applications, i.e. training or inference programs.

Requirements and Installation¶

The build process is tested with Python 3.6, Ubuntu 18.04.5 LTS

# Clone the repo
git clone --recursive https://github.com/kubeedge/sedna.git
cd sedna/lib

# Build the pip package
python setup.py bdist_wheel

# Install the pip package
pip install dist/sedna*.whl

Install via Setuptools

# Install dependence
pip install -r requirements.txt

# Install sedna
python setup.py install --user

Use Python SDK¶

(optional) Check Sedna version

$ python -c "import sedna; print(sedna.__version__)"

Import the required modules as follows:

from sedna.core.joint_inference import JointInference, BigModelService
from sedna.core.federated_learning import FederatedLearning
from sedna.core.incremental_learning import IncrementalLearning
from sedna.core.lifelong_learning import LifelongLearning

Define an Estimator:

import os

# Keras
import keras
from keras.layers import Dense, MaxPooling2D, Conv2D, Flatten, Dropout
from keras.models import Sequential

os.environ['BACKEND_TYPE'] = 'KERAS'

def KerasEstimator():
    model = Sequential()
    model.add(Conv2D(64, kernel_size=(3, 3),
                     activation="relu", strides=(2, 2),
                     input_shape=(128, 128, 3)))
    model.add(MaxPooling2D(pool_size=(2, 2)))
    model.add(Conv2D(32, kernel_size=(3, 3), activation="relu"))
    model.add(MaxPooling2D(pool_size=(2, 2)))
    model.add(Flatten())
    model.add(Dropout(0.25))
    model.add(Dense(64, activation="relu"))
    model.add(Dense(32, activation="relu"))
    model.add(Dropout(0.5))
    model.add(Dense(2, activation="softmax"))

    model.compile(loss="categorical_crossentropy",
                  optimizer="adam",
                  metrics=["accuracy"])
    loss = keras.losses.CategoricalCrossentropy(from_logits=True)
    metrics = [keras.metrics.categorical_accuracy]
    optimizer = keras.optimizers.Adam(learning_rate=0.1)
    model.compile(loss=loss, optimizer=optimizer, metrics=metrics)
    return model

# XGBOOST

import os
import xgboost

os.environ['BACKEND_TYPE'] = 'SKLEARN'

XGBEstimator = xgboost.XGBClassifier(
        learning_rate=0.1,
        n_estimators=600,
        max_depth=2,
        min_child_weight=1,
        gamma=0,
        subsample=0.8,
        colsample_bytree=0.8,
        objective="multi:softmax",
        num_class=3,
        nthread=4,
        seed=27
 )

# Customize

class Estimator:

    def __init__(self, **kwargs):
        ...

    def load(self, model_url=""):
        ...

    def save(self, model_path=None):
        ...

    def predict(self, data, **kwargs):
        ...

    def evaluate(self, valid_data, **kwargs):
        ...

    def train(self, train_data, valid_data=None, **kwargs):
        ...

Notes: Estimator is a high-level API that greatly simplifies machine learning programming. Estimators encapsulate training, evaluation, prediction, and exporting for your model.

Initialize a Incremental Learning Job:

   # get hard exmaple mining algorithm from config
   hard_example_mining = IncrementalLearning.get_hem_algorithm_from_config(
       threshold_img=0.9
   )

   # create Incremental Learning infernece instance
   il_job = IncrementalLearning(
       estimator=Estimator,
       hard_example_mining=hard_example_mining
   )

where:

IncrementalLearning is the Cloud-edge job you want to access.
Estimator is the base model for your ML job.
hard_example_mining is the parameters of incremental learning job.

Inference

Note: The job parameters of each feature are different.

Running Job - training / inference / evaluation.
```
   results, final_res, is_hard_example = il_job.inference(
           img_rgb,
           post_process=deal_infer_rsl,
           input_shape=input_shape
   )

where:
```
- img_rgb is the sample used to inference
- deal_infer_rsl is a function used to process result after model predict
- input_shape is the parameters of Estimator in inference
- results is the result predicted by model
- final_res is the result after process by deal_infer_rsl
- is_hard_example tells if the sample is hard sample or not

Customize algorithm¶

Sedna provides a class called class_factory.py in common package, in which only a few lines of changes are required to become a module of sedna.

Two classes are defined in class_factory.py, namely ClassType and ClassFactory.

ClassFactory can register the modules you want to reuse through decorators. For example, in the following code example, you have customized an hard_example_mining algorithm, you only need to add a line of ClassFactory.register(ClassType.HEM) to complete the registration.

@ClassFactory.register(ClassType.HEM, alias="Threshold")
class ThresholdFilter(BaseFilter, abc.ABC):
    def __init__(self, threshold=0.5, **kwargs):
        self.threshold = float(threshold)

    def __call__(self, infer_result=None):
        # if invalid input, return False
        if not (infer_result
                and all(map(lambda x: len(x) > 4, infer_result))):
            return False

        image_score = 0

        for bbox in infer_result:
            image_score += bbox[4]

        average_score = image_score / (len(infer_result) or 1)
        return average_score < self.threshold

After registration, you only need to change the name of the hem and parameters in the yaml file, and then the corresponding class will be automatically called according to the name.

deploySpec:
    hardExampleMining:
      name: "Threshold"
      parameters:
        - key: "threshold"
          value: "0.9"