Digital Twins

An IOTICS Digital Twin is a virtual representation in IOTICS of a real entity. An entity can be a physical device, a person, a data source, a database, whatever is “real” for the domain.

📘

We often use Twin as an abbreviation for Digital Twin

On this page we will cover how to:

You can see the entire IOTICS API Reference here.


Overview

Take a look at our Digital Twin Introductory video to start with:

On a technical level Digital Twins consist of semantic metadata stored as RDF triples (properties of the twin and the object it represents) and streaming data feeds.

The separation of metadata and data is a core concept of IOTICS and makes Digital Twins machine-readable and machine-actionable. The use of semantic web linked-data patterns allows the twin to be a “pointer” to other sources of data, e.g. a web-accessible dataset.

Implementing Digital Twins first requires the creation of a Model of a Digital Twin. A Model is the “template” ie. the structure of the future twins. The model describes the properties and fields the twin should have, whereas the Digital Twins describe the instance of the asset, dataset or concept itself.

Digital Twin instances are then created based on a chosen Model. They contain the specific metadata and data of the modelled asset, concept or person.

Digital Twin Structure

An IOTICS Digital Twin is made of Five parts:

  1. Its IOTICS Identity
  2. Metadata Properties
  3. Relationship Properties
  4. Streaming Data Feeds
  5. Behaviour

Identity

Each twin has a self-sovereign identity conforming to the W3C “Decentralised ID”(DID) specification. The DID is the “subject” in the RDF Subject/Predicate/Object triples. Read more in our Identity section.

Metadata Property

The Metadata properties are used to:

  • describe the Digital Twin itself - access permissions, feed control, etc
  • describes the asset, data set or concept being represented - serial number, location, owner, etc
  • link or reference “outside” of IOTICSpace using URLs as the object of the triple
  • set the level of searchability

Relationship Property

The Relationship properties are used to describes the relationship between Digital Twins, for example "has", "relates to", etc. Semantic modelling is more flexible than relationships implemented in a hierarchical method. For example:

  • Engine A: belongs to Car A ← hierarchical relationship
  • Car A: relates to Bicycle B ← not hierarchical relationship
  • Car A: being driven by Person B ← temporary association
  • Car A: has 3 or 4 wheels ← relationship + condition - has to be related to either 3 or 4 wheels

Data Feeds

Digital Twins can publish or follow one or more data feeds and are used to:

  • contain real-time updates to the state of the twin - dataset has changed
  • contain one or more data fields, with each data field containing metadata about itself

Setting your headers

The headers are an important part of your API request, representing the meta-data of your request. Before submitting a request you need to ensure you have set your headers:

Header

Type

Description

Mandatory

Iotics-ClientRef

String

Any responses associated with the request will include this reference

Optional

Iotics-ClientAppId

String

User namespace used to group all the requests/responses

Mandatory

Iotics-TransactionRef

String

Used to loosely link requests/responses in a distributed environment each layer can add its own id to the list

Limited to:

  • max 16 elements per list
  • max 36 characters

Optional


Access permissions

When working with Digital Twins you can set an access level for complete control over who can access your data.

IOTICS distinguishes between two settings:

  1. Visibility: whether a Digital Twin's metadata is visible (or not) to another IOTICSpace (the F in FAIR)
    • this is set by updating the Host Twin's or Digital Twin's Visibility property
  2. Accessibility: whether a Digital twin's data can be accessed (or not) by a Digital Twin from another IOTICSpace (the A in FAIR)
    • this is set by updating the Host Twin's or Digital Twin's AllowList property
    • it can be enabled for the entire IOTICSpace and selectively on a twin-by-twin basis

For more information about access permissions, including for examples of how to update them, go to Selective Data Sharing.


Creating Twins

To create new Digital Twins you need to make a POST request to the IOTICS API's Twin service. Your request will need to provide the below payload parameters.

Body parameters

The payload in your request will be made up of:

Parameter

Type

Description

twinId

Object

TwinID is a unique twin identifier.

twinId:value

String

Twin Identifier (using DID format)

Example

🚧

Before running this example, make sure you have created your user credentials

If you need help on how to do this, have a look here.

