Utilizing the SensorThing API
Register Entities¶

Preliminary Steps¶

This section contains the preliminary steps to set up the base URL, headers, and import necessary libraries.

Replace IST_SOS_ENDPOINT in the following script with your istSOS base URL (http://localhost:8018/istsos4/v1.1 or https://istsos.org/v4/v1.1).

import json
import re
from datetime import datetime

import requests
from IPython.display import Markdown, display

IST_SOS_ENDPOINT = "http://localhost:8018/istsos4/v1.1"

username = input("Enter your username: ")
password = input("Enter your password: ")

if not username or not password:
    print("Username or password is empty")

else:
    data = {
        "username": username,
        "password": password,
        "grant_type": "password",
    }

    response = requests.post(IST_SOS_ENDPOINT + "/Login", data=data)
    if response.status_code == 200:
        token = response.json()["access_token"]
        print(
            f"Token expires at: { datetime.fromtimestamp(response.json()['expires_in'])}"
        )
        prefix = username + "-"
        print("Your station name will be prefixed with: " + prefix)
    else:
        result = json.dumps(response.json(), indent=2)
        display(Markdown(f"```json\n{result}\n```"))

Create a Thing¶

and store the id of the created thing in things_id

body = {
    "name": f"{prefix}FIU_VAL",
    "description": "Water level, water temperature and water electrical conductivity recorder Ticino river",
    "properties": {
        "keywords": "water,river,height,temperature,conductivity,ACSOT",
        "description": "River level, water temperature and water electrical conductivity fiume Ticino valle",
    },
}

response = requests.post(
    IST_SOS_ENDPOINT + "/Things",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create new thing",
    },
)

if response.status_code == 201:
    print(f"Thing created successfully ({response.headers['location']})")
    match = re.search(r'\((\d+)\)', response.headers['location'])
    if match:
        thing_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

Create a Location¶

To create a Location and link it to an existing Thing, you include the Thing's ID in the JSON payload of the Location you are creating.
..and store the id of the created thing in location_id

body = {
    "name": f"{prefix}fiume Ticino valle",
    "description": "",
    "encodingType": "application/vnd.geo+json",
    "location": {
        "type": "Point",
        "coordinates": [
            8.956099,
            46.172245
        ]
    },
    "Things": [
        { "@iot.id": thing_id}
    ]
}

response = requests.post(
    IST_SOS_ENDPOINT + "/Locations",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create new location",
    }
)

if response.status_code == 201:
    print(f"Location created successfully ({response.headers['location']})")
    match = re.search(r'\((\d+)\)', response.headers['location'])
    if match:
        location_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

Create an Observed Property¶

..and store the id of the created thing in observed_property_id

body = {
    "name": f"{prefix}ground:water:voltage",
    "description": "Ground water voltage",
    "properties": {},
    "definition": "{}",
}

response = requests.post(
    IST_SOS_ENDPOINT + "/ObservedProperties",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create new ObservedProperty",
    },
)

if response.status_code == 201:
    print(
        f"ObservedProperty created successfully ({response.headers['location']})"
    )
    match = re.search(r"\((\d+)\)", response.headers["location"])
    if match:
        observed_property_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

Create a Sensor¶

..and store the id of the created thing in sensor_id

body = {
    "name": f"{prefix}Ecolog 1000",
    "description": "",
    "properties": {},
    "encodingType": "application/json",
    "metadata": '{"brand": "OTT", "type": "Pressure, temperature, electrical conductivity sensor"}',
}

response = requests.post(
    IST_SOS_ENDPOINT + "/Sensors",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create new Sensor",
    },
)

if response.status_code == 201:
    print(f"Sensor created successfully ({response.headers['location']})")
    match = re.search(r"\((\d+)\)", response.headers["location"])
    if match:
        sensor_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

Create a Datastream¶

The Datastream requires a Thing, Sensor and ObservedProperty.
.. we use the thing_id, sensor_id and observed_property_id

body = {
    "name": f"{prefix}V_FIU_VAL",
    "description": "",
    "observationType": "",
    "unitOfMeasurement": {"name": "Voltage", "symbol": "V", "definition": ""},
    "Thing": {"@iot.id": thing_id},
    "Sensor": {"@iot.id": sensor_id},
    "ObservedProperty": {"@iot.id": observed_property_id},
}

