Live Objects - complete developer guide

Use case 3: air quality monitoring

Pre-requisites:

the event processing feature is enabled for the tenant.
the tenant has a valid Live Objects API key.

Use case description :

Monitor 2 pollutants (NO2 and PM10)
Trigger INFO or ALERT events when thresholds are reached.
Trigger daily pollution level state change events for each pollutant.

This example includes SIMPLE EVENT PROCESSING rules and STATE PROCESSING rules.

The REST queries for this example are available here and can be run in Postman.

Air quality information is available for every monitoring station in a city. 3 different types of message are available:

hourly pollution level for each pollutant (data message sent every hour).
pollution level for the last 3 hours for each pollutant (data message sent every hour).
daily average level for each pollutant (data message once a day at 0 a.m.).

Information/Alert thresholds are defined for each pollutant type.

Daily pollution level.

For NO2, the threshold to trigger the ALERT is lower if the daily state for the previous day is MEDIUM or HIGH. The daily calculated state for NO2 must be stored by your application in the tenant context. Example:

PUT liveobjects.orange-business.com/api/v0/eventprocessing/context/no2-previous-day-medium-level-reached

{
  "contextData": true,
  "tags": [
    "previous day"
  ]
}

Event triggering on air quality:

6-hour INFO: when the information level is reached in a monitoring station for NO2 or PM10. Then, wait for 6 hours before getting any new "information level reached" event.
real-time ALERT
- when the alert level is reached in a monitoring station for NO2 or PM10.
- when the information level is reached in a monitoring station for NO2 and the daily pollution level for previous day was MEDIUM or HIGH
daily pollution level:when the daily pollution level changes, like for example: LOW→MEDIUM or MEDIUM→HIGH

Streams of messages

A stream of data messages is attached to a monitoring station. The messages from the "paris-centre" monitoring station will be sent in a distinct stream from the "place de l’Opéra" monitoring station.

Data messages

hourly pollution level (sent every hour)

{
   "streamId": "paris-centre-hourly",
    "location": {
        "lon":2.2945, "lat": 48.8584
   },
   "model": "model_hourly",
   "value": {
      "type":"hourly",
      "NO2":450,
      "PM10":17,
      "monitoring-station":"paris-centre"
   },
   "timestamp": "2017-07-27T13:00:00Z"
}

hourly pollution level for the last 3 hours (sent every hour)

{
   "streamId":  "paris-centre-last-3-hours",
    "location": {
        "lon":2.2945, "lat": 48.8584
   },
   "model": "model_last_three_hours",
   "value": {
      "type":"last_three_hours",
      "data1": {"value":{"NO2":420,"PM10":16},"timestamp":"2017-07-27T11:00:00Z"},
      "data2": {"value": {"NO2":401,"PM10":14},"timestamp":"2017-07-27T12:00:00Z"},
      "data3": {"value": {"NO2":450,"PM10":17},"timestamp":"2017-07-27T13:00:00Z"},
      "monitoring-station":"paris-centre"
   },
   "timestamp":"2017-07-27T13:00:00Z"
}

daily average (sent once a day)

{
   "streamId": "paris-centre-daily",
    "location": {
        "lon":2.2945, "lat": 48.8584
   },
   "model": "model_daily",
   "value": {
      "type":"daily",
      "avg-NO2":250,
      "avg-PM10":53,
      "monitoring-station":"paris-centre"
   },
   "timestamp":"2017-07-28T00:00:00Z"
}

Test steps

Step1: Context provisioning

Create the information/alert levels for each pollutant.

PUT liveobjects.orange-business.com/api/v0/eventprocessing/context/no2-alert-threshold-1

{
  "contextData": 400,
  "tags": [
    "threshold","alert","no2"
  ]
}

For the provisioning of the other thresholds, please check the postman requests.

Step2: State processing rule provisioning

Once the threshold levels are provisioned in the context, the rules need to be provisioned:
- For events on exceeding thresholds, use Simple Event Processing (matching rules to check if an event should be triggered + firing rule for the frequency of triggering).
- For events on daily pollution state change, use State Processing.

Before provisioning the state processing rule, it is useful to run the state processing function on a test data message.

POST liveobjects.orange-business.com/api/v0/eventprocessing/stateprocessing-rule/test

