Connectors

Ingesting or exporting data from an IOTICSpace and its Digital Twins requires the creation of a software application called "Connector" through the IOTICS API.

This page covers:


Introduction to Connectors

A Connector is a software application that interacts with Digital Twins in an IOTICSpace. It acts as the bridge between an IOTICSpace and the data source.

We distinguish between three different types:

1. Publisher Connector: imports data into an IOTICSpace
2. Follower Connector: exports data from an IOTICSpace
3. Synthesiser Connector: transforms existing data within an IOTICSpace

Publisher Connector

A Publisher Connector is a software application that works as a bridge between a data source and an IOTICSpace. It enables external data to be imported into an IOTICSpace and published by one or more Digital Twins, and therefore shared with and used by other Digital Twins in the ecosystem.

A Publisher Connector:

  • typically handles one single source of data, and is made up of one Twin Model with one or more related Twins from Model. If multiple sources have to be imported and synchronised, several Twin Models may have to be created, each with their specific metadata;
  • periodically imports and publishes new data through the generic Digital Twins. The data can be published either in batches or in single samples, and at fixed or variable intervals. Some Twins may also need to share data with a different frequency than others;
  • runs continuously. Once created, it is good practice to allow all the Twins in the IOTICSpace to maintain their state of consistency - be it actively sharing new data, stay idle or become obsolete and be deleted automatically.
Representation of a Publisher ConnectorRepresentation of a Publisher Connector

Representation of a Publisher Connector

How to create a Publisher Connector with the IOTICS API

Implementing a Publisher Connector leads to the creation of a set of Twin Publishers that, simultaneously or in turn, publish data. A prerequisite is the development of one or more Twin Models that will serve as a template for the Twin Publishers.

The following example shows a Publisher Connector that simulates 10 temperature sensors. Each gets data from the environment and shares sensor readings every 10 seconds. We first create a Twin Model with a Feed, and then create the 10 Sensor Twins, who periodically download new data from a Server and share it.

Click to see the example code of a Publisher Connector

1. Get a fresh token to be used in the requests' headers;
2. Create a Twin Model;
  2.1 Build list of Properties;
  2.2 Build list of Feeds;
  2.3 Get a new Twin DID;
  2.4 Use the Upsert Twin API;
3. Create Twins from Model;
  3.1 Use the Describe Local Twin API to get the Twin Model's metadata;
  3.2 Build list of Properties;
  3.3 Use the Describe Feed API to get the the Twin Model Feed's metadata;
  3.4 Build list of Feeds;
  For every Sensor:
  3.5 Get a new Twin DID;
  3.6 Use the Upsert Twin API;
Every 10 seconds:
4. Download new data from the server;
5. Share data into the Space;
import base64
import json
from datetime import datetime, timezone
from time import sleep
from typing import List

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")


