How to test edge configuration package locally

When the pipeline is exported as an edge package, the following checks are automatically performed:

• Verifying that the Python version required in the package is supported by a known version of AI Inference Server.

• Verifying that all the required Python packages exist for the target platform and are included in the edge package.

NOTE:The simaticai.deployment.Pipeline.export() method is available from AI SDK v2.1.0.

For concrete programming details, please refer to the SDK API reference manual.

Class LocalPipelineRunner in submodule simaticai.testing.pipeline_runner can be used to locally mimic the behavior of the AI Inference Server concerning loading and running inference pipelines. This is a quick and easy way to find programming or configuration errors before deploying the package.

The local pipeline runner performs the simulation of the server environment as follows.

1. It unpacks the pipeline components into a test folder, similar to how it would happen on the inference server.

2. It creates for each component an own Python virtual environment.

3. It installs required Python packages from the wheels if provided in the package or from pypi.org.

4. It installs a mock of the log_module (see Mocking the logger of AI Inference Server).

5. It updates pipeline parameters if applicable.

6. It feeds the pipeline with input data by triggering the entrypoints of the components accordingly.

7. It collects the sequence of pipeline outputs for a given sequence of pipeline inputs.

The local pipeline runner also lets you exercise your pipeline component by component. In other words, you can feed single components with inputs and verify the output produced.

If the pipeline has parameters, they will be supplied with their default values. You can also change parameter values using the update_parameters() method. This way you can test your pipeline with different parameters. You can use this method before calling run_component() or run_pipeline() but you cannot change pipeline parameters while these methods are running.

Furthermore, the local pipeline runner creates a report file that contains information about the run. The name of the report file starts with the name of the package, and ends with _execution_report_[n].md, where n means the n-th run. The report file is placed in the same folder where the pipeline package is located.

The report file has the following content:

• Folder structure of the package zip file.

• Content of the PythonPackages.zip subfolders.

• Installed packages for the virtual environment that the local pipeline runner uses.

• The payload counts of the various components.

• Warnings during the run, if there were any.

From a test strategy and risk based testing point of view, we suggest that you verify the business logic within the pipeline components in unit tests as with any ordinary Python program, and use the local pipeline runner to cover test risks such as:

• Mismatch between pipeline and component input and output variable names.

• Required Python packages not covered by requirements.txt.

• Source or other files missing from the package.

• Interface mismatch between subsequent pipeline components.

• Entrypoint cannot consume input data due to data format mismatch.

• Entrypoint produces output data in wrong format.

• For some reason, the pipeline doesn't work end-to-end as intended.

One crucial point for making the local test faithful concerning data input and output formats is understanding how data connections in AI Inference Server work. The following data connection types are straightforward:

• Databus

• External Databus

• IE Vision Connector

For these data connection types, AI Inference Server passes the MQTT payload string directly as the value of the connected pipeline input variable. In many use cases where you use this data connection type, your pipeline has a single input variable of type string. This means that you must pass local pipeline runner a Python dictionary with a single element.

Taking the pipeline from the Image Classification end-to-end tutorial for example, you have a single input variable vision_payload. To run your pipeline for two consecutive input images, you must invoke the pipeline runner like this:

pipeline_input1 = { 'vision_payload': mqtt_payload1 }
pipeline_input2 = { 'vision_payload': mqtt_payload2 }
pipeline_output = runner.run_pipeline([pipeline_input1, pipeline_input2])

For a complete code example that shows how to feed a pipeline with a single string input variable in a local test, see the Local Pipeline Test Notebook in the Image Classification end-to-end tutorial.

​The SIMATIC S7 Connector and the SIMATIC S7+ Connector data connection type requires more attention. These connectors are typically used in time series use cases. The S7+ connector currently supports bulk and timeseries formats. From the user's perspective on the AI Inference Server there is practically no difference between the two, only in the way the inter-signal alignment handles them. With the bulk format each variable can have its own timestamp and the inter-signal alignment can split them into separate groups. With the timeseries format there is only one common timestamp for all values, therefore no splitting and realigning happens, this makes the processing a bit quicker.

Using the S7 type connections the AI Inference Server processes the MQTT payload according to the S7 Connector format and only passes on the values of the PLC variables, but not the metadata. So, if you intend to use your pipeline with the S7 or S7+ connector, you need to feed it with dictionaries holding the PLC tag values.

Taking the pipeline from the State Identifier end-to-end tutorial for example, you have input variables ph1, ph2 and ph3, which are meant to be used with the SIMATIC S7 Connector data connection type. To mimic how AI Inference Server feeds the pipeline, you must call the pipeline runner like this:

pipeline_input1 = {'ph1': 4732.89, 'ph2': 4654.44, 'ph3': 4835.02}
pipeline_input2 = {'ph1': 4909.13, 'ph2': 4775.16, 'ph3': 4996.67}
pipeline_output = runner.run_pipeline([pipeline_input1, pipeline_input2])

For a complete code example that shows how to feed a pipeline with an input line of PLC tag values in a local test, see the Local Pipeline Test Notebook in the State Identifier end-to-end tutorial.

The local runner works with batches of input data and processes the whole input batch component by component. In other words, given a sequence of pipeline inputs, the entire sequence is first processed by the first components, and only then is the output of the first component processed by the second component. This is different from the runtime environment on AI Inference Server, where the components in the pipeline potentially start consuming input as soon as the preceding component has produced output.

You cannot fully test input and output data formats, as these depend on the data connection settings in AI Inference Server, and you have to provide the local runner the input data in the representation that matches the output side of the connector. This means that if your assumptions on the data connection settings or the resulting data formats are wrong, your tests will provide misleading results, too.

The local runner can only simulate linear pipelines, where the pipeline input variables are only used by one component, each component uses only the outputs of the previous components, and the pipeline output only consists of variables from the last component.

Furthermore, the results obtained in local tests are not fully representative for AI Inference Server, including but not limited to the following aspects:

• The local version of Python might differ from that in the Inference Server.

• The local architecture might differ, resulting in different builds of imported Python packages being used.

• The local runner executes only one instance of the Python code, regardless of the parallelism setting in the configuration.

Despite all these limitations, we would like to highlight again our recommendation to test your pipeline locally before deploying it. Most likely it will save you more time and trouble compared to skipping this step.

Class ComponentRunner in submodule simaticai.testing.component_runner is a similar tool to LocalPipelineRunner, capable of testing a single Python component of the pipeline.

The component runner performs the simulation of the server environment as follows.

1. Creates a working directory for the component.

2. Copies the required resources of the component to the working directory. In case the source folder is the same as the working directory, the resources are not copied, so any changes in the source will affect the working directory.

3. Enlists the dependencies into the requirements.txt. If the component is a GPU runtime component, requirements.txt is extended with the required onnx and onnxruntime dependencies.

4. Adds the runner script to the working directory.

5. Creates a Python virtual environment in the working directory. It has the same version as the current Python interpreter.

6. Installs the required Python dependencies.

The run() method runs the component runner with a given input payload by triggering the entrypoints of the component, and collects the output payload accordingly.

with ComponentRunner(python_component, test_dir) as component_runner:
    output_payload = component_runner.run(input_payload)

If the component has parameters, you can set them by using the set_parameters() method. Similarly to the LocalPipelineRunner, you can use this method before calling run() on the component runner, but you cannot change pipeline parameters while these methods are running.