{
  "currentState": {},
  "data": {
    "metadata": {
      "connector": "http"
      },
    "streamId": "paris-centre-daily",
    "location": {
        "lon":2.2945, "lat": 48.8584
    },
    "model": "model_daily",
    "value": {
      "type":"daily",
      "avg-NO2":9,
      "avg-PM10":9,
      "monitoring-station":"paris-centre"
    },
    "timestamp":"2017-07-27T00:00:00Z"
  },
  "stateProcessingFunction": {
    "if": [
      {
        "and":
          [
            { "<": [
              { "var": "value.avg-NO2"},
              {"ctx": "no2-alert-threshold-2"}
            ]},
            {"==": [{ "var": "value.type"}, "daily"]}
          ]
      },
      "LOW",
      {
        "and":
          [
            { "<": [
              {"ctx": "no2-alert-threshold-2"},
              { "var": "value.avg-NO2"},
              {"ctx": "no2-alert-threshold-1"}
            ]},
            {"==": [{ "var": "value.type"}, "daily"]}
          ]
        },
      "MEDIUM",
      {
        "and":
          [
            { ">": [
              { "var": "value.avg-NO2"},
              {"ctx": "no2-alert-threshold-1"}
            ]},
            {"==": [{ "var": "value.type"}, "daily"]}
          ]
      },
      "HIGH"
    ]
  }
}

Response:

{
    "stateFunctionValid": true,
    "dataValid": true,
    "stateFunctionResult": "LOW"
}

The test endpoint expects, as input, the current state, a data message and the state function. The response returns the function status (valid or not), the data status (valid or not) and the result of the state function applied to the data.

The threshold values are retrieved from the tenant context (ctx).

Now that the state function is tested, you can provision the state processing rule.

Daily pollution state processing rule:

POST liveobjects.orange-business.com/api/v0/eventprocessing/stateprocessing-rule

{
  "name": "NO2 daily pollution level",
  "enabled": true,
  "stateFunction": {
    "if": [
      {
        "and":
          [
            { "<": [
              { "var": "value.avg-NO2"},
              {"ctx": "no2-alert-threshold-2"}
            ]},
            {"==": [{ "var": "value.type"}, "daily"]}
          ]
      },
      "LOW",
      {
        "and":
          [
            { "<": [
              {"ctx": "no2-alert-threshold-2"},
              { "var": "value.avg-NO2"},
              {"ctx": "no2-alert-threshold-1"}
            ]},
            {"==": [{ "var": "value.type"}, "daily"]}
          ]
      },
      "MEDIUM",
      {
        "and":
          [
            { ">": [
              { "var": "value.avg-NO2"},
              {"ctx": "no2-alert-threshold-1"}
            ]},
            {"==": [{ "var": "value.type"}, "daily"]}
          ]
      },
      "HIGH"
    ]
  },
  "stateKeyPath": "streamId"
}

For the PM10 pollutant, the process is the same to create the state processing rule.

Step3: matching rule provisioning

Example: for NO2, if the previous day ended with a "MEDIUM" or "HIGH" level, the ALERT threshold level is 200 microgram/m3 instead of 400. When receiving the daily pollution state change event every night, your application must store it in the context (key name in the context: "no2-previous-day-medium-level-reached" and "no2-previous-day-high-level-reached", value: true or false) in order to be used in the real-time alerts.
the previous day state is set in the tenant context every night, by your application, based on the daily state event sent by the state processing rule.

More info on matching rules.

A testing point is available to prepare the matching rule and test it on a data message.

POST liveobjects.orange-business.com/api/v0/eventprocessing/matching-rule/test

{
  "data": {
    "metadata": {
      "connector": "http"
      },
   "streamId": "paris-centre-hourly",
    "location": {
        "lon":2.2945, "lat": 48.8584
   },
   "model": "model_hourly",
   "value": {
      "type":"hourly",
      "NO2":201,
      "PM10":15,
      "monitoring-station":"paris-centre"
   },
   "timestamp":"2017-07-27T02:00:00Z"
  },
  "dataPredicate": {
    "and": [
      {
        ">": [{"var": "value.NO2"}, {"ctx": "no2-alert-threshold-2" }]
      },
      {
        "or": [
          {"==": [{"ctx": "no2-previous-day-medium-level-reached"},true]},
          {"==": [{"ctx": "no2-previous-day-high-level-reached"},true]}
        ]
      },
      {
        "==": [{"var": "value.type"},"hourly"]
      }
    ]
  }
}