# REQUIREMENTS:
# Create a python venv and activate it
# pip install -U pip setuptools wheel
# pip install iotics-identity
# USER_KEY_NAME, AGENT_KEY_NAME, USER_SEED, AGENT_SEED
​
from typing import List
from uuid import uuid4
​
from iotics.lib.identity.api.high_level_api import get_rest_high_level_identity_api
from requests import request
​
RESOLVER_URL = "resolver_url"
HOST = "host_url"
​
USER_KEY_NAME = "user_key_name"
AGENT_KEY_NAME = "agent_key_name"
USER_SEED = bytes.fromhex("user_seed")
AGENT_SEED = bytes.fromhex("agent_seed")
​
​
def make_api_call(method: str, url: str, json: dict):
    response = request(method=method, url=url, headers=headers, json=json)
    response.raise_for_status()
​
    return response.json()
​
​
def create_twin(twin_key_name: str):
    twin_registered_id = api.create_twin_with_control_delegation(
        twin_seed=AGENT_SEED,
        twin_key_name=twin_key_name,
        agent_registered_identity=agent_registered_id,
        delegation_name="#ControlDeleg",
    )
​
    make_api_call(
        method="POST",
        url=f"{HOST}/qapi/twins",
        json={"twinId": {"value": twin_registered_id.did}},
    )
​
    return twin_registered_id.did
​
​
def update_twin_with_metadata(
    twin_id: str,
    properties: List[dict] = None,
    location: dict = None,
    visibility: str = "PRIVATE",
):
    payload = {"newVisibility": {"visibility": visibility}}
​
    if location:
        payload["location"] = location
​
    if properties:
        payload["properties"] = {"added": []}
        for prop in properties:
            payload["properties"]["added"].append(prop)
​
    make_api_call(method="PATCH", url=f"{HOST}/qapi/twins/{twin_id}", json=payload)
​
​
def create_feed(twin_id: str, feed_id: str):
    make_api_call(
        method="POST",
        url=f"{HOST}/qapi/twins/{twin_id}/feeds",
        json={"feedId": {"value": feed_id}},
    )
​
​
def update_feed(
    twin_id: str,
    feed_id: str,
    metadata: List[dict],
    properties: List[dict] = None,
    cleared_all: bool = True,
    store_last=True,
):
    payload = {"storeLast": store_last, "values": {"added": metadata}}
​
    if properties:
        payload["properties"] = {"added": properties, "clearedAll": cleared_all}
​
    make_api_call(
        method="PATCH", url=f"{HOST}/qapi/twins/{twin_id}/feeds/{feed_id}", json=payload
    )
​
​
# Set Up API, token and headers
api = get_rest_high_level_identity_api(resolver_url=RESOLVER_URL)
(
    user_registered_id,
    agent_registered_id,
) = api.create_user_and_agent_with_auth_delegation(
    user_seed=USER_SEED,
    user_key_name=USER_KEY_NAME,
    agent_seed=AGENT_SEED,
    agent_key_name=AGENT_KEY_NAME,
    delegation_name="#AuthDeleg",
)
​
token = api.create_agent_auth_token(
    agent_registered_identity=agent_registered_id,
    user_did=user_registered_id.did,
    duration=30,
)
​
headers = {
    "accept": "application/json",
    "Iotics-ClientAppId": f"create_twin_{uuid4()}",
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json",
}
​
# Create twin
twin_did = create_twin(twin_key_name="#SensorTwin")
​
print("Twin created !!!")
print("TwinId:", twin_did)
​
# Add Metadata
update_twin_with_metadata(
    twin_id=twin_did,
    properties=[
        {
            "key": "http://www.w3.org/2000/01/rdf-schema#label",
            "langLiteralValue": {"value": "Temperature Sensor", "lang": "en"},
        },
        {
            "key": "http://www.w3.org/2000/01/rdf-schema#comment",
            "langLiteralValue": {
                "value": "A temperature sensor that shares temperature data",
                "lang": "en",
            },
        },
    ],
    location={"location": {"lat": 51.5, "lon": -0.1}},
)
​
print("Twin updated with metadata")
​
# Add Feed
create_feed(twin_id=twin_did, feed_id="currentTemp")
​
print("Feed created and added to the twin")
​
# Add Value
update_feed(
    twin_id=twin_did,
    feed_id="currentTemp",
    metadata=[
        {
            "comment": "Temperature in degrees Celsius",
            "dataType": "decimal",
            "label": "value",
            "unit": "http://purl.obolibrary.org/obo/UO_0000027",
        }
    ],
)
​
print("Value added to the Feed")