class PublisherConnector:
    def __init__(self):
        self._api = None
        self._user_registered_id = None
        self._agent_registered_id = None
        self._headers = None
        self._twins_info = None

    def setup(self):
        self._api = get_rest_high_level_identity_api(resolver_url=RESOLVER_URL)
        (
            self._user_registered_id,
            self._agent_registered_id,
        ) = self._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",
        )

        self._headers = {
            "accept": "application/json",
            "Iotics-ClientAppId": "publisher_connector",
            "Content-Type": "application/json",
        }
        self._refresh_token()
        self._twins_info = {}

    def _refresh_token(self, duration: int = 60):
        print("Refreshing token")
        print("---")
        token = self._api.create_agent_auth_token(
            agent_registered_identity=self._agent_registered_id,
            user_did=self._user_registered_id.did,
            duration=duration,
        )

        self._headers["Authorization"] = f"Bearer {token}"

    def _get_new_did(self, twin_key_name: str):
        twin_registered_id = self._api.create_twin_with_control_delegation(
            twin_seed=AGENT_SEED,
            twin_key_name=twin_key_name,
            agent_registered_identity=self._agent_registered_id,
            delegation_name="#ControlDeleg",
        )

        return twin_registered_id.did

    def _make_api_call(
        self, method: str, url: str, json: dict = None, retry: bool = True
    ):
        try:
            response = request(method=method, url=url, headers=self._headers, json=json)
            response.raise_for_status()
        except Exception as ex:
            if retry:
                self._refresh_token()
                return self._make_api_call(method, url, json, retry=False)
            else:
                print("Retried once. Still getting error", ex)

        return response.json()

    def _create_new_twin(
        self,
        twin_key_name: str,
        properties: List[str] = None,
        feeds: List[str] = None,
        visibility: str = "PRIVATE",
        location: dict = None,
    ):
        twin_did = self._get_new_did(twin_key_name)

        self._upsert_twin(
            twin_id=twin_did,
            properties=properties,
            feeds=feeds,
            visibility=visibility,
            location=location,
        )

        return twin_did

    def _describe_feed(self, twin_id: str, feed_id: str):
        feed_description = self._make_api_call(
            method="GET",
            url=f"{HOST}/qapi/twins/{twin_id}/feeds/{feed_id}",
        )

        return feed_description

    def _describe_local_twin(self, twin_id: str):
        twin_description = self._make_api_call(
            method="GET", url=f"{HOST}/qapi/twins/{twin_id}"
        )

        return twin_description

    def _upsert_twin(
        self,
        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

        self._make_api_call(method="PUT", url=f"{HOST}/qapi/twins", json=payload)

        print(f"Twin {twin_id} created succesfully with Metadata, Feeds and Values")
        print("---")

    def _share_data(self, twin_label: str, twin_id: str, feed_id: str, value):
        data_to_share = {"reading": value}
        encoded_data = base64.b64encode(json.dumps(data_to_share).encode()).decode()
        data_to_share_payload = {
            "sample": {
                "data": encoded_data,
                "mime": "application/json",
                "occurredAt": datetime.now(tz=timezone.utc).isoformat(
                    timespec="seconds"
                ),
            }
        }
        self._make_api_call(
            method="POST",
            url=f"{HOST}/qapi/twins/{twin_id}/feeds/{feed_id}/shares",
            json=data_to_share_payload,
        )

        print(f"Shared {data_to_share} from Twin {twin_label}")

    def create_twin_model(self):
        print("Creating Twin Model")
        # Build a list of Properties for the Twin Model
        property_list = [
            # Twin Model "special" property
            {
                "key": "http://www.w3.org/1999/02/22-rdf-syntax-ns#type",
                "uriValue": {"value": "https://data.iotics.com/app#Model"},
            },
            # Add Label
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#label",
                "langLiteralValue": {
                    "value": "Temperature Sensor Publisher Model",
                    "lang": "en",
                },
            },
            # Add Comment
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#comment",
                "langLiteralValue": {
                    "value": "A temperature sensor that shares temperature data",
                    "lang": "en",
                },
            },
            # Add colour
            {
                "key": "https://data.iotics.com/app#color",
                "stringLiteralValue": {"value": "#9aceff"},
            },
            # Add Space name
            {
                "key": "https://data.iotics.com/app#spaceName",
                "stringLiteralValue": {"value": "my-space"},
            },
            # Add CreatedAt property
            {
                "key": "https://data.iotics.com/app#createdAt",
                "literalValue": {
                    "dataType": "dateTime",
                    "value": datetime.now(tz=timezone.utc).isoformat(),
                },
            },
            # Add UpdatedAt property
            {
                "key": "https://data.iotics.com/app#updatedAt",
                "literalValue": {
                    "dataType": "dateTime",
                    "value": datetime.now(tz=timezone.utc).isoformat(),
                },
            },
        ]

        # Build a list of Feeds for the Twin Model
        feed_list = [
            # Add a single Feed called "currentTemp"
            {
                "id": "currentTemp",
                "storeLast": True,  # We want to store the last value sent
                "properties": [
                    # Add a Label for the Feed
                    {
                        "key": "http://www.w3.org/2000/01/rdf-schema#label",
                        "langLiteralValue": {
                            "value": "Temperature",
                            "lang": "en",
                        },
                    }
                ],
                # Add a single Value called "reading"
                "values": [
                    {
                        "comment": "Temperature in degrees Celsius",
                        "dataType": "decimal",
                        "label": "reading",
                        "unit": "http://qudt.org/vocab/unit/DEG_C",
                    }
                ],
            },
        ]

        model_twin_did = self._create_new_twin(
            twin_key_name="TwinPublisherModel",
            properties=property_list,
            feeds=feed_list,
        )

        return model_twin_did

    def create_twins_from_model(self, twin_model_id: str):
        twin_model = self._describe_local_twin(twin_id=twin_model_id)

        # Build a list of properties for the generic Twin
        property_list = [
            # Add the generic twins' "special" properties
            {
                "key": "https://data.iotics.com/app#model",
                "uriValue": {"value": twin_model_id},
            },
            {
                "key": "https://data.iotics.com/app#createdFrom",
                "uriValue": {"value": "https://data.iotics.com/app#ByModel"},
            },
        ]

        # Add to the list all the Twin Model's properties but consider the exceptions
        for prop in twin_model["result"]["properties"]:
            # This was the "special" property of the Twin Model, we don't need it here
            if prop["key"] == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type":
                continue
            # Don't add the Label property of the Twin Model.
            # It'll be added later based on the number of the sensor
            if prop["key"] == "http://www.w3.org/2000/01/rdf-schema#label":
                continue
            # Change the Created At value
            if prop["key"] == "https://data.iotics.com/app#createdAt":
                prop["literalValue"]["value"] = datetime.now(
                    tz=timezone.utc
                ).isoformat()
            # Change the Updated At value
            if prop["key"] == "https://data.iotics.com/app#updatedAt":
                prop["literalValue"]["value"] = datetime.now(
                    tz=timezone.utc
                ).isoformat()

            property_list.append(prop)

        # Build a list of Feeds for the Twin Model
        feed_list = []

        # Scan the Twin Model's feeds list to get the Twin Model's Feed IDs.
        # Then use the Feed Describe API to get the Feeds' metadata
        for feed in twin_model["result"]["feeds"]:
            feed_id = feed["feedId"]["value"]

            # Use the Feed Describe API
            feed_description = self._describe_feed(
                twin_id=twin_model_id, feed_id=feed_id
            )
            feed_properties = feed_description["result"]["properties"]
            feed_values = feed_description["result"]["values"]
            store_last = feed_description["result"]["storeLast"]

            # Append to the list all the Feed's metadata
            feed_list.append(
                {
                    "id": feed_id,
                    "properties": feed_properties,
                    "values": feed_values,
                    "storeLast": store_last,
                }
            )

        # Finally create the 10 Publisher Twins
        for sensor_number in range(1, 11):
            twin_key_name = f"temp_sensor_{sensor_number}"
            twin_label = f"Temperature Sensor Publisher {sensor_number}"

            # Add the Label property according to the number of the Sensor
            property_list_with_label = property_list.copy()
            property_list_with_label.append(
                {
                    "key": "http://www.w3.org/2000/01/rdf-schema#label",
                    "langLiteralValue": {"value": twin_label, "lang": "en"},
                }
            )

            print(f"Creating Twin from Model - {twin_label}")

            twin_from_model_did = self._create_new_twin(
                twin_key_name=twin_key_name,
                properties=property_list_with_label,
                feeds=feed_list,
                location={"lat": 51.5, "lon": -0.1},
            )

            # Add the Twin's info in the mapping dictionary
            twin_info = {
                twin_key_name: {
                    "twin_did": twin_from_model_did,
                    "twin_label": twin_label,
                }
            }
            self._twins_info.update(twin_info)

    def publish_data(self, data: dict):
        for temp_data in data:
            temperature = temp_data["temp"]
            sensor_id = temp_data["sensor_id"]

            self._share_data(
                twin_label=self._twins_info[sensor_id]["twin_label"],
                twin_id=self._twins_info[sensor_id]["twin_did"],
                feed_id="currentTemp",
                value=temperature,
            )

    def _get_twin_label(self, property_list: List[dict]):
        for prop in property_list:
            if prop["key"] == "http://www.w3.org/2000/01/rdf-schema#label":
                return prop["langLiteralValue"]["value"]

        return None


def get_sensor_data():
    response = request(method="GET", url="http://flaskapi.dev.iotics.com/sensor_temp")
    sensor_data = response.json()

    return sensor_data


def main():
    publisher_connector = PublisherConnector()
    publisher_connector.setup()
    model_twin_did = publisher_connector.create_twin_model()
    publisher_connector.create_twins_from_model(twin_model_id=model_twin_did)

    while True:
        data = get_sensor_data()
        publisher_connector.publish_data(data)
        sleep(10)
        print("---")


if __name__ == "__main__":
    main()

Follower Connector

The mirrored version of the Publisher Connector is the Follower Connector. It is an application that works as a bridge between an IOTICSpace and the data consumer and/or data storage.

🚧

A Follower Connector can run in the same or a different IOTICSpace than the Twin Publishers