Response:

{
    "dataPredicateValid": true,
    "dataValid": true,
    "dataPredicateResult": true
}

Now, provision the matching-rule:

POST liveobjects.orange-business.com/api/v0/eventprocessing/matching-rule

{
  "name": "no2-alert-level-reached-threshold2",
  "dataPredicate": {
    "and": [
      {
        ">": [{"var": "value.NO2"}, {"ctx": "no2-alert-threshold-2" }]
      },
      {
        "or": [
          {"==": [{"ctx": "no2-previous-day-medium-level-reached"},true]},
          {"==": [{"ctx": "no2-previous-day-high-level-reached"},true]}
        ]
      },
      {
        "==": [{"var": "value.type"},"hourly"]
      }
    ]
  },
  "enabled": true
}

Response:

{
  "id": "0476993c-b7cc-49a7-9a86-87431ead76e7",
  "name": "no2-alert-level-reached-threshold2",
  "enabled": true,
  "dataPredicate": {
    "and": [
      {
        ">": [{"var": "value.NO2"}, {"ctx": "no2-alert-threshold-2"}]
      },
      {
        "or": [
          {"==": [{"ctx": "no2-previous-day-medium-level-reached"}, true]},
          {"==": [{"ctx": "no2-previous-day-high-level-reached"}, true]}
        ]
      },
      {
        "==": [{"var": "value.type"}, "hourly"]
      }
    ]
  }
}

Step4: firing rule provisioning

When the matching rule is ready, a firing rule must be provisioned in order to set the frequency for event triggering (ONCE, ALWAYS, SLEEP).
For the matching rule described in the previous step, an event is sent everytime the ALERT threshold is reached in a monitoring station.

More info on firing rules.

POST liveobjects.orange-business.com/api/v0/eventprocessing/firing-rule

{
  "aggregationKeys": [
    "streamId"
  ],
  "enabled": true,
  "firingType": "ALWAYS",
  "matchingRuleIds": [
    "0476993c-b7cc-49a7-9a86-87431ead76e7"
  ],
  "name": "firing NO2 alert 200"
}

Another example of firing rule: for the PM10/NO2 INFO event, for a monitoring station, when an event is triggered, we do not want to receive any other INFO event in the next 6 hours. The firingType is set to SLEEP:

POST liveobjects.orange-business.com/api/v0/eventprocessing/firing-rule

{
  "aggregationKeys": [
    "streamId"
  ],
  "enabled": true,
  "firingType": "SLEEP",
  "matchingRuleIds": [
    "4578993c-b7cc-49a7-9a86-87431ead96a9"
  ],
  "name": "firing PM10 INFO",
    "sleepDuration": "PT6H"
}

The sleepDuration is expressed in a iso8601-duration format.

For the provisioning of the other SEP rules, please check the postman requests.

Step5: Send data messages

In this test, the data messages are sent using Live Objects REST http API.

Data message 1: hourly

POST liveobjects.orange-business.com/api/v0/data/streams/paris-centre-hourly

{
   "location": {
        "lon":2.2945, "lat": 48.8584
   },
   "model": "model_hourly",
   "value": {
      "type":"hourly",
      "NO2": 250,
      "PM10":45,
      "monitoring-station":"paris-centre"
   },
   "timestamp": "2017-07-27T14:00:00Z"
}

Data message 2: last 3 hours

POST liveobjects.orange-business.com/api/v0/data/streams/paris-centre-last-3-hours

{
    "location": {
        "lon":2.2945, "lat": 48.8584
   },
   "model": "model_last_three_hours",
   "value": {
      "type":"last_three_hours",
      "data1": {"value":{"NO2":420,"PM10":16},"timestamp":"2017-07-27T11:00:00Z"},
      "data2": {"value": {"NO2":401,"PM10":14},"timestamp":"2017-07-27T12:00:00Z"},
      "data3": {"value": {"NO2":450,"PM10":17},"timestamp":"2017-07-27T13:00:00Z"},
      "monitoring-station":"paris-centre"
   },
   "timestamp":"2017-07-27T13:00:00Z"
}

Data message 3: daily average

POST liveobjects.orange-business.com/api/v0/data/streams/paris-centre-daily