Describe Twins (view)

Once you have Twins in place you can start to view your Twins with a GET request to the Twin service.

Path parameters

The path tells the API which Twin you are accessing and is made up of:

Parameter

Type

Description

twinId

String

The unique ID of the Twin set during your POST request

Query parameters

Defines the language code for properties in the Twin.

Parameter

Type

Description

lang

String

Language code for properties

Example

This is an example of a GET request to searching for the Twin created in the POST example:

# REQUIREMENTS:
# Create a python venv and activate it
# pip install -U pip setuptools wheel
# pip install iotics-identity
# USER_KEY_NAME, AGENT_KEY_NAME, USER_SEED, AGENT_SEED

import json
from uuid import uuid4

from iotics.lib.identity.api.high_level_api import get_rest_high_level_identity_api
from requests import request

RESOLVER_URL = "resolver_url"
HOST = "host_url"

USER_KEY_NAME = "user_key_name"
AGENT_KEY_NAME = "agent_key_name"
USER_SEED = bytes.fromhex("user_seed")
AGENT_SEED = bytes.fromhex("agent_seed")


def describe_remote_twin(headers: dict, host_id: str, twin_id: str):
    resp = request(
        method="GET",
        url=f"{HOST}/qapi/hosts/{host_id}/twins/{twin_id}",
        headers=headers,
    )

    resp.raise_for_status()

    # Return twin description
    return json.loads(resp.text)


# Not used in this example
def describe_local_twin(headers: dict, twin_id: str):
    resp = request(method="GET", url=f"{HOST}/qapi/twins/{twin_id}", headers=headers)

    resp.raise_for_status()

    # Return twin description
    return json.loads(resp.text)


def create_headers(client_app_id: uuid4, token: str):
    return {
        "accept": "application/json",
        "Iotics-ClientAppId": f"describe_{client_app_id}",
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }


api = get_rest_high_level_identity_api(resolver_url=RESOLVER_URL)
(
    user_registered_id,
    agent_registered_id,
) = api.create_user_and_agent_with_auth_delegation(
    user_seed=USER_SEED,
    user_key_name=USER_KEY_NAME,
    agent_seed=AGENT_SEED,
    agent_key_name=AGENT_KEY_NAME,
    delegation_name="#AuthDeleg",
)

twin_description = describe_remote_twin(
    headers=create_headers(
        client_app_id=uuid4(),
        token=api.create_agent_auth_token(
            agent_registered_identity=agent_registered_id,
            user_did=user_registered_id.did,
            duration=30,
        ),
    ),
    host_id="16Uiu2HAm1yPT21xjhG3L5GZ8JyNgpmQPoHvdfQsWNhRXLymEBY61",
    twin_id="did:iotics:iotB5LGtcymXUFfs72K421R3Zbjaxd7jN72q",
)

print(json.dumps(twin_description, indent=4))

Updating Twins

To update and populate your Twins with data you need to send a PATCH request to the API. You just need to provide one or more of the below fields to be updated.

Path parameters

The path tells the API which Digital Twin you are accessing and is made up of:

Parameter

Type

Description

twinId

String

The unique ID of the Twin set during your POST request

Body parameters

The body of your request defines what you're updating. You can update these parameters:

🚧

Note

Changes are applied in this order:

  1. Visibility
  2. Properties
  3. Location

Location

Allows you to add or delete a locations on your Twin.

Parameter

Type

Description

location

Object

Used to add a location to your Twin

location:location

Object

The location being added

location:location:lat

Double

The latitude of the location being added

location:location:lon

Double

The longitude of the location being added

newVisibility

Sets the visibility of the Twin to Private or Public, as defined in access permissions.

Parameter

Type

Description

newVisibility

Object

Used to set the visibility of your Twin

newVisibility:Visibility

String

PRIVATE or PUBLIC

Properties

When you are updating the properties, you can(in order of operations):

  • Clear all properties
  • Delete a property
  • Delete properties by key
  • Add properties

A property is made up of one key and one value. The value can be one of:

  • langLiteralValue
  • literalValue
  • stringLiteralValue
  • uriValue

🚧

Note:

One choose one type of value for the property.