Use remote search and remote follow if the IOTICSpaces are not the same.

Representation of a Follower Connector in the same Space as the Twin PublishersRepresentation of a Follower Connector in the same Space as the Twin Publishers

Representation of a Follower Connector in the same Space as the Twin Publishers

How to create a Follower Connector with the IOTICS API

Similar to the Publisher Connector, a set of Twin Followers are needed in order to ask the Twin Publishers' feed for the last shared value. One or more Twin Models will serve as a template for the Twin Followers so they can be classified according to their specific nature.

The following example shows how to set up a Follower Connector. It automatically creates a Twin Model and a generic Twin that follows the Sensor Twins' feed of the Publisher Connector, as demonstrated above, and asks for their last shared value every 10 seconds.

Click to see an example of a Follower Connector

Instructions:

  1. Run the publisher_connector.py in a terminal;
  2. Run the follower_connector.py code below in another terminal and in the same IOTICSpace as the Publisher Connector.
  3. Look at the data received on your terminal;
1. Get a fresh token to be used in the requests' headers;
2. Create a Twin Model;
  2.1 Build list of Properties;
  2.2 Get a new Twin DID;
  2.3 Use the Upsert Twin API;
3. Create a Twin from Model;
  3.1 Use the Describe Local Twin API to get the Twin Model's metadata;
  3.2 Build list of Properties;
  3.3 Get a new Twin DID;
  3.4 Use the Upsert Twin API;
4. Search for Twin Publishers;
Every 10 seconds:
5. Get the Twin Publishers' last shared data.
import base64
import json
from datetime import datetime, timezone, timedelta
from time import sleep
from typing import List

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")


class FollowerConnector:
    def __init__(self):
        self._api = None
        self._user_registered_id = None
        self._agent_registered_id = None
        self._headers = None

    def setup(self):
        self._api = get_rest_high_level_identity_api(resolver_url=RESOLVER_URL)
        (
            self._user_registered_id,
            self._agent_registered_id,
        ) = self._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",
        )

        self._headers = {
            "accept": "application/json",
            "Iotics-ClientAppId": "follower_connector",
            "Content-Type": "application/json",
        }
        self._refresh_token()

    def _refresh_token(self, duration: int = 60):
        print("Refreshing token")
        print("---")
        token = self._api.create_agent_auth_token(
            agent_registered_identity=self._agent_registered_id,
            user_did=self._user_registered_id.did,
            duration=duration,
        )

        self._headers["Authorization"] = f"Bearer {token}"

    def _get_new_did(self, twin_key_name: str):
        twin_registered_id = self._api.create_twin_with_control_delegation(
            twin_seed=AGENT_SEED,
            twin_key_name=twin_key_name,
            agent_registered_identity=self._agent_registered_id,
            delegation_name="#ControlDeleg",
        )

        return twin_registered_id.did

    def _make_api_call(
        self, method: str, url: str, json: dict = None, retry: bool = True
    ):
        try:
            response = request(method=method, url=url, headers=self._headers, json=json)
            response.raise_for_status()
        except Exception as ex:
            if retry:
                self._refresh_token()
                return self._make_api_call(method, url, json, retry=False)

            print("Retried once. Still getting error", ex)

        return response.json()

    def _create_new_twin(
        self,
        twin_key_name: str,
        properties: List[str] = None,
        feeds: List[str] = None,
        visibility: str = "PRIVATE",
        location: dict = None,
    ):
        twin_did = self._get_new_did(twin_key_name)

        self._upsert_twin(
            twin_id=twin_did,
            properties=properties,
            feeds=feeds,
            visibility=visibility,
            location=location,
        )

        return twin_did

    def _describe_local_twin(self, twin_id: str):
        twin_description = self._make_api_call(
            method="GET", url=f"{HOST}/qapi/twins/{twin_id}"
        )

        return twin_description

    def _upsert_twin(
        self,
        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

        self._make_api_call(method="PUT", url=f"{HOST}/qapi/twins", json=payload)

        print(f"Twin {twin_id} created succesfully with Metadata, Feeds and Values")
        print("---")

    def create_twin_model(self):
        print("Creating Twin Model")
        # Build a list of Properties for the Twin Model
        property_list = [
            # Twin Model "special" property
            {
                "key": "http://www.w3.org/1999/02/22-rdf-syntax-ns#type",
                "uriValue": {"value": "https://data.iotics.com/app#Model"},
            },
            # Add Label
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#label",
                "langLiteralValue": {
                    "value": "Temperature Sensor Follower Model",
                    "lang": "en",
                },
            },
            # Add Comment
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#comment",
                "langLiteralValue": {
                    "value": "A temperature sensor that follows temperature data",
                    "lang": "en",
                },
            },
            # Add colour
            {
                "key": "https://data.iotics.com/app#color",
                "stringLiteralValue": {"value": "#9aceff"},
            },
            # Add Space name
            {
                "key": "https://data.iotics.com/app#spaceName",
                "stringLiteralValue": {"value": "my-space"},
            },
            # Add CreatedAt property
            {
                "key": "https://data.iotics.com/app#createdAt",
                "literalValue": {
                    "dataType": "dateTime",
                    "value": datetime.now(tz=timezone.utc).isoformat(),
                },
            },
            # Add UpdatedAt property
            {
                "key": "https://data.iotics.com/app#updatedAt",
                "literalValue": {
                    "dataType": "dateTime",
                    "value": datetime.now(tz=timezone.utc).isoformat(),
                },
            },
        ]

        model_twin_did = self._create_new_twin(
            twin_key_name="TwinFollowerModel", properties=property_list
        )

        return model_twin_did

    def create_twin_from_model(self, twin_model_id: str):
        twin_model = self._describe_local_twin(twin_id=twin_model_id)

        # Build a list of properties for the generic Twin
        property_list = [
            # Add the generic twins' "special" properties
            {
                "key": "https://data.iotics.com/app#model",
                "uriValue": {"value": twin_model_id},
            },
            {
                "key": "https://data.iotics.com/app#createdFrom",
                "uriValue": {"value": "https://data.iotics.com/app#ByModel"},
            },
        ]

        # Add to the list all the Twin Model's properties but consider the exceptions
        for prop in twin_model["result"]["properties"]:
            # This was the "special" property of the Twin Model, we don't need it here
            if prop["key"] == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type":
                continue
            # Don't add the Label property of the Twin Model.
            # It'll be added later based on the number of the sensor
            if prop["key"] == "http://www.w3.org/2000/01/rdf-schema#label":
                continue
            # Change the Created At value
            if prop["key"] == "https://data.iotics.com/app#createdAt":
                prop["literalValue"]["value"] = datetime.now(
                    tz=timezone.utc
                ).isoformat()
            # Change the Updated At value
            if prop["key"] == "https://data.iotics.com/app#updatedAt":
                prop["literalValue"]["value"] = datetime.now(
                    tz=timezone.utc
                ).isoformat()

            property_list.append(prop)

        # Finally create the Sensor Twin Follower
        twin_key_name = "temp_sensor_follower"
        twin_label = "Temperature Sensor Follower"

        # Add the Label property according to the number of the Sensor
        property_list_with_label = property_list.copy()
        property_list_with_label.append(
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#label",
                "langLiteralValue": {"value": twin_label, "lang": "en"},
            }
        )

        print(f"Creating Twin from Model - {twin_label}")

        twin_from_model_did = self._create_new_twin(
            twin_key_name=twin_key_name,
            properties=property_list_with_label,
            location={"lat": 51.5, "lon": -0.1},
        )

        return twin_from_model_did

    def search_for_twin_publishers(self):
        # Initialise an empty list.
        # It will contain the list of Twins retrieved by the search
        twins_list = []

        # Add a new temporary field in the headers.
        # Client request timeout is used to stop the request processing once the timeout is reached
        headers = self._headers.copy()
        headers.update(
            {
                "Iotics-RequestTimeout": (
                    datetime.now(tz=timezone.utc) + timedelta(seconds=10)
                ).isoformat(),
            }
        )

        with request(
            method="POST",
            url=f"{HOST}/qapi/searches",
            headers=headers,
            stream=True,
            verify=False,
            params={"scope": "LOCAL"},
            json={
                "responseType": "FULL",
                "filter": {
                    "properties": [
                        {
                            "key": "https://data.iotics.com/app#createdFrom",
                            "uriValue": {
                                "value": "https://data.iotics.com/app#ByModel"
                            },
                        },
                    ],
                    "text": "shares",
                },
            },
        ) as resp:
            # Raises HTTPError, if one occurred
            resp.raise_for_status()
            # Iterates over the response data, one line at a time
            for chunk in resp.iter_lines():
                response = json.loads(chunk)
                twins_found = []
                try:
                    twins_found = response["result"]["payload"]["twins"]
                except (KeyError, IndexError):
                    continue
                finally:
                    if twins_found:
                        # Append the twins found to the list of twins
                        twins_list.extend(twins_found)

        print(f"Found {len(twins_list)} twin(s)")

        return twins_list

    def follow_feed(self, twins_found_list: List[dict], twin_follower_did: str):
        for twin in twins_found_list:
            # Get the sensor number from the Twin's label last string
            twin_publisher_label = self.get_twin_label(twin["properties"])
            twin_publisher_id = twin["id"]["value"]

            # We are supposed to know in advance the Feed ID we want our Twin Follower to follow
            twin_publisher_feed_id = "currentTemp"

            # Get the latest shared value
            encoded_data = self._make_api_call(
                method="GET",
                url=f"{HOST}/qapi/twins/{twin_follower_did}/interests/twins/{twin_publisher_id}/feeds/{twin_publisher_feed_id}/samples/last",
            )

            time = encoded_data["feedData"]["occurredAt"]
            data = encoded_data["feedData"]["data"]

            feed_data = json.loads(base64.b64decode(data).decode("ascii"))

            print(
                f"Last shared data of Twin {twin_publisher_label} occurred at utc time {time}: {feed_data}"
            )

    @staticmethod
    def get_twin_label(property_list: List[dict]):
        for prop in property_list:
            if prop["key"] == "http://www.w3.org/2000/01/rdf-schema#label":
                return prop["langLiteralValue"]["value"]

        return None