response = requests.post(
    IST_SOS_ENDPOINT + "/Datastreams",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create new Datastream",
    },
)

if response.status_code == 201:
    print(f"Datastream created successfully ({response.headers['location']})")
    match = re.search(r"\((\d+)\)", response.headers["location"])
    if match:
        datastream_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

Create an Observation¶

When creating Observations, the following additional rules apply:

If the phenomenonTime is not specified in the JSON payload, the server will automatically assign the current time as the phenomenonTime.
If the featureOfInterest is not provided, the server will generate a FeatureOfInterest based on the Location associated with the Thing from the relevant Datastream.

Create Observation (Datastream in the JSON and FeatureOfInterest set to the Thing's Location)¶

body = {"result": 3.63, "Datastream": {"@iot.id": datastream_id}}

response = requests.post(
    IST_SOS_ENDPOINT + "/Observations",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create new Observation",
    },
)

if response.status_code == 201:
    print(f"Observation created successfully ({response.headers['location']})")
    match = re.search(r"\((\d+)\)", response.headers["location"])
    if match:
        observation_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

Create Observation (Datastream in the JSON and new FeatureOfInterest in the JSON)¶

body = {
    "result": 3.62,
    "Datastream": {"@iot.id": datastream_id},
    "FeatureOfInterest": {
        "name": f"{prefix}A weather station 1.",
        "description": "A weather station.",
        "feature": {"type": "Point", "coordinates": [8.956099, 46.172335]},
        "encodingType": "application/vnd.geo+json",
    },
}

response = requests.post(
    IST_SOS_ENDPOINT + f"/Observations",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create new Observation",
    },
)

if response.status_code == 201:
    print(f"Observation created successfully ({response.headers['location']})")
    match = re.search(r"\((\d+)\)", response.headers["location"])
    if match:
        observation_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

Create Observation (Datastream in the URL and new FeatureOfInterest in the JSON)¶

body = {
    "result": 23,
    "FeatureOfInterest": {
        "name": f"{prefix}A weather station 2.",
        "description": "A weather station.",
        "feature": {"type": "Point", "coordinates": [8.956229, 46.172245]},
        "encodingType": "application/vnd.geo+json",
    },
}

response = requests.post(
    IST_SOS_ENDPOINT + f"/Datastreams({datastream_id})/Observations",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create new Observation",
    },
)

if response.status_code == 201:
    print(f"Observation created successfully ({response.headers['location']})")
    match = re.search(r"\((\d+)\)", response.headers["location"])
    if match:
        observation_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

It is possible to create an entities, and its relations, in one POST, by giving the full related entity in the JSON instead of only the entitiy id.

body = {
    "unitOfMeasurement": {"name": "", "symbol": "RSSI", "definition": ""},
    "description": "",
    "name": f"{prefix}RSSI_FIU_VAL",
    "observationType": "",
    "ObservedProperty": { "@iot.id": observed_property_id},
    "Sensor": {"@iot.id": sensor_id},
    "Thing": {"@iot.id": thing_id},
    "Observations": [
        {
            "result": 1,
        },
        {
            "result": 1,
            "FeatureOfInterest": {
                "name": f"{prefix}A weather station 3.",
                "description": "A weather station.",
                "feature": {
                    "type": "Point",
                    "coordinates": [8.956099, 46.172135],
                },
                "encodingType": "application/vnd.geo+json",
            },
        },
        {
            "result": 0,
            "FeatureOfInterest": {
                "name": f"{prefix}A weather station 4.",
                "description": "A weather station.",
                "feature": {
                    "type": "Point",
                    "coordinates": [8.956219, 46.172245],
                },
                "encodingType": "application/vnd.geo+json",
            },
        },
    ],
}

response = requests.post(
    IST_SOS_ENDPOINT + "/Datastreams",
    data=json.dumps(body),
    headers={
        "Content-type": "application/json",
        "Authorization": f"Bearer {token}",
        "Commit-message": "Create Datastrem and related Observations",
    },
)

if response.status_code == 201:
    print(f"Datastream created successfully ({response.headers['location']})")
    match = re.search(r"\((\d+)\)", response.headers["location"])
    if match:
        datastream_id = int(match.group(1))
    else:
        print("No number found in parentheses.")
else:
    result = json.dumps(response.json(), indent=2)
    display(Markdown(f"```json\n{result}\n```"))

Utilizing the SensorThing APIRegister Entities¶