Parameter

Type

Description

properties

Object

Used to give your Twin properties

properties:added

Array of objects

The properties being added

properties:added:key

String

The key (predicate) of the property

properties:added:langLiteralValue

Object

Use a langLiteralValue to add a property

properties:added:langLiteralValue:lang

String

The 2 character language code for your property

properties:added:langLiteralValue:value

String

The value of the property you're adding

properties:added:literalValue

Object

Use a literalValue to add a property

properties:added:literalValue:dataType

String

XSD data type.
Currently supports:
dateTime, time, date, boolean, integer, nonPositiveInteger, negativeInteger, nonNegativeInteger, positiveInteger,
long, unsignedLong, int, unsignedInt, short, unsignedShort, byte, unsignedByte, anyURI

properties:added:literalValue:value

String

String representation of the value according to XSD datatype specification

properties:added:stringLiteralValue

Object

Use a stringLiteralValue to add a property

properties:added:stringLiteralValue:value

String

The value of the property you're adding

properties:added:uriValue

Object

Use a URI to add a property

properties:added:uriValue:value

String

The value of the property you're adding

properties:clearedAll

Boolean

Delete all properties currently set on the entity.

properties:deleted

Array of objects

Delete specific exact properties (by key and value). This operation is ignored if clearAll is True.

properties:deleted:key

String

The key (predicate) of the property

properties:deleted:langLiteralValue

Object

Use a langLiteralValue to delete a property

properties:deleted:langLiteralValue:lang

String

The 2 character language code for your property

properties:deleted:langLiteralValue:value

String

The value of the property you're deleting

properties:deleted:literalValue

Object

Use a literalValue to delete a property

properties:deleted:literalValue:dataType

String

XSD data type.
Currently supports:
dateTime, time, date, boolean, integer, nonPositiveInteger, negativeInteger, nonNegativeInteger, positiveInteger,
long, unsignedLong, int, unsignedInt, short, unsignedShort, byte, unsignedByte, anyURI

properties:deleted:literalValue:value

String

string representation of the value according to XSD datatype specification

properties:deleted:stringLiteralValue

Object

Use a stringLiteral to delete a property

properties:deleted:stringLiteralValue:value

String

The value of the property you're deleting

properties:deleted:uriValue

Object

Use a URI to delete a property

properties:deleted:uriValue:value

String

The value of the property you're adding

properties:deletedByKey

Array of strings

Delete any properties with the given keys (predicates). This operation is ignored if clearAll is True

Example

This is an example PATCH request updating the Twin created and found in the previous examples:

In this example we've added a new property and set the Twin's visibility to PUBLIC.

curl --request PATCH \
     --url https://example.iotics.space/qapi/twins/did%3Aiotics%3Adidexample1234abcd \
     --header 'Accept: application/json' \
     --header 'Content-Type: application/json' \
     --header 'Iotics-ClientAppId: ExampleAppID' \
     --data '
{
     "newVisibility": {
          "visibility": "PUBLIC"
     },
     "properties": {
          "added": [
               {
                    "langLiteralValue": {
                         "lang": "en",
                         "value": "Example Property Value"
                    },
                    "literalValue": {},
                    "stringLiteralValue": {},
                    "uriValue": {}
               }
          ]
     }
}
'
import requests

url = "https://example.iotics.space/qapi/twins/did%3Aiotics%3Adidexample1234abcd"

payload = {
    "comments": {"added": [
            {
                "lang": "en",
                "value": "Example comment"
            }
        ]},
    "newVisibility": {"visibility": "PUBLIC"},
    "properties": {"added": [
            {
                "langLiteralValue": {
                    "lang": "en",
                    "value": "Example Property Value"
                },
                "literalValue": {},
                "stringLiteralValue": {},
                "uriValue": {}
            }
        ]}
}
headers = {
    "Accept": "application/json",
    "Iotics-ClientAppId": "ExampleAppID",
    "Content-Type": "application/json"
}

response = requests.request("PATCH", url, json=payload, headers=headers)

print(response.text)
OkHttpClient client = new OkHttpClient();