{
    "location": {
        "lon":2.2945, "lat": 48.8584
   },
   "model": "model_daily",
   "value": {
      "type":"daily",
      "avg-NO2":92,
      "avg-PM10":20,
      "monitoring-station":"paris-centre"
   },
   "timestamp":"2017-07-28T00:00:00Z"
}

In this example, the daily average is calculated in another system. Another option would be to calculate it in a recurrent query on the hourly stream, and then create the daily data message.

POST liveobjects.orange-business.com/api/v0/data/search

{ "size":0,
  "query": {
    "filtered": {
      "filter": {
        "bool":{
          "must": [
            {
              "term": {
                "streamId": "paris-centre-hourly"
              }
            },
            {
              "range": {
                "timestamp": {
                  "gte":"2017-07-27",
                  "lt":"2017-07-28"
                }
              }
            }
          ]
        }
      }
    }
  },
  "aggs": {
    "avg-NO2": {
      "avg": { "field": "@model_hourly.value.NO2" }
    },
    "avg-PM10": {
      "avg": { "field": "@model_hourly.value.PM10" }
    }
  }
}

response:

{
    "took": 124,
    "hits": {
        "total": 1
    },
    "aggregations": {
        "avg-PM10": {
            "value": 41
        },
        "avg-NO2": {
            "value": 450
        }
    }
}

Step6: Get events

The tenant can be notified of the triggered events. The events are also stored in a dedicated stream which can be requested using the Datamanagement data search API.

POST liveobjects.orange-business.com/api/v0/data/search

{
  "from": 0,
  "size": 10,
  "query": {
    "filtered": {
      "filter": {
        "bool":{
          "must": [
            {
              "term": {
                "streamId": "event:paris-centre-hourly"
              }
            },
            {
              "range": {
                "timestamp": {
                  "gte":"2017-08-03",
                  "lt":"2017-08-07"
                }
              }
            }
          ]
        }
      }
    }
  }
}

The response contains the list of events on the hourly stream for the "paris-centre" monitoring station.

{
  "took": 22,
  "hits": {
    "total": 2,
    "hits": [
      {
        "_source": {
          "metadata": null,
          "streamId": "event:paris-centre-hourly",
          "created": "2017-08-07T11:16:01.920Z",
          "location": {
            "provider": null,
            "alt": null,
            "accuracy": null,
            "lon": 2.2945,
            "lat": 48.8584
          },
          "model": "event:model_hourly",
          "id": "59884bf1e9cf83391a49ee61",
          "value": {
            "tenantId": "597f812389179c3436edf332",
            "matchingContext": {
              "matchingRule": {
                "dataPredicate": "{
                  \"and\":[
                    {\">\":[
                      {\"var\":\"value.NO2\"},
                      {\"ctx\":\"no2-info-threshold-1\"}
                    ]},
                    {\"==\":[
                      {\"var\":\"value.type\"},
                      \"hourly\"
                    ]}
                  ]
                }",
                "name": "no2-info-level-reached",
                "id": "84b1cbd7-5184-4460-b05e-41236fbfe770",
                "enabled": true
              },
              "data": {
                "metadata": {
                  "connector": "http"
                },
                "streamId": "paris-centre-hourly",
                "location": {
                  "lon": 2.2945,
                  "lat": 48.8584
                },
                "model": "model_hourly",
                "value": {
                  "NO2": 450,
                  "PM10": 41,
                  "type": "hourly",
                  "monitoring-station": "paris-centre"
                },
                "timestamp": "2017-07-27T14:00:00Z"
              },
              "tenantId": "597f812389179c3436edf332",
              "timestamp": "2017-08-07T11:16:04.859Z"
            },
            "timestamp": "2017-08-07T11:16:04.875Z",
            "firingRule": {
              "name": "firing NO2 INFO",
              "matchingRuleIds": [
                "84b1cbd7-5184-4460-b05e-41236fbfe770"
              ],
              "sleepDuration": "PT6H",
              "id": "f1dfc01d-a236-4bf6-b8a7-fdb2aa6a4e10",
              "aggregationKeys": [
                "streamId"
              ],
              "firingType": "SLEEP",
              "enabled": true
            }
          },
          "timestamp": "2017-08-07T11:16:04.875Z",
          "tags": [
            "event"
          ]
        }
      }
...
}

The events can also be retrieved with MQTT on a specific topic.