def main():
    follower_connector = FollowerConnector()
    follower_connector.setup()
    model_twin_did = follower_connector.create_twin_model()
    follower_twin_did = follower_connector.create_twin_from_model(
        twin_model_id=model_twin_did
    )
    twins_list = follower_connector.search_for_twin_publishers()

    while True:
        follower_connector.follow_feed(
            twins_found_list=twins_list, twin_follower_did=follower_twin_did
        )
        sleep(10)
        print("---")


if __name__ == "__main__":
    main()

How to export data from an IOTICSpace

Data can be exported outside IOTICS in order to persistently store a Digital Twin's published data.

The code below provides an example of how to allow the follower_connector.py to store the sensor data received into either a CSV file or an SQLite database.

Click to see the code

Instructions:

  1. Install a new python library to use the SQLite Engine: pip install SQLAlchemy;
  2. Create a new directory called export_data. Within the latter create 2 other directories: csv and sqlite;
  3. Add the sqlite_engine.py code below to the sqlite directory and the csv_engine.py to the csv directory as follows:
├── publisher_connector.py
├── follower_connector_export.py
├── export_data
│   ├── csv
│   │   └── csv_engine.py
│   └── sqlite
│       └── sqlite_engine.py
  1. Run the publisher_connector.py in a terminal and the follower_connector_export.py code below in another terminal;
  2. Change the export_data_class class (line 352) with either CSV or SQLite;
  3. Look at the data being added to the temperatures.csv file or temperatures.db according to the choice above.
import base64
import json
from datetime import datetime, timedelta, timezone
from time import sleep
from typing import List

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

from export_data.csv.csv_engine import CSV
from export_data.sqlite.sqlite_engine import SQLite

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")