MediaType mediaType = MediaType.parse("application/json");
RequestBody body = RequestBody.create(mediaType, "{\"comments\":{\"added\":[{\"lang\":\"en\",\"value\":\"example comment\"}]},\"newVisibility\":{\"visibility\":\"PUBLIC\"},\"properties\":{\"added\":[{\"langLiteralValue\":{\"lang\":\"en\",\"value\":\"Example Property Value\"},\"literalValue\":{},\"stringLiteralValue\":{},\"uriValue\":{}}]}}");
Request request = new Request.Builder()
  .url("https://example.iotics.space/qapi/twins/did%3Aiotics%3Adidexample1234abcd")
  .patch(body)
  .addHeader("Accept", "application/json")
  .addHeader("Iotics-ClientAppId", "ExampleAppID")
  .addHeader("Content-Type", "application/json")
  .build();

Response response = client.newCall(request).execute();

Upsert Twins

To populate your create and populate Twins at the same time you need to send a PUT request to the API. You just need to provide one or more of the below fields to be updated.

Body parameters

Feeds

Feeds to set metadata for Twin.

Parameter

Type

Description

feeds

Array of objects

Used to add a feed to your Twin

Location

Allows you to add or delete a locations on your Twin.

Parameter

Type

Description

location

Object

Used to add a location to your Twin

location:lat

Double

The latitude of the location being added

location:lon

Double

The longitude of the location being added

Properties

When you are updating the properties, you can(in order of operations):

  • Clear all properties
  • Delete a property
  • Delete properties by key
  • Add properties

A property is made up of one key and one value. The value can be one of:

  • langLiteralValue
  • literalValue
  • stringLiteralValue
  • uriValue

🚧

Note:

One choose one type of value for the property.

Parameter

Type

Description

properties

Array of objects

The properties being added

properties:key

String

The key (predicate) of the property

properties:langLiteralValue

Object

Use a langLiteralValue to add a property

properties:langLiteralValue:lang

String

The 2 character language code for your property

properties:added:langLiteralValue:value

String

The value of the property you're adding

properties:literalValue

Object

Use a literalValue to add a property

properties:literalValue:dataType

String

XSD data type.
Currently supports:
dateTime, time, date, boolean, integer, nonPositiveInteger, negativeInteger, nonNegativeInteger, positiveInteger,
long, unsignedLong, int, unsignedInt, short, unsignedShort, byte, unsignedByte, anyURI

properties:literalValue:value

String

String representation of the value according to XSD datatype specification

properties:stringLiteralValue

Object

Use a stringLiteralValue to add a property

properties:stringLiteralValue:value

String

The value of the property you're adding

properties:uriValue

Object

Use a URI to add a property

properties:uriValue:value

String

The value of the property you're adding

properties:clearedAll

Boolean

Delete all properties currently set on the entity.

TwinID

Used to set the unique Twin identifier.

Parameter

Type

Description

twinId

String

TwinID is a unique Twin identifier.

Visibility

Sets the visibility of the Twin to Private or Public, as defined in access permissions.

Parameter

Type

Description

newVisibility

Object

Used to set the visibility of your Twin

newVisibility:Visibility

String

PRIVATE or PUBLIC

Example

Please note that the output of this example is the same as the Create Twin example, however using the Upsert Twin makes the code more efficient.

# REQUIREMENTS:
# Create a python venv and activate it
# pip install -U pip setuptools wheel
# pip install iotics-identity
# USER_KEY_NAME, AGENT_KEY_NAME, USER_SEED, AGENT_SEED
​
from typing import List
from uuid import uuid4
​
from iotics.lib.identity.api.high_level_api import get_rest_high_level_identity_api
from requests import request
​
RESOLVER_URL = "resolver_url"
HOST = "host_url"
​
USER_KEY_NAME = "user_key_name"
AGENT_KEY_NAME = "agent_key_name"
USER_SEED = bytes.fromhex("user_seed")
AGENT_SEED = bytes.fromhex("agent_seed")
​
​
def make_api_call(method: str, url: str, json: dict):
    response = request(method=method, url=url, headers=headers, json=json)
    response.raise_for_status()
​
    return response.json()
​
​
def create_twin(twin_key_name: str):
    twin_registered_id = api.create_twin_with_control_delegation(
        twin_seed=AGENT_SEED,
        twin_key_name=twin_key_name,
        agent_registered_identity=agent_registered_id,
        delegation_name="#ControlDeleg",
    )
