Digital Twins

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

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

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();

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

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