class FollowerConnectorExport:
    def __init__(self, export_data_class):
        self._api = None
        self._user_registered_id = None
        self._agent_registered_id = None
        self._headers = None
        self._twins_info = None
        self._export_data = export_data_class

    def setup(self):
        self._api = get_rest_high_level_identity_api(resolver_url=RESOLVER_URL)
        (
            self._user_registered_id,
            self._agent_registered_id,
        ) = self._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",
        )

        self._headers = {
            "accept": "application/json",
            "Iotics-ClientAppId": "follower_connector",
            "Content-Type": "application/json",
        }
        self._refresh_token()
        self._twins_info = {}

    def _refresh_token(self, duration: int = 60):
        print("Refreshing token")
        print("---")
        token = self._api.create_agent_auth_token(
            agent_registered_identity=self._agent_registered_id,
            user_did=self._user_registered_id.did,
            duration=duration,
        )

        self._headers["Authorization"] = f"Bearer {token}"

    def _get_new_did(self, twin_key_name: str):
        twin_registered_id = self._api.create_twin_with_control_delegation(
            twin_seed=AGENT_SEED,
            twin_key_name=twin_key_name,
            agent_registered_identity=self._agent_registered_id,
            delegation_name="#ControlDeleg",
        )

        return twin_registered_id.did

    def _make_api_call(
        self, method: str, url: str, json: dict = None, retry: bool = True
    ):
        try:
            response = request(method=method, url=url, headers=self._headers, json=json)
            response.raise_for_status()
        except Exception as ex:
            if retry:
                self._refresh_token()
                return self._make_api_call(method, url, json, retry=False)
            else:
                print("Retried once. Still getting error", ex)

        return response.json()

    def _create_new_twin(
        self,
        twin_key_name: str,
        properties: List[str] = None,
        feeds: List[str] = None,
        visibility: str = "PRIVATE",
        location: dict = None,
    ):
        twin_did = self._get_new_did(twin_key_name)

        self._upsert_twin(
            twin_id=twin_did,
            properties=properties,
            feeds=feeds,
            visibility=visibility,
            location=location,
        )

        return twin_did

    def _describe_local_twin(self, twin_id: str):
        twin_description = self._make_api_call(
            method="GET", url=f"{HOST}/qapi/twins/{twin_id}"
        )

        return twin_description

    def _upsert_twin(
        self,
        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

        self._make_api_call(method="PUT", url=f"{HOST}/qapi/twins", json=payload)

        print(f"Twin {twin_id} created succesfully with Metadata, Feeds and Values")
        print("---")

    def create_twin_model(self):
        print("Creating Twin Model")
        # Build a list of Properties for the Twin Model
        property_list = [
            # Twin Model "special" property
            {
                "key": "http://www.w3.org/1999/02/22-rdf-syntax-ns#type",
                "uriValue": {"value": "https://data.iotics.com/app#Model"},
            },
            # Add Label
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#label",
                "langLiteralValue": {
                    "value": "Temperature Sensor Follower Model",
                    "lang": "en",
                },
            },
            # Add Comment
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#comment",
                "langLiteralValue": {
                    "value": "A temperature sensor that follows temperature data",
                    "lang": "en",
                },
            },
            # Add colour
            {
                "key": "https://data.iotics.com/app#color",
                "stringLiteralValue": {"value": "#9aceff"},
            },
            # Add Space name
            {
                "key": "https://data.iotics.com/app#spaceName",
                "stringLiteralValue": {"value": "my-space"},
            },
            # Add CreatedAt property
            {
                "key": "https://data.iotics.com/app#createdAt",
                "literalValue": {
                    "dataType": "dateTime",
                    "value": datetime.now(tz=timezone.utc).isoformat(),
                },
            },
            # Add UpdatedAt property
            {
                "key": "https://data.iotics.com/app#updatedAt",
                "literalValue": {
                    "dataType": "dateTime",
                    "value": datetime.now(tz=timezone.utc).isoformat(),
                },
            },
        ]

        model_twin_did = self._create_new_twin(
            twin_key_name="TwinFollowerModel", properties=property_list
        )

        return model_twin_did

    def create_twins_from_model(self, twin_model_id: str):
        twin_model = self._describe_local_twin(twin_id=twin_model_id)

        # Build a list of properties for the generic Twin
        property_list = [
            # Add the generic twins' "special" properties
            {
                "key": "https://data.iotics.com/app#model",
                "uriValue": {"value": twin_model_id},
            },
            {
                "key": "https://data.iotics.com/app#createdFrom",
                "uriValue": {"value": "https://data.iotics.com/app#ByModel"},
            },
        ]

        # Add to the list all the Twin Model's properties but consider the exceptions
        for prop in twin_model["result"]["properties"]:
            # This was the "special" property of the Twin Model, we don't need it here
            if prop["key"] == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type":
                continue
            # Don't add the Label property of the Twin Model.
            # It'll be added later based on the number of the sensor
            if prop["key"] == "http://www.w3.org/2000/01/rdf-schema#label":
                continue
            # Change the Created At value
            if prop["key"] == "https://data.iotics.com/app#createdAt":
                prop["literalValue"]["value"] = datetime.now(
                    tz=timezone.utc
                ).isoformat()
            # Change the Updated At value
            if prop["key"] == "https://data.iotics.com/app#updatedAt":
                prop["literalValue"]["value"] = datetime.now(
                    tz=timezone.utc
                ).isoformat()

            property_list.append(prop)

        # Finally create the 10 Sensor Twin Followers
        for sensor_number in range(1, 11):
            twin_key_name = f"temp_sensor_follower_{sensor_number}"
            twin_label = f"Temperature Sensor Follower {sensor_number}"

            # Add the Label property according to the number of the Sensor
            property_list_with_label = property_list.copy()
            property_list_with_label.append(
                {
                    "key": "http://www.w3.org/2000/01/rdf-schema#label",
                    "langLiteralValue": {"value": twin_label, "lang": "en"},
                }
            )

            print(f"Creating Twin from Model - {twin_label}")

            twin_from_model_did = self._create_new_twin(
                twin_key_name=twin_key_name,
                properties=property_list,
                location={"lat": 51.5, "lon": -0.1},
            )

            # Add the Twin's info in the mapping dictionary
            twin_info = {sensor_number: twin_from_model_did}
            self._twins_info.update(twin_info)

    def search_for_twin_publishers(self):
        # Initialise an empty list.
        # It will contain the list of Twins retrieved by the search
        twins_list = []

        # Add a new temporary field in the headers.
        # Client request timeout is used to stop the request processing once the timeout is reached
        headers = self._headers.copy()
        headers.update(
            {
                "Iotics-RequestTimeout": (
                    datetime.now(tz=timezone.utc) + timedelta(seconds=10)
                ).isoformat(),
            }
        )

        with request(
            method="POST",
            url=f"{HOST}/qapi/searches",
            headers=headers,
            stream=True,
            verify=False,
            params={"scope": "LOCAL"},
            json={
                "responseType": "FULL",
                "filter": {
                    "properties": [
                        {
                            "key": "https://data.iotics.com/app#createdFrom",
                            "uriValue": {
                                "value": "https://data.iotics.com/app#ByModel"
                            },
                        },
                    ],
                    "text": "shares",
                },
            },
        ) as resp:
            # Raises HTTPError, if one occurred
            resp.raise_for_status()
            # Iterates over the response data, one line at a time
            for chunk in resp.iter_lines():
                response = json.loads(chunk)
                twins_found = []
                try:
                    twins_found = response["result"]["payload"]["twins"]
                except (KeyError, IndexError):
                    continue
                finally:
                    if twins_found:
                        # Append the twins found to the list of twins
                        twins_list.extend(twins_found)

        print(f"Found {len(twins_list)} twin(s)")

        return twins_list

    def follow_data(self, twins_list: List[dict]):
        for twin in twins_list:
            # Get the sensor number from the Twin's label last string
            twin_publisher_label = self._get_twin_label(twin["properties"])
            sensor_number = twin_publisher_label.split(" ")[-1]
            # Check if the current Twin is the Twin Model.
            # In such case we don't want to follow it.
            if not sensor_number.isdigit():
                continue

            twin_publisher_id = twin["id"]["value"]
            twin_follower_id = self._twins_info.get(int(sensor_number))

            # We are supposed to know in advance the Feed ID we want our Twin Follower to follow
            twin_publisher_feed_id = "currentTemp"

            # Get the latest shared value
            encoded_data = self._make_api_call(
                method="GET",
                url=f"{HOST}/qapi/twins/{twin_follower_id}/interests/twins/{twin_publisher_id}/feeds/{twin_publisher_feed_id}/samples/last",
            )

            datetime = encoded_data["feedData"]["occurredAt"]
            data = encoded_data["feedData"]["data"]

            feed_data = json.loads(base64.b64decode(data).decode("ascii"))

            print(
                f"Last shared data of Twin {twin_publisher_label} occurred at utc time {datetime}: {feed_data}"
            )

            # Store data externally
            self._export_data.export(
                datetime=datetime,
                received_from=twin_publisher_label,
                data=feed_data.get("reading"),
            )

    def _get_twin_label(self, property_list: List[dict]):
        for prop in property_list:
            if prop["key"] == "http://www.w3.org/2000/01/rdf-schema#label":
                return prop["langLiteralValue"]["value"]

        return None