​
    make_api_call(
        method="POST",
        url=f"{HOST}/qapi/twins",
        json={"twinId": {"value": twin_registered_id.did}},
    )
​
    return twin_registered_id.did
​
​
def upsert_twin(
    twin_id: str,
    visibility: str = "PRIVATE",
    feeds: List[dict] = None,
    location: dict = None,
    properties: List[dict] = None,
):
    payload = {"twinId": twin_id, "visibility": visibility}
​
    if location:
        payload["location"] = location
    if feeds:
        payload["feeds"] = feeds
    if properties:
        payload["properties"] = properties
​
    make_api_call(method="PUT", url=f"{HOST}/qapi/twins", json=payload)
​
​
# Set Up API, token and headers
api = get_rest_high_level_identity_api(resolver_url=RESOLVER_URL)
(
    user_registered_id,
    agent_registered_id,
) = api.create_user_and_agent_with_auth_delegation(
    user_seed=USER_SEED,
    user_key_name=USER_KEY_NAME,
    agent_seed=AGENT_SEED,
    agent_key_name=AGENT_KEY_NAME,
    delegation_name="#AuthDeleg",
)
​
token = api.create_agent_auth_token(
    agent_registered_identity=agent_registered_id,
    user_did=user_registered_id.did,
    duration=30,
)
​
headers = {
    "accept": "application/json",
    "Iotics-ClientAppId": f"create_twin_{uuid4()}",
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json",
}
​
# Create twin
twin_did = create_twin(twin_key_name="#SensorTwin")
​
print("Twin created !!!")
print("TwinId:", twin_did)
​
# Add Metadata, Feed and Value
upsert_twin(
    twin_id=twin_did,
    properties=[
        {
            "key": "http://www.w3.org/2000/01/rdf-schema#label",
            "langLiteralValue": {"value": "Temperature Sensor", "lang": "en"},
        },
        {
            "key": "http://www.w3.org/2000/01/rdf-schema#comment",
            "langLiteralValue": {
                "value": "A temperature sensor that shares temperature data",
                "lang": "en",
            },
        },
    ],
    feeds=[
        {
            "id": "currentTemp",
            "storeLast": True,
            "properties": [
                {
                    "key": "http://www.w3.org/2000/01/rdf-schema#label",
                    "langLiteralValue": {"value": "currentTemp", "lang": "en"},
                }
            ],
            "values": [
                {
                    "comment": "Temperature in degrees Celsius",
                    "dataType": "decimal",
                    "label": "value",
                    "unit": "http://purl.obolibrary.org/obo/UO_0000027",
                }
            ],
        }
    ],
    location={"lat": 51.5, "lon": -0.1},
)
​
print("Metadata, Feed and Value added to the twin")

Follow Twins

Using the Interest Service you can follow desired Twins that you can access.

Path parameters

Parameter

Type

Description

followerTwinId

String

The ID of the twin that is following

hostId

String

The ID of the host

followedTwinId

String

The ID of the twin that is being followed

followedFeedId

String

The ID of the feed that is being followed

Example

An example request to follow Twins.

# REQUIREMENTS:
# Create a python venv and activate it
# pip install -U pip setuptools wheel
# pip install iotics-identity shortuuid
# pip install deps/iotic.web.stomp-1.0.6.tar.gz (if the user has been provided with the iotics-host-lib. Need to clarify this)
# USER_KEY_NAME, AGENT_KEY_NAME, USER_SEED, AGENT_SEED, QAPI_STOMP_URL

import base64
import json
from time import sleep
from uuid import uuid4

import shortuuid
from iotic.web.stomp.client import StompWSConnection12
from iotics.lib.identity.api.high_level_api import get_rest_high_level_identity_api
from stomp import ConnectionListener

RESOLVER_URL = "resolver_url"
HOST = "host_url"

USER_KEY_NAME = "user_key_name"
AGENT_KEY_NAME = "agent_key_name"
USER_SEED = bytes.fromhex("user_seed")
AGENT_SEED = bytes.fromhex("agent_seed")
QAPI_STOMP_URL = "qapi_stomp_url"


