Upsert Twins
Upsert Twin deletes an existing Digital Twin and recreates it from scratch with new values. The same operation also allows adding Feeds and Inputs with Metadata in a single call to streamline the process and reduce the number of API calls.
This page covers:
Introduction to Upsert Digital Twins
Upsert Twin allows, in this order, to:
- Delete the Twin
- Create the Twin
- Update the Twin
- Add Feeds and Inputs with Metadata
Be careful with the use of the Upsert Twin. It will delete and re-create the Twin only with the values specified in the body of the request.
The body parameters required to use the Upsert Twin API are the same as for Update Twin API in addition with the ones described in the table below.
| Key | Description |
|---|---|
| Twin DID | The DID of the Twin that you want to upsert. Mandatory |
| Feeds | The list of Feeds you want to add to the Twin including: - the Feed(s) metadata; - the Value(s) metadata. Optional |
| Inputs | The list of Inputs you want to add to the Twin including: - the Input(s) metadata; - the Value(s) metadata. Optional |
Upsert a Twin with the IOTICS API
Click to see the prerequisites
1. Create Twin with control delegation and get a new "twin_registered_id" (DID)
2. Use the Upsert Twin API call
from requests import request
response = request(
method="PUT",
url=f"{HOST}/qapi/twins", # Replace with URL of the Host
headers=headers, # It includes the token
json={
"twinId": {"hostId": host_id, "id": twin_did},
"location": {"lat": 51.5, "lon": -0.1},
"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": "reading",
"unit": "http://purl.obolibrary.org/obo/UO_0000027",
}
],
}
],
},
)
response.raise_for_status()
Upsert Twin Tutorial
Please note that the following snippets provides exactly the same result as the Update Twin Tutorial.
Click to see the entire code on how to upsert a Twin
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 IoticsRest:
def __init__(self):
self._high_level_api = get_rest_high_level_identity_api(
resolver_url=RESOLVER_URL
)
(
self._user_registered_id,
self._agent_registered_id,
) = self._high_level_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 = self._high_level_api.create_agent_auth_token(
agent_registered_identity=self._agent_registered_id,
user_did=self._user_registered_id.did,
duration=10,
)
self._headers = {
"accept": "application/json",
"Iotics-ClientAppId": "example_code",
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
def _make_api_call(self, method: str, url: str, json: dict = None):
response = request(method=method, url=url, headers=self._headers, json=json)
response.raise_for_status()
return response.json()
def create_twin_identity(self, twin_key_name: str):
twin_registered_id = self._high_level_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 get_host_id(self):
host_id = self._make_api_call(method="GET", url=f"{HOST}/qapi/host/id")
return host_id["hostId"]
def upsert_twin(
self,
twin_did: str,
host_id: str,
feeds: List[dict] = None,
inputs: List[dict] = None,
location: dict = None,
properties: List[dict] = None,
):
payload = {"twinId": {"hostId": host_id, "id": twin_did}}
if location:
payload["location"] = location
if feeds:
payload["feeds"] = feeds
if inputs:
payload["inputs"] = inputs
if properties:
payload["properties"] = properties
self._make_api_call(method="PUT", url=f"{HOST}/qapi/twins", json=payload)
def main():
iotics_rest = IoticsRest()
twin_did = iotics_rest.create_twin_identity(twin_key_name="TwinSensor")
host_id = iotics_rest.get_host_id()
iotics_rest.upsert_twin(
twin_did=twin_did,
host_id=host_id,
location={"lat": 51.5, "lon": -0.1},
properties=[
# Label
{
"key": "http://www.w3.org/2000/01/rdf-schema#label",
"langLiteralValue": {"value": "Twin Model", "lang": "en"},
},
# Comment
{
"key": "http://www.w3.org/2000/01/rdf-schema#comment",
"langLiteralValue": {
"value": "An example of a Twin Model",
"lang": "en",
},
},
# Colour
{
"key": "https://data.iotics.com/app#color",
"stringLiteralValue": {"value": "#d6df23"},
},
# Space Name
{
"key": "https://data.iotics.com/app#spaceName",
"stringLiteralValue": {"value": "my-space-name"},
},
],
feeds=[
{
"id": "temperature",
"storeLast": True,
"properties": [
# Add a Label
{
"key": "http://www.w3.org/2000/01/rdf-schema#label",
"langLiteralValue": {
"value": "Current Temperature",
"lang": "en",
},
},
# Add a comment
{
"key": "http://www.w3.org/2000/01/rdf-schema#comment",
"langLiteralValue": {
"value": "The current temperature reading",
"lang": "en",
},
},
],
"values": [
{
"comment": "Temperature in degrees Celsius",
"dataType": "decimal",
"label": "reading",
"unit": "http://qudt.org/vocab/unit/DEG_C",
}
],
}
],
inputs=[
{
"id": "on_off_switch",
"properties": [
# Add a Label
{
"key": "http://www.w3.org/2000/01/rdf-schema#label",
"langLiteralValue": {"value": "On/Off Switch", "lang": "en"},
},
# Add a comment
{
"key": "http://www.w3.org/2000/01/rdf-schema#comment",
"langLiteralValue": {
"value": "Allows the sensor to turn ON/OFF",
"lang": "en",
},
},
],
"values": [
{
"comment": "Switch of the sensor",
"dataType": "boolean",
"label": "switch",
}
],
}
],
)
if __name__ == "__main__":
main()
Updated about 1 month ago