def main():
    # Replace the 'export_data_class' class with 'CSV' to store the data into a CSV file
    follower_connector = FollowerConnectorExport(export_data_class=SQLite())
    follower_connector.setup()
    model_twin_did = follower_connector.create_twin_model()
    follower_connector.create_twins_from_model(twin_model_id=model_twin_did)
    twins_list = follower_connector.search_for_twin_publishers()

    while True:
        follower_connector.follow_data(twins_list)
        sleep(10)
        print("---")


if __name__ == "__main__":
    main()
from sqlalchemy import Column, Float, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DB_NAME = "sqlite:///export_data/sqlite/temperatures.db"
Base = declarative_base()


class SensorReading(Base):
    __tablename__ = "SensorReading"

    id = Column(Integer, primary_key=True)
    timestamp = Column(String(100))
    sensor_number = Column(Integer)
    sensor_reading = Column(Float)


class SQLite:
    def __init__(self, echo=False):
        engine = create_engine(DB_NAME, echo=echo)
        Base.metadata.create_all(bind=engine)
        Session = sessionmaker(engine)
        self._session = Session()

    def _store(self, item: SensorReading):
        with self._session as session:
            session.add(item)
            session.commit()

        print("Item stored correctly")

    def export(self, datetime: str, received_from: str, data: dict):
        sensor_reading_obj = SensorReading(
            timestamp=datetime, sensor_number=received_from, sensor_reading=data
        )

        self._store(sensor_reading_obj)
import csv
from collections import namedtuple

CSV_FILE = "./export_data/csv/temperatures.csv"
CSV_FIELDS = ["timestamp", "sensor_number", "sensor_reading"]
SensorReading = namedtuple("SensorReading", CSV_FIELDS)


class CSV:
    def __init__(self, n_items=10):
        self._items_list = []
        self._n_items = n_items

        with open(CSV_FILE, "w") as csv_file:
            writer = csv.writer(csv_file)
            # Write header
            writer.writerow(CSV_FIELDS)

    def _store(self):
        with open(CSV_FILE, "a") as csv_file:
            writer = csv.writer(csv_file)
            for item in self._items_list:
                writer.writerow(item)

        print("Item stored correctly")
        self._items_list.clear()

    def export(self, datetime: str, received_from: str, data: dict):
        new_item = SensorReading(
            timestamp=datetime, sensor_number=received_from, sensor_reading=data
        )
        self._items_list.append(new_item)

        if len(self._items_list) >= self._n_items:
            self._store()

Synthesiser Connector

A Synthesiser Connector, or simply Synthesiser, is an application that continuously allows Follower Twins to receive, transform and send the data back into the Space.

In other words, a Synthesiser creates a set of Twins that:

  1. continuously follow Publisher Twins' feeds;
  2. "synthesise" or interoperate the data;
  3. publish the new data back as a feed.
Representation of a Synthesiser ConnectorRepresentation of a Synthesiser Connector

Representation of a Synthesiser Connector

How to create a Synthesiser Connector with the IOTICS API

A Synthesiser application creates a set of Follower Twins with one or more Feeds so that they both receive data and share the new data.

The following snippet provides an example of a Synthesiser that:

  1. Creates 10 Synthesiser Twins with a single Feed;
  2. Follows the Publisher Connector Twins' feed with temperature readings;
  3. Computes the average temperature for each sensor Twin;
  4. Shares the average data as a feed.
Click to see the code

Instructions:

  1. Run the publisher_connector.py in a terminal;
  2. Run the synthesiser_connector.py code below in another terminal.
1. Get a fresh token to be used in the requests' headers;
2. Create a Twin Model;
  2.1 Build list of Properties;
  2.2 Build list of Feeds;
  2.3 Get a new Twin DID;
  2.4 Use the Upsert Twin API;
3. Create Twins from Model;
  3.1 Use the Describe Local Twin API to get the Twin Model's metadata;
  3.2 Build list of Properties;
  3.3 Use the Describe Feed API to get the  the Twin Model's Feed's metadata;
  3.4 Build list of Feeds;
  For each Twin to follow:
  3.5 Get a new Twin DID;
  3.6 Use the Upsert Twin API;
Every 10 seconds:
4. Get the Twin Publishers' last shared data;
5. Compute the average temperature;
6. Share data;
import base64
import json
from datetime import datetime, timezone, timedelta
from time import sleep
from typing import List

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")