class StompHandler:
    def __init__(self, endpoint, follow_callback, token_duration=10):
        self._endpoint = endpoint
        self._stomp_listener = StompListener(
            follow_callback=follow_callback,
            disconnect_handler=self.disconnect_handler,
        )
        self._token_duration = token_duration
        self._headers = None
        self._stomp_client = None
        self._subscriptions = set()

    def setup(self):
        token = api.create_agent_auth_token(
            agent_registered_identity=agent_registered_id,
            user_did=user_registered_id.did,
            duration=self._token_duration,
        )
        self._headers = {
            "accept": "application/json",
            "Iotics-ClientAppId": f"follow_{uuid4()}",
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        }

        self._stomp_client = StompWSConnection12(endpoint=self._endpoint)
        self._stomp_client.set_listener("stomp_listener", self._stomp_listener)
        self._stomp_client.connect(wait=True, passcode=token)
        self._resubscribe_all()

    def _resubscribe_all(self):
        for destination, subscription_id in self._subscriptions:
            self.subscribe(destination=destination, subscription_id=subscription_id)

    def subscribe(self, destination, subscription_id):
        self._stomp_client.subscribe(
            destination=destination, id=subscription_id, headers=self._headers
        )
        self._subscriptions.add((destination, subscription_id))

    def disconnect_handler(self):
        self.disconnect()

        # Generate a new token
        if self._stomp_listener.regenerate_token:
            self._stomp_listener.regenerate_token = False
            self.setup()

    def disconnect(self):
        self._stomp_client.remove_listener("stomp_listener")
        self._stomp_client.disconnect()


class StompListener(ConnectionListener):
    def __init__(self, follow_callback, disconnect_handler):
        self._follow_callback = follow_callback
        self._disconnect_handler = disconnect_handler
        self.regenerate_token = False

    def on_error(self, headers, body):
        payload = json.loads(body)

        if "token expired" in payload["message"]:
            self.regenerate_token = True
        else:
            print("Received an error:", payload)

    def on_message(self, headers, body):
        self._follow_callback(headers, body)

    def on_disconnected(self):
        sleep(1)
        try:
            self._disconnect_handler()
        except Exception as ex:
            print("An exception is raised", ex)


def subscribe_to_remote_feed(
    follower_twin_id, followed_twin_id, remote_host_id, followed_feed_name, callback
):
    stomp_handler = StompHandler(endpoint=QAPI_STOMP_URL, follow_callback=callback)
    stomp_handler.setup()
    stomp_handler.subscribe(
        destination=f"/qapi/twins/{follower_twin_id}/interests/hosts/{remote_host_id}/twins/{followed_twin_id}/feeds/{followed_feed_name}",
        subscription_id=f"d-poc-{shortuuid.random(8)}",
    )


# Not used in this example
def subscribe_to_local_feed(
    follower_twin_id, followed_twin_id, followed_feed_name, callback
):
    stomp_handler = StompHandler(endpoint=QAPI_STOMP_URL, follow_callback=callback)
    stomp_handler.setup()
    stomp_handler.subscribe(
        destination=f"/qapi/twins/{follower_twin_id}/interests/twins/{followed_twin_id}/feeds/{followed_feed_name}",
        subscription_id=f"d-poc-{shortuuid.random(8)}",
    )


def follow_callback(headers, body):
    payload = json.loads(body)
    time = payload["feedData"]["occurredAt"]
    data = payload["feedData"]["data"]

    feed_data = json.loads(base64.b64decode(data).decode("ascii"))
    print(f"New message occurred at {time}: {feed_data}")


api = get_rest_high_level_identity_api(resolver_url=RESOLVER_URL)
(
    user_registered_id,
    agent_registered_id,
) = api.create_user_and_agent_with_auth_delegation(
    user_seed=USER_SEED,
    user_key_name=USER_KEY_NAME,
    agent_seed=AGENT_SEED,
    agent_key_name=AGENT_KEY_NAME,
    delegation_name="#AuthDeleg",
)

subscribe_to_remote_feed(
    follower_twin_id="did:iotics:iotFuMFofrVDazerTBdGRCmdJvP3AGnyHtbU",
    remote_host_id="16Uiu2HAmFJHXoVtNhubHmjPXe1mhF9zvkd8rgKoFeRjbMVoyw3if",
    followed_twin_id="did:iotics:iotFuMFofrVDazerTBdGRCmdJvP3AGnyHtbU",
    followed_feed_name="bike-occupancy",
    callback=follow_callback,
)

while True:
    sleep(5)

Did this page help you?