HOWTO: Python interoperability with Snap4City, smart city API

×

Warning message

You can't delete this newsletter because it has not been sent to all its subscribers.

Main elements of the Advanced Smart City APIs are:

·         Smart City API

o   Mainly to get data from the platform

o   To access the KB, Service Map resources and query

o   Several filtering and capabilities, see also section on Dashboards.

o   See this section.

·         Entity Directory (IoT Directory) API

o   To register new devices, pose queries, etc.

·         Orion Broker API, NGSI V2

o   Mainly to post new messages on registered entities

o   To communicate with Orion Brokers exploiting the Secure Filter of Snap4City.

o   See Section IV.C.9

·         MyKPI API

o   To access (read data, and send data) at MyKPI and other personal data safer

·         Heatmap, etc., API

o   To save and access to HeatMaps of the Heatmap server.

o   Similarly, it can access to OD matrices, Traffic Flows, etc.

All the Advanced Smart City API may be managed by an API Management system for accounting and billing if needed to control business, traffic, usage, impose limitations, etc. See: https://www.snap4city.org/827

Each instance of the Snap4City Platform has its own Smart City API, in addition, there is an integrated federated service (called Super) to query all the installations which are federated each other (the federation is decided by each platform owner). Several separated federations may be created as well. Each installation may expose both local Smart City API and Super Smart City API. They have the same syntax but different prefix call.

 

In each call to the API, you can

  • filter by category (nature, Subnature), etc., to obtain a set of devices data and thus also their ServiceURI.
  • Retrieve the data of the single Entity Instance (device)/ServiceURI.
  • Retrieve the data of a set of Entity Instances (devices)/ServiceURIs.

 

For example, on using Smart City APIs: (see training slide part 8th)

Time Series are attached to Devices which are identified by ServiceURI. To access at the Time Series (also called real time data) you can:

  • From Processing Logic (IoT App) use the block «service info dev». In this case, you automatically access to your private and delegated data. You do not need to perform the authentication since it is performed directly from the microservice Processing Logic (IoT App) context, both on cloud and on edge.
  • From Python/Rstudio, Web and Mobile App, you can call Smart City API, see in this section and in Part 7 of the course.
  • (1) Retrieve data from Processing Logic (IoT App) and pass them to Python/Rstudio in JSON as presented in other sections, or (2) pass to them only some parameters such as the GPS location and categories. This approach is viable for small amount of data, such as some thousands. For larger amount of data or to be more efficient we suggest using case (2) which is a direct access to the Smart City API from the Data Analytic.

 

Access to public data does not need to be authenticated, the Smart City API can provide public data without any authentication.

Access to private data from Smart City API can be performed according to the Open ID connect approach. You have to get the Access Token and then perform your API call for query selection as above.

In the following Sections and in the training course sections you can get an example about how to get access to the Private data, and to send data on the platform in the secure manner:

https://www.snap4city.org/download/video/course/da/

https://www.snap4city.org/download/video/course/app/

EXAMPLES:

I.A.1.Development: using Advanced Smart City API in Python

get token

import requests

# this is only an example do not use in production,

# consider that each call it will create a new session on keycloak for each request,

# the token should be reused until it is expired and when expired it can be 'refreshed' using the refresh token

def getTokenViaUserCredentials(username,password):

    payload = {

        'f': 'json',

        'client_id': xxxxx-tool',

        'grant_type': 'password',

        'username': username,

        'password': password

    }

    header = {

        'Content-Type': 'application/x-www-form-urlencoded'

    }

    urlToken = "https://www.snap4city.org/auth/realms/master/protocol/openid-connect/token"

    response = requests.request("POST", urlToken, data=payload, headers=header)

    token = response.json()

    return token

 

def getTokenViaRefreshToken(old_token):

    if not 'refresh_token' in old_token:

        return None

   

    payload = {

        'f': 'json',

        'client_id': xxxxx-tool',

        'grant_type': 'refresh_token',

        'refresh_token': old_token['refresh_token']

    }

    header = {

        'Content-Type': 'application/x-www-form-urlencoded'

    }

    urlToken = "https://www.snap4city.org/auth/realms/master/protocol/openid-connect/token"

    response = requests.request("POST", urlToken, data=payload, headers=header)

    token = response.json()

    return token

 

get historical data via serviceuri

#get the token of the user

token = getTokenViaUserCredentials ('…user…','…password…')

print('token', token)

if 'access_token' in token :

    head = {

                "Content-Type": "application/json",

                "Authorization": f"Bearer {token['access_token']}"

            }

    api_url ="https://<xxxxx>/ServiceMap/api/v1/"

    service_uri = "http://www.disit.org/km4city/resource/iot/<xxxxx>/<xxxxx>/ccccc_Device_01"

    response = requests.request("GET", api_url + "?serviceUri=" + service_uri +"&fromTime=60-day", headers=head)

    print(response.json())

else:

    print('failed getting access token', token)

 

update value of a device

if 'access_token' in token :

    head = {

                "Content-Type": "application/json",

                "Authorization": f"Bearer {token['access_token']}"

            }

    api_url = "https://iot-app.snap4city.org/orion-broker-<xxxxx>/v2/"

    device_id = "ccccc_Device_01"

    payload = {

        "dateObserved": {

            "value": "2023-01-27T01:01:02.000Z",

            "type": "string"

          },

          "FatigueValue": {

            "value": 13.1,

            "type": "number"

      }   

    }

    response = requests.request("PATCH", api_url + "entities/" + device_id +"/attrs?type=Sensor&elementid=" + device_id, json=payload, headers=head)

    if response.status_code!=204 :

        print(response.json())

    else:

        print(device_id+" updated!")

else:

    print('failed getting access token', token)