class SynthesiserConnector:
    def __init__(self):
        self._api = None
        self._user_registered_id = None
        self._agent_registered_id = None
        self._headers = None
        self._twins_info = None

    def setup(self):
        self._api = get_rest_high_level_identity_api(resolver_url=RESOLVER_URL)
        (
            self._user_registered_id,
            self._agent_registered_id,
        ) = self._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",
        )

        self._headers = {
            "accept": "application/json",
            "Iotics-ClientAppId": "synthesiser_connector",
            "Content-Type": "application/json",
        }
        self._refresh_token()
        self._twins_info = {}

    def _refresh_token(self, duration: int = 60):
        print("Refreshing token")
        print("---")
        token = self._api.create_agent_auth_token(
            agent_registered_identity=self._agent_registered_id,
            user_did=self._user_registered_id.did,
            duration=duration,
        )

        self._headers["Authorization"] = f"Bearer {token}"

    def _get_new_did(self, twin_key_name: str):
        twin_registered_id = self._api.create_twin_with_control_delegation(
            twin_seed=AGENT_SEED,
            twin_key_name=twin_key_name,
            agent_registered_identity=self._agent_registered_id,
            delegation_name="#ControlDeleg",
        )

        return twin_registered_id.did

    def _make_api_call(
        self, method: str, url: str, json: dict = None, retry: bool = True
    ):
        try:
            response = request(method=method, url=url, headers=self._headers, json=json)
            response.raise_for_status()
        except Exception as ex:
            if retry:
                self._refresh_token()
                return self._make_api_call(method, url, json, retry=False)
            else:
                print("Retried once. Still getting error", ex)

        return response.json()

    def _create_new_twin(
        self,
        twin_key_name: str,
        properties: List[str] = None,
        feeds: List[str] = None,
        visibility: str = "PRIVATE",
        location: dict = None,
    ):
        twin_did = self._get_new_did(twin_key_name)

        self._upsert_twin(
            twin_id=twin_did,
            properties=properties,
            feeds=feeds,
            visibility=visibility,
            location=location,
        )

        return twin_did

    def _describe_feed(self, twin_id: str, feed_id: str):
        feed_description = self._make_api_call(
            method="GET",
            url=f"{HOST}/qapi/twins/{twin_id}/feeds/{feed_id}",
        )

        return feed_description

    def _describe_local_twin(self, twin_id: str):
        twin_description = self._make_api_call(
            method="GET", url=f"{HOST}/qapi/twins/{twin_id}"
        )

        return twin_description

    def _upsert_twin(
        self,
        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

        self._make_api_call(method="PUT", url=f"{HOST}/qapi/twins", json=payload)

        print(f"Twin {twin_id} created succesfully with Metadata, Feeds and Values")
        print("---")

    def _share_data(self, twin_label: str, twin_id: str, feed_id: str, value):
        data_to_share = {"average": value}
        encoded_data = base64.b64encode(json.dumps(data_to_share).encode()).decode()
        data_to_share_payload = {
            "sample": {
                "data": encoded_data,
                "mime": "application/json",
                "occurredAt": datetime.now(tz=timezone.utc).isoformat(
                    timespec="seconds"
                ),
            }
        }
        self._make_api_call(
            method="POST",
            url=f"{HOST}/qapi/twins/{twin_id}/feeds/{feed_id}/shares",
            json=data_to_share_payload,
        )

        print(f"Shared {data_to_share} from Twin {twin_label}")

    def create_twin_model(self):
        print("Creating Twin Model")
        # Build a list of Properties for the Twin Model
        property_list = [
            # Twin Model "special" property
            {
                "key": "http://www.w3.org/1999/02/22-rdf-syntax-ns#type",
                "uriValue": {"value": "https://data.iotics.com/app#Model"},
            },
            # Add Label
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#label",
                "langLiteralValue": {
                    "value": "Temperature Sensor Synthesiser Model",
                    "lang": "en",
                },
            },
            # Add Comment
            {
                "key": "http://www.w3.org/2000/01/rdf-schema#comment",
                "langLiteralValue": {
                    "value": "A temperature sensor that follows temperature data",
                    "lang": "en",
                },
            },
            # Add colour
            {
                "key": "https://data.iotics.com/app#color",
                "stringLiteralValue": {"value": "#9aceff"},
            },
            # Add Space name
            {
                "key": "https://data.iotics.com/app#spaceName",
                "stringLiteralValue": {"value": "my-space"},
            },
            # Add CreatedAt property
            {
                "key": "https://data.iotics.com/app#createdAt",
                "literalValue": {
                    "dataType": "dateTime",
                    "value": datetime.now(tz=timezone.utc).isoformat(),
                },
            },
            # Add UpdatedAt property
            {
                "key": "https://data.iotics.com/app#updatedAt",
                "literalValue": {
                    "dataType": "dateTime",
                    "value": datetime.now(tz=timezone.utc).isoformat(),
                },
            },
        ]

        # Build a list of Feeds for the Twin Model
        feed_list = [
            # Add a single Feed called "currentTemp"
            {
                "id": "averageTemp",
                "storeLast": True,  # We want to store the last value sent
                "properties": [
                    # Add a Label for the Feed
                    {
                        "key": "http://www.w3.org/2000/01/rdf-schema#label",
                        "langLiteralValue": {
                            "value": "Average Temperature",
                            "lang": "en",
                        },
                    }
                ],
                # Add a single Value called "reading"
                "values": [
                    {
                        "comment": "Temperature in degrees Celsius",
                        "dataType": "decimal",
                        "label": "average",
                        "unit": "http://qudt.org/vocab/unit/DEG_C",
                    }
                ],
            },
        ]

        model_twin_did = self._create_new_twin(
            twin_key_name="TwinSynthesiserModel",
            properties=property_list,
            feeds=feed_list,
        )

        return model_twin_did

    def create_twins_from_model(self, twin_model_id: str):
        twin_model = self._describe_local_twin(twin_id=twin_model_id)

        # Build a list of properties for the generic Twin
        property_list = [
            # Add the generic twins' "special" properties
            {
                "key": "https://data.iotics.com/app#model",
                "uriValue": {"value": twin_model_id},
            },
            {
                "key": "https://data.iotics.com/app#createdFrom",
                "uriValue": {"value": "https://data.iotics.com/app#ByModel"},
            },
        ]

        # Add to the list all the Twin Model's properties but consider the exceptions
        for prop in twin_model["result"]["properties"]:
            # This was the "special" property of the Twin Model, we don't need it here
            if prop["key"] == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type":
                continue
            # Don't add the Label property of the Twin Model.
            # It'll be added later based on the number of the sensor
            if prop["key"] == "http://www.w3.org/2000/01/rdf-schema#label":
                continue
            # Change the Created At value
            if prop["key"] == "https://data.iotics.com/app#createdAt":
                prop["literalValue"]["value"] = datetime.now(
                    tz=timezone.utc
                ).isoformat()
            # Change the Updated At value
            if prop["key"] == "https://data.iotics.com/app#updatedAt":
                prop["literalValue"]["value"] = datetime.now(
                    tz=timezone.utc
                ).isoformat()

            property_list.append(prop)

        # Build a list of Feeds for the Twin Model
        feed_list = []

        # Scan the Twin Model's feeds list to get the Twin Model's Feed IDs.
        # Then use the Feed Describe API to get the Feeds' metadata
        for feed in twin_model["result"]["feeds"]:
            feed_id = feed["feedId"]["value"]

            # Use the Feed Describe API
            feed_description = self._describe_feed(
                twin_id=twin_model_id, feed_id=feed_id
            )
            feed_properties = feed_description["result"]["properties"]
            feed_values = feed_description["result"]["values"]
            store_last = feed_description["result"]["storeLast"]

            # Append to the list all the Feed's metadata
            feed_list.append(
                {
                    "id": feed_id,
                    "properties": feed_properties,
                    "values": feed_values,
                    "storeLast": store_last,
                }
            )

        # Finally create the 10 Twin Synthesisers
        for sensor_number in range(1, 11):
            twin_key_name = f"avg_temp_sensor_{sensor_number}"
            twin_label = f"Temperature Sensor Synthesiser {sensor_number}"

            # Add the Label property according to the number of the Sensor
            property_list_with_label = property_list.copy()
            property_list_with_label.append(
                {
                    "key": "http://www.w3.org/2000/01/rdf-schema#label",
                    "langLiteralValue": {"value": twin_label, "lang": "en"},
                }
            )

            print(f"Creating Twin from Model - {twin_label}")

            twin_from_model_did = self._create_new_twin(
                twin_key_name=twin_key_name,
                properties=property_list,
                feeds=feed_list,
                location={"lat": 51.5, "lon": -0.1},
            )

            # Add the Twin's info in the mapping dictionary
            twin_info = {
                sensor_number: {
                    "twin_label": twin_label,
                    "twin_id": twin_from_model_did,
                    "feed_id": feed_id,
                    "previous_mean": 0,
                    "sample_count": 1,
                }
            }
            self._twins_info.update(twin_info)

    def search_for_twin_publishers(self):
        # Initialise an empty list.
        # It will contain the list of Twins retrieved by the search
        twins_list = []

        # Add a new temporary field in the headers.
        # Client request timeout is used to stop the request processing once the timeout is reached
        headers = self._headers.copy()
        headers.update(
            {
                "Iotics-RequestTimeout": (
                    datetime.now(tz=timezone.utc) + timedelta(seconds=10)
                ).isoformat(),
            }
        )

        with request(
            method="POST",
            url=f"{HOST}/qapi/searches",
            headers=headers,
            stream=True,
            verify=False,
            params={"scope": "LOCAL"},
            json={
                "responseType": "FULL",
                "filter": {
                    "properties": [
                        {
                            "key": "https://data.iotics.com/app#createdFrom",
                            "uriValue": {
                                "value": "https://data.iotics.com/app#ByModel"
                            },
                        },
                    ],
                    "text": "shares",
                },
            },
        ) as resp:
            # Raises HTTPError, if one occurred
            resp.raise_for_status()
            # Iterates over the response data, one line at a time
            for chunk in resp.iter_lines():
                response = json.loads(chunk)
                twins_found = []
                try:
                    twins_found = response["result"]["payload"]["twins"]
                except (KeyError, IndexError):
                    continue
                finally:
                    if twins_found:
                        # Append the twins found to the list of twins
                        twins_list.extend(twins_found)

        print(f"Found {len(twins_list)} twin(s)")

        return twins_list

    def synthesise_data(self, twins_list: List[dict]):
        for twin in twins_list:
            # Get the sensor number from the Twin's label last string
            twin_publisher_label = self._get_twin_label(twin["properties"])
            sensor_number = twin_publisher_label.split(" ")[-1]
            # Check if the current Twin is the Twin Model.
            # In such case we don't want to follow it.
            if not sensor_number.isdigit():
                continue

            twin_publisher_id = twin["id"]["value"]
            twin_info = self._twins_info.get(int(sensor_number))
            synthesiser_twin_id = twin_info["twin_id"]
            synthesiser_twin_previous_mean = twin_info["previous_mean"]
            synthesiser_twin_sample_count = twin_info["sample_count"]

            # We are supposed to know in advance the Feed ID we want our Twin Synthesiser to follow
            twin_publisher_feed_id = "currentTemp"

            # Get the latest shared value
            encoded_data = self._make_api_call(
                method="GET",
                url=f"{HOST}/qapi/twins/{synthesiser_twin_id}/interests/twins/{twin_publisher_id}/feeds/{twin_publisher_feed_id}/samples/last",
            )

            time = encoded_data["feedData"]["occurredAt"]
            data = encoded_data["feedData"]["data"]

            feed_data = json.loads(base64.b64decode(data).decode("ascii"))

            print(
                f"Last shared data of Twin {twin_publisher_label} occurred at utc time {time}: {feed_data}"
            )

            new_sample = feed_data["reading"]
            new_mean = (
                (synthesiser_twin_sample_count - 1) * synthesiser_twin_previous_mean
                + new_sample
            ) / synthesiser_twin_sample_count

            # Publish the new data
            self.publish_data(twin_info=twin_info, data=new_mean)

            # Update the average
            self._twins_info[int(sensor_number)]["previous_mean"] = new_mean
            self._twins_info[int(sensor_number)]["sample_count"] += 1

    def publish_data(self, twin_info: dict, data: dict):
        synthesiser_twin_id = twin_info["twin_id"]
        synthesiser_feed_id = twin_info["feed_id"]
        synthesiser_twin_label = twin_info["twin_label"]

        self._share_data(
            twin_label=synthesiser_twin_label,
            twin_id=synthesiser_twin_id,
            feed_id=synthesiser_feed_id,
            value=data,
        )

    def _get_twin_label(self, property_list: List[dict]):
        for prop in property_list:
            if prop["key"] == "http://www.w3.org/2000/01/rdf-schema#label":
                return prop["langLiteralValue"]["value"]

        return None


def main():
    synthesiser_connector = SynthesiserConnector()
    synthesiser_connector.setup()
    model_twin_did = synthesiser_connector.create_twin_model()
    synthesiser_connector.create_twins_from_model(twin_model_id=model_twin_did)
    twins_list = synthesiser_connector.search_for_twin_publishers()

    while True:
        synthesiser_connector.synthesise_data(twins_list)
        sleep(10)
        print("---")


if __name__ == "__main__":
    main()

Did this page help you?