app.py

#!/usr/bin/env python3

"""
Listens to LoRa platform POSTS (sensor readings), reformats data, and forwards to IoT platform.

In simple words this module:
- Initializes Device entities
- Updates Device and Flowerbed entities' attributes.

# datetime for now should be datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z")
"""
import re
import os
import json
import base64
import time
import dateutil.parser
import datetime

from geojson import Point
from flask import Flask, jsonify
from flask import request
import requests as client_request

app = Flask(__name__, )
print("Starting to collect..")

# get config from environment
COLLECTION_DIR = os.getenv('COLLECTION_DIR', default='.')
URL_BASE = 'http://{}:1026'.format(os.getenv('ORION_HOST', default='localhost'))

LORA_PLATFORM_URL = os.getenv('LORA_PLATFORM_URL')
LORA_PLATFORM_SECRET_FILE = os.getenv('LORA_PLATFORM_SECRET_FILE', 'secret_carouge_lora')

with open(LORA_PLATFORM_SECRET_FILE, 'r') as file:
    secret = file.read().replace('\n', '')

assert secret, 'No auth token defined for LoRa platform API'

# AUTH Header for lora platform
h = {'Authorization': secret}

# URN schema for Device entities
URN_BASE_DEVICE_ENTITY = "urn:ngsi-ld:Device:Device-"
URN_BASE_FLOWERBED_ENTITY = "urn:ngsi-ld:FlowerBed:FlowerBed-"

# build header for POST
http_header_post = {
    'Fiware-Service': 'carouge',
    'Content-Type': 'application/json',
    'Accept': 'application/json',
}

# build header for GET
http_header_get = {
    'Fiware-Service': 'carouge',
    'Accept': 'application/json',
}


def get_ngsiv2_typed_description(val):
    if isinstance(val, Point):
        return {'type': 'geo:json', 'value': dict(val)}
    elif isinstance(val, int) or isinstance(val, float):
        return {'type': 'Number', 'value': val}
    elif isinstance(val, str):
        pat = r'[<>"\'=;()]'  # see https://fiware-orion.readthedocs.io/en/master/user/forbidden_characters/index.html
        r = re.compile(pat)
        if r.search(val) is None:
            new_val = val
        else:
            new_val = re.sub(pat, '', val)
            app.logger.warning("Special char(s) deleted. This <{}> turned into this <{}>".format(val, new_val))
        return {'type': 'Text', 'value': new_val}
    elif isinstance(val, datetime.datetime):
        # replace +00:00 is the same as Z but Orion doesnt like it :/
        return {'type': 'DateTime', 'value': str(val.isoformat()).replace("+00:00", "Z")}
    elif isinstance(val, dict):
        return {'value': val}  # bypassed
    elif isinstance(val, list):
        return {'value': val}  # bypassed
    elif isinstance(val, type(None)):
        return {'type': 'Text', 'value': None}
    else:
        raise NotImplementedError("Couldn't get NGSIv2 type for object {} of type {}".format(val, type(val)))


# = = = = = = = = API HANDLERS = = = = = = = =  = = =

class InvalidUsage(Exception):
    status_code = 400

    def __init__(self, message, status_code=None, payload=None):
        Exception.__init__(self)
        self.message = message
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload

    def to_dict(self):
        rv = dict(self.payload or ())
        rv['message'] = self.message
        return rv


@app.errorhandler(InvalidUsage)
def handle_invalid_usage(error):
    response = jsonify(error.to_dict())
    response.status_code = error.status_code
    return response


@app.route('/')
@app.route('/healthcheck')
def healthcheck():
    return 'This service is up and running!'


# = = = = = = = = MOISTURE SENSORS AUX FUNCTIONS = = = = = = = =

def _log_level_of_moisture(cb, sensor):
    """ Recommended levels for standard plants:
        0-10 Centibars = Saturated soil
        10-30 Centibars = Soil is adequately wet (except coarse sands, which are drying)
        30-60 Centibars = Usual range for irrigation (most soils)
        60-100 Centibars = Usual range for irrigation in heavy clay
        100-200 Centibars = Soil is becoming dangerously dry- proceed with caution!

    :param cb:
    :param sensor:
    :return:
    """

    if cb < 10:
        app.logger.warning("{} indicates soil is saturated wet: {} cb!".format(sensor, cb))
    elif cb < 30:
        app.logger.info("{} indicates soil is adequately wet: {} cb".format(sensor, cb))
    elif cb < 60:
        app.logger.info("{} indicates soil in usual range of irrigation for moist soils: {} cb".format(sensor, cb))
    elif cb < 100:
        app.logger.warning("{} indicates soil needs irrigation: {} cb".format(sensor, cb))
    elif cb < 200:
        app.logger.warning("{} indicates soil is dangerously dry: {} cb".format(sensor, cb))
    else:
        app.logger.error("Measurement not in expected range: {} cb".format(cb))


def _log_level_of_battery_level(bat, sensor):

    if bat < 0.5:
        app.logger.warning("{} battery level is critically low: {} %!".format(sensor, bat))
    else:
        app.logger.info("{} battery level is: {} %".format(sensor, bat))


def _conv_micro_volts_to_moiture_cb(micro_volts):
    """
    Sensor Voltage range is from 0V to 3V.
    This voltage range is linearly dependent to the real value of the sensor, which is expressed in cb (kPa):

    - 0V = Plants are wet - do not need watering (0cb)
    - 3V = Plant are dry - need watering (239cb)

    :param micro_volts:
    :return: pressure (in cb ~ KPa)

    """
    assert type(micro_volts) is int

    if micro_volts == 0:
        return 0

    return int((micro_volts * 239) / 3000000)


def _conv_micro_volts_to_battery_level(micro_volts):
    """Battery is 0-12v
    but is connected with a Voltage divider R1=R2= 5 K ohm
    https://en.wikipedia.org/wiki/Voltage_divider

    - measured voltage is half on the real value
    - measured voltage is in micro volts

    hence:
        bat_percetage = 100 * (v_input /1.000.000 ) * 2 / 12

    :param micro_volts:
    :return: battery percentage 0-100%

    """
    assert type(micro_volts) is int

    if micro_volts == 0:
        return 0

    return 2 * (micro_volts / 1000000) / 12


def _get_reading(message_data):
    """ Lora Bridge device has to channels (A and B).
    channel A is connected to moisture sensors
    channel B is connected to battery voltage output

    To read correctly the value from the channel A, you should take the byte 3 to 5 (included).
    To read correctly the value from the channel B, you should take the byte 6 to 8 (included).
    Then, you should decoded from Hex to Decimal.
    IMPORTANT: The value is expressed in µV if sensor is configured as voltage input (1V = 1.000.000µV)
    For more info look at LoRaWAN 863-870 ANALOG PWR - Guide utilisateur / User guide version V2.0.2


    >>> base64.standard_b64decode("QsABAqD7AVL5yQ==").hex()
    '42c00102a0fb0152f9c9'

    # deplexing data:
    # 42 -> it's a data frame!
    # c0 ->
    # 01 -> configured as 0-10v input for channel A
    # 02a0fb -> data channel A
    # 01 -> configured as 0-10v input for channel A
    # 52f9c9 -> data channel B

    :param message_data:
    :return: moisture, battery_level
    """

    reading_hex = base64.standard_b64decode(message_data).hex()

    if reading_hex[0:2] == '42':  # it's a dataFrame

        moisture = _conv_micro_volts_to_moiture_cb(int(reading_hex[6:12], 16))
        battery_level = _conv_micro_volts_to_battery_level(int(reading_hex[14:20], 16))
        return moisture, battery_level

    else:
        msg = 'Not a sensor reading (expected 0x42) {}'.format(reading_hex)
        app.logger.warning(msg)
        raise InvalidUsage(message=msg)


# = = = = = = = = CONTEXT BROKER FUNCTIONS = = = = = = = =
def _update_flowerbed_entity(soil_moisture, sensor_eui):
    flowerbed_id = None

    # check if refDevice attr defined for any FlowerBed
    url = "{}/v2/entities/?q=refDevice=={}&type=FlowerBed".format(URL_BASE, sensor_eui)
    r = client_request.get(
        url=url,
        headers=http_header_get
    )
    if r.ok and r.json():
        flowerbed_id = r.json()[0]['id']

    if not flowerbed_id:
        msg = 'No FlowerBed associated to sensor {}'.format(sensor_eui)
        app.logger.warning(msg)
        return msg

    app.logger.debug('Found FlowerBed {} associated to sensor {}'.format(flowerbed_id, sensor_eui))

    body = {"soilMoisture": get_ngsiv2_typed_description(soil_moisture)}
    url = "{}/v2/entities/{}/attrs".format(URL_BASE, flowerbed_id)
    r = client_request.patch(
        url=url,
        headers=http_header_post,
        json=body
    )

    app.logger.debug('Sending data to CTX broker: {}'.format(body))

    if not (200 <= r.status_code < 300):
        app.logger.error('HTTP status code not 2xx, something went wrong..')
        app.logger.error(r.status_code)
        app.logger.error(r.reason)
        app.logger.error(r.json())
        return "Got error {}".format(r.reason)
    else:
        app.logger.info('Entity updated (code {}), url {}'.format(r.status_code, url))
        return "ok"


def _update_device_entity(deveui, comment, longitude, latitude, last_reception, dataFrame,
                          dataFrame_rssi, dataFrame_timestamp):
    """ Creates/Updates LoRa devs into ctx broker

    Output Data model:
        -id (static)
        -type (static)
        -serialNumber (static)
        -controlledProperty (static)
        -owner (static)
        -source (static)
        -description
        -name
        -dateLastValueReported
        -value
        -location
        -rssi
        -batteryLevel

    NOTES:
        name is regexed from LoRa platform <comment> field of device,
        e.g. This device is below the big tree next to the tram station {device_name}=dev_2 , note by Maurizio!"
        incurs into having name="dev_2"

    """
    assert deveui

    body = {}

    app.logger.info('Creating/Updating devices in CTX broker for deveui {}'.format(deveui))

    # static values
    body.update({'id': '{}{}'.format(URN_BASE_DEVICE_ENTITY, deveui[-4:])})
    body.update({'type': 'Device'})
    body.update({'serialNumber': get_ngsiv2_typed_description(deveui)})
    body.update({'controlledProperty': get_ngsiv2_typed_description(['soil_moisture'])})
    body.update({'owner': get_ngsiv2_typed_description('Carouge')})
    body.update({'source': get_ngsiv2_typed_description(LORA_PLATFORM_URL)})

    # dynamic/real-time values
    if longitude and latitude:
        body.update({'location': get_ngsiv2_typed_description(Point([longitude, latitude]))})

    if comment:
        body.update({'description': get_ngsiv2_typed_description(comment)})

        # parse device name hidden inside comment field
        pattern = r'{device_name}=([^\s]+)'
        result = re.search(pattern, comment)
        if result:
            body.update({'name': get_ngsiv2_typed_description(result.groups()[0])})

    if last_reception or dataFrame_timestamp:
        in_date = None
        try:
            in_date = dateutil.parser.isoparse(last_reception)
            assert in_date and isinstance(in_date, datetime.datetime), \
                'Didnt get expected type from LoRa platform for last_reception'
        except TypeError:
            pass
        try:
            in_date = dateutil.parser.isoparse(dataFrame_timestamp)
            assert in_date and isinstance(in_date, datetime.datetime), \
                'Didnt get expected type from LoRa platform for dataFrame_timestamp'
        except TypeError:
            pass
        if in_date:
            body.update({'dateLastValueReported': get_ngsiv2_typed_description(in_date)})
        else:
            app.logger.warning("No date parsed for updaging Device field: dateLastValueReported")

    if dataFrame:
        moist, bat = _get_reading(dataFrame)
        body.update({'value': get_ngsiv2_typed_description(moist)})
        body.update({'batteryLevel': get_ngsiv2_typed_description(bat)})

    if dataFrame_rssi:
        body.update({'rssi': get_ngsiv2_typed_description(dataFrame_rssi)})

    app.logger.debug('Sending data to CTX broker: {}'.format(body))

    url = "{}/v2/entities/?options=upsert".format(URL_BASE)
    r = client_request.post(
        url=url,
        json=body,
        headers=http_header_post
    )

    if not (200 <= r.status_code < 300):
        app.logger.error('HTTP status code not 2xx, something went wrong..')
        app.logger.error(r.status_code)
        app.logger.error(r.reason)
        app.logger.error(r.json())
        return "Got error {}".format(r.reason)
    else:
        app.logger.info('Entity updated (code {}), url {}'.format(r.status_code, url))
        return "ok"


# = = = = = = = = OTHER AUX FUNCTIONS = = = = = = = =
def _dump_raw_data_to_filesystem():
    app.logger.info('dumping request to FS')
    try:
        filepath = os.path.join(COLLECTION_DIR, '{}.json'.format(int(time.time())))

        with open(filepath, encoding='utf-8', mode='w') as f:
            json.dump(
                obj=request.json,
                fp=f
            )

    except Exception as e:
        app.logger.error(e)
        return 'error'

    app.logger.info("Dumped raw data at file {}".format(filepath))
    return 'ok'


def _dummy_post_handler():
    if request.data:
        app.logger.info("Request data: %s" % request.data)
    if request.form:
        app.logger.info("Request form: %s" % request.form)

    return 'Got POST with json %s' % request.json


# = = = = = = = = FLASK ROUTES = = = = = = = =
@app.route('/dca-carouge-watering-sensed-data', methods=['POST'])
def post_dca_carouge_watering_sensed_data():
    return _dummy_post_handler()


@app.route('/dca-carouge-watering-sensed-data/test ', methods=['POST'])
def post_dca_carouge_watering_sensed_data_test():
    return _dummy_post_handler()


@app.route('/dca-carouge-watering-sensed-data/rest/callback/payloads/ul', methods=['POST'])
def post_dca_carouge_watering_sensed_data_payload():
    # get values

    soil_humidity, battery_level = _get_reading(request.json['dataFrame'])
    sensor_eui = request.json['deveui']

    # log reading
    _log_level_of_moisture(soil_humidity, sensor_eui)
    _log_level_of_battery_level(battery_level,sensor_eui)

    # saves locally and pushed data to FlowerBed and Device entities
    resp = {}
    resp.update({"fs_dump": _dump_raw_data_to_filesystem()})
    resp.update({"update_flowerbed_entities": _update_flowerbed_entity(
        soil_humidity,
        sensor_eui
    )})
    resp.update({"update_device_entities": _update_device_entity(
        deveui=request.json['deveui'],
        comment=None,
        longitude=None,
        latitude=None,
        last_reception=None,
        dataFrame=request.json['dataFrame'],
        dataFrame_rssi=request.json['rssi'],
        dataFrame_timestamp=request.json['timestamp']
    )})

    return resp


@app.route('/dca-carouge-watering-sensed-data/rest/callback/nodeinfo', methods=['PUT'])
def put_dca_carouge_watering_sensed_node_info():
    _dummy_post_handler()
    raise InvalidUsage(message='This is still not implemented', payload=request.json)


@app.route('/dca-carouge-watering-sensed-data', methods=['GET'])
def get_dca_carouge_watering_sensed_data():
    raise InvalidUsage(message='This is still not implemented')


@app.route('/dca-carouge-watering-sensed-data/rest/callback/payloads/ul', methods=['GET'])
def get_dca_carouge_watering_sensed_data_callback():
    raise InvalidUsage(message='This is still not implemented')


# = = = = = = = = MAIN = = = = = = = =
def init():
    # initialization of devices info in CTX broker

    # pull data from LoRa platform
    lora_nodes_info = client_request.get(url=LORA_PLATFORM_URL + '/rest/nodes', headers=h).json()

    # push data to CTX broker
    for i in lora_nodes_info:
        _update_device_entity(
            deveui=i['deveui'],
            comment=i['comment'],
            longitude=i['longitude'],
            latitude=i['latitude'],
            last_reception=i['last_reception'],
            dataFrame=None,
            dataFrame_rssi=None,
            dataFrame_timestamp=None
        )


def main():
    app.run(host="0.0.0.0", debug=True, port=80)


if __name__ == "__main__":
    init()
    main()