transitional-sync-middleware

A transitional fork of aas-middleware by Sebastian Behrendt. Nobody should build on it.

This exists so that kapps_semantic_middleware has a public dependency chain. It is sdm4fzi/aas_middleware taken from branch rest_generalization at commit 14aa60a (2025-09-11), plus four correctness fixes to its sync and MQTT layers and one manifest correction, all made while building against it. Those fixes are the whole reason this is here: none of them is in any released aas-middleware, and the last release, 0.2.6, predates them. They are listed in CHANGELOG.md.

It is transitional, and it is going away. The sync layer this package is named for is being absorbed into kapps_semantic_middleware over the next few releases, and this distribution is retired when that lands. It is published Development Status :: 7 - Inactive from its first release for exactly that reason. If you want the middleware itself rather than a stopgap, aas-middleware is Sebastian Behrendt's own distribution and the root of this one.

The box still holds more than the label says. The code below ships verbatim, so it carries the asset administration shell client, the GraphQL API and the formatting layer -- none of which kapps_semantic_middleware uses. Stripping the package down to the sync core is a later release's work. Everything documented from here on therefore describes aas-middleware as Sebastian Behrendt wrote it, under a different name.

transitional-sync-middleware has the goal to make information flow and orchestration in industrial environments easier and more automated. To do so, it is build upon three concepts: well defined data models, connectors, and, workflows. The data models are used to define the structure of the data that is exchanged between different systems. The connectors are used to connect the data models to technologically different data sources and sinks. Workflows are used to define the orchestration of the data flow between different systems, connectors, and, data models.

transitional-sync-middleware uses modern api technologies (Rest or GraphQL) to make it easy to access the integrated data models or exchange data with low-level sensors and actuators. The middleware can be used to integrate different systems in the industrial environment, such as MES, ERP, SCADA, PLC, sensors, actuators and asset administration shells. By the modular and extensible design of transitional-sync-middleware, it can be used from small use cases, such as streaming sensor values, to large integrations and workflows for automated production planning and control.

For more information how to use the transitional-sync-middleware, read the getting started below or refer to the documentation of the package.

Installation

To install the package, run the following command in the terminal:

pip install transitional-sync-middleware

To install the package with extras for industrial data integration (OPC UA and MQTT), run the following command in the terminal:

pip install transitional-sync-middleware[industrial]

Alternatively, you can install the package with poetry for development:

poetry shell
poetry install

Please note that the package is only compatible with Python 3.10 or higher.

Getting Started

In the following, we will consider a minimal example to demonstrate the usage of the package. The example is also available in the examples of the repository and consists of defining a simple asset administration shell based data model to describe a product, serializing this data model to asset administration shells, making the data model available with rest API and defining examplary connectors and workflows.

For further examples, like connecting the data model to a data source, or creating workflows, refer to the tutorials in the documentation.

Defining a simple data model and formatting

At first, we create a simple data model with the basic building blocks of the aas meta model:

import typing
import transitional_sync_middleware

class BillOfMaterialInfo(transitional_sync_middleware.SubmodelElementCollection):
    manufacterer: str
    product_type: str

class BillOfMaterial(transitional_sync_middleware.Submodel):
    components: typing.List[str]
    bill_of_material_info: BillOfMaterialInfo


class ProcessModel(transitional_sync_middleware.Submodel):
    processes: typing.List[str]


class Product(transitional_sync_middleware.AAS):
    bill_of_material: BillOfMaterial
    process_model: typing.Optional[ProcessModel]

The data model consists of a product that has a process model and a bill of material. The process model and the bill of material are contain a list of processes and components, respectively. Note, that the transitional-sync-middleware can also use data models that are not compliant with the aas meta model. However, some features, such as storing the data model in a basyx aas server, requires that the data model can be translated to the aas meta model.

To be able to instantiate an data model, we create an instance of the Product:

example_product = Product(
    id="example_product_id",
    id_short="example_product_id",
    description="Example Product",
    bill_of_material=BillOfMaterial(
        id="example_bom_id",
        id_short="example_bom_id",
        description="Example Bill of Material",
        components=["component_1", "component_2"],
        bill_of_material_info=BillOfMaterialInfo(
            id="example_bom_info_id",
            id_short="example_bom_info_id",
            description="Example Bill of Material Info",
            manufacterer="Example Manufacterer",
            product_type="Example Product Type",
        ),
    ),
    process_model=ProcessModel(
        id="example_process_model_id",
        id_short="example_process_model_id",
        description="Example Process Model",
        processes=["process_1", "process_2"],
    ),
)

With this instance of the product, we can create a DataModel.

data_model = transitional_sync_middleware.DataModel.from_models(example_product)

The data model is a container for instances and types of data models. It makes access to individual objects easy and allows for formatting. E.g. we can easily transform the data model to either aas components of the basyx python sdk:

basyx_object_store = transitional_sync_middleware.formatting.BasyxFormatter().serialize(data_model)

Or serialize it to a JSON-serialized asset administration shell according to the official specification of the asset administration shell :

formatter = transitional_sync_middleware.formatting.aasJsonFormatter()
json_aas = formatter.serialize(data_model)
print(json_aas)

This formatting transformation can also be reversed, so JSON-serialized or Basyx aas can be used to create data models.

Starting the transitional-sync-middleware with internal storage and aas persistence

To start the transitional-sync-middleware and make our data model available through a rest API, we need to create an instance of the middleware and load the data model:

middleware = transitional_sync_middleware.Middleware()
middleware.load_data_model("example", data_model, persist_instances=True)
middleware.generate_rest_api_for_data_model("example")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(middleware.app)

With the option persist_instances=True, the middleware stores the instances of the data model in its internal memory. Moreover, a CRUD (create, read, update, delete) rest API is generated for the data model with the generate_rest_api_for_data_model function.

Under the hood, the transitional-sync-middleware uses the fastapi framework to generate the rest API. Therefore, we get an openAPI specification of this API under http://localhost:8000/docs. The data model can be accessed via the API with the following curl command:

curl -X 'GET' \
  'http://127.0.0.1:8000/Product/' \
  -H 'accept: application/json'

We can also instantiate new product data models with post requests:

curl -X 'POST' \
  'http://127.0.0.1:8000/Product/' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id_short": "product2",
  "description": "",
  "id": "product2",
  "bill_of_material": {
    "id_short": "product2_bom",
    "description": "",
    "id": "product2_bom",
    "semantic_id": "",
    "components": [
      "bearing", "screw"
    ],
    "bill_of_material_info": {
      "id_short": "product2_bom_info",
      "description": "",
      "semantic_id": "",
      "manufacterer": "other manufacturer",
      "product_type": "wheel"
    }
  },
  "process_model": {
    "id_short": "product2_process_model",
    "description": "",
    "id": "product2_process_model",
    "semantic_id": "",
    "processes": [
      "assembly"
    ]
  }
}'

Besides a rest API, transitional-sync-middleware also provides a GraphQL API. To make a data model available via the GraphQL API, use the generate_graphql_api_for_data_model function.

middleware.generate_graphql_api_for_data_model("example")

The GraphQL API can be accessed under http://localhost:8000/graphql, where also find the GraphiQL Playground for testing. Some fields of the instances of the product data model can be accessed via the API with the following curl command:

curl 'http://127.0.0.1:8000/graphql/?' \
  -H 'Accept: application/json' \
  -H 'Content-Type: application/json' \
  --data-raw '{"query":"{\n  Product {\n    idShort\n    billOfMaterial {\n      components\n    }\n  }\n}"}'

You probably saw during the startup of the middleware a warning log message like:

WARNING:transitional_sync_middleware.middleware.registries:No persistence factory found for data_model_name='example' model_id=None contained_model_id=None field_id=None. Using default persistence factory.

This message indicates that the transitional-sync-middleware has no specified persistence factory and stores the data model in internal storage. Besides internal storage, the transitional-sync-middleware can utilize different persistence mechanisms. For now, transitional-sync-middleware supports the storage of the data model in a basyx aas server (mongo db will come later).

Running the middleware with persistent storage of the data model in a basyx aas server can be done by interchanging the default Middleware with the AasMiddleware:

middleware = transitional_sync_middleware.AasMiddleware()
middleware.load_aas_persistent_data_model(
    "example", data_model, "localhost", 8081, "localhost", 8081, persist_instances=True
)
middleware.generate_rest_api_for_data_model("example")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(middleware.app)

Note, that on port 8081 a aas repository and submodel repository need to be running, e.g. a basyx java server. The repository already comes with a docker-compose file that can be used to start the AAS and Submodel-server. To start the servers with the linked compose file, run the following command in the terminal:

docker-compose -f docker-compose-dev.yaml up

The created rest API can be accessed in the same way as before.

Creating an examplary connector and workflow

A connector in the transitional-sync-middleware is a component that makes a connector to a data source (provider) or a data sink (consumer). The transitional-sync-middleware comes with a lot of default connectors (OPC UA, MQTT, http client and server, websocket client and server, webhook), but you can add custom ones. A connector only needs to specify a certain interface, that consists of a connect, disconnect, consume and provide function. We define an examplary connector like this and add it to the middleware:

class TrivialConnector:
    def __init__(self):
        pass

    async def connect(self):
        pass

    async def disconnect(self):
        pass

    async def consume(self, body: str) -> None:
        print(body)
        pass

    async def provide(self) -> typing.Any:
        return "trivial connector example value"
    
example_connector = TrivialConnector()
middleware.add_connector("test_connector", example_connector, model_type=str)

We can execute this connector either for reading values with the folling get request:

curl -X 'GET' \
  'http://127.0.0.1:8000/connectors/test_connector/value' \
  -H 'accept: application/json'

Or send data with the connector to the data sink:

curl -X 'POST' \
  'http://127.0.0.1:8000/connectors/test_connector/value?value=connector_example_input_value' \
  -H 'accept: application/json' \
  -d ''

Lastly, we can define workflows for more complex automation tasks. A workflow is a function that can be executed via the API the middleware:

@middleware.workflow()
def example_workflow(a: str) -> str:
    print(a)
    return a

To run the workflow, execute the following request:

curl -X 'POST' \
  'http://127.0.0.1:8000/workflows/example_workflow/execute?arg=work_flow_input_value' \
  -H 'accept: application/json' \
  -d ''

We can also interrupt the workflow, when it is running:

curl -X 'GET' \
  'http://127.0.0.1:8000/workflows/example_workflow/interrupt' \
  -H 'accept: application/json'

The contents above only demostrate the most fundamental features of the transitional-sync-middleware. For more advanced features, like connecting this fundametals for automated information flow in a production environment, refer to the documentation.

Contributing

Please do not send contributions here. This fork is retired within a few releases and nothing merged into it survives.

Improvements to the middleware itself belong upstream, with Sebastian Behrendt at sdm4fzi/aas_middleware.

License

The package is licensed under the MIT license.

Acknowledgements

We extend our sincere thanks to the German Federal Ministry for Economic Affairs and Climate Action (BMWK) for supporting this research project 13IK001ZF “Software-Defined Manufacturing for the automotive and supplying industry https://www.sdm4fzi.de/.