ucloud.py

# This file is part of the Arduino IoT Cloud Python client.
# Copyright (c) 2022 Arduino SA
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.

import time
import logging
import cbor2
from senml import SenmlPack
from senml import SenmlRecord
from arduino_iot_cloud.umqtt import MQTTClient
import asyncio
from asyncio import CancelledError
try:
    from asyncio import InvalidStateError
except (ImportError, AttributeError):
    # MicroPython doesn't have this exception
    class InvalidStateError(Exception):
        pass
try:
    from arduino_iot_cloud._version import __version__
except (ImportError, AttributeError):
    __version__ = "1.5.0"

# Server/port for basic auth.
_DEFAULT_SERVER = "iot.arduino.cc"

# Default port for cert based auth and basic auth.
_DEFAULT_PORT = (8885, 8884)


class DoneException(Exception):
    pass


def timestamp():
    return int(time.time())


def timestamp_ms():
    return time.time_ns() // 1000000


def log_level_enabled(level):
    return logging.getLogger().isEnabledFor(level)


class ArduinoCloudObject(SenmlRecord):
    def __init__(self, name, **kwargs):
        self.on_read = kwargs.pop("on_read", None)
        self.on_write = kwargs.pop("on_write", None)
        self.on_run = kwargs.pop("on_run", None)
        self.interval = kwargs.pop("interval", 1.0)
        self.backoff = kwargs.pop("backoff", None)
        self.args = kwargs.pop("args", None)
        value = kwargs.pop("value", None)
        if keys := kwargs.pop("keys", {}):
            value = {   # Create a complex object (with sub-records).
                k: ArduinoCloudObject(f"{name}:{k}", value=v, callback=self.senml_callback)
                for (k, v) in {k: kwargs.pop(k, None) for k in keys}.items()
            }
        self._updated = False
        self.on_write_scheduled = False
        self.timestamp = timestamp()
        self.last_poll = timestamp_ms()
        self.runnable = any((self.on_run, self.on_read, self.on_write))
        callback = kwargs.pop("callback", self.senml_callback)
        for key in kwargs:  # kwargs should be empty by now, unless a wrong attr was used.
            raise TypeError(f"'{self.__class__.__name__}' got an unexpected keyword argument '{key}'")
        super().__init__(name, value=value, callback=callback)

    def __repr__(self):
        return f"{self.value}"

    def __contains__(self, key):
        return isinstance(self.value, dict) and key in self._value

    @property
    def updated(self):
        if isinstance(self.value, dict):
            return any(r._updated for r in self.value.values())
        return self._updated

    @updated.setter
    def updated(self, value):
        if isinstance(self.value, dict):
            for r in self.value.values():
                r._updated = value
        self._updated = value

    @property
    def initialized(self):
        if isinstance(self.value, dict):
            return all(r.initialized for r in self.value.values())
        return self.value is not None

    @SenmlRecord.value.setter
    def value(self, value):
        if value is not None:
            if self.value is not None:
                # This is a workaround for the cloud float/int conversion bug.
                if isinstance(self.value, float) and isinstance(value, int):
                    value = float(value)
                if not isinstance(self.value, type(value)):
                    raise TypeError(
                        f"{self.name} set to invalid data type, expected: {type(self.value)} got: {type(value)}"
                    )
            self._updated = True
            self.timestamp = timestamp()
            if log_level_enabled(logging.DEBUG):
                logging.debug(
                    f"%s: {self.name} value: {value} ts: {self.timestamp}"
                    % ("Init" if self.value is None else "Update")
                )
        self._value = value

    def __getattr__(self, attr):
        if isinstance(self.__dict__.get("_value", None), dict) and attr in self._value:
            return self._value[attr].value
        raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{attr}'")

    def __setattr__(self, attr, value):
        if isinstance(self.__dict__.get("_value", None), dict) and attr in self._value:
            self._value[attr].value = value
        else:
            super().__setattr__(attr, value)

    def _build_rec_dict(self, naming_map, appendTo):
        # This function builds a dict of records from a pack, which gets converted to CBOR and
        # pushed to the cloud on the next update.
        if isinstance(self.value, dict):
            for r in self.value.values():
                r._build_rec_dict(naming_map, appendTo)
        else:
            super()._build_rec_dict(naming_map, appendTo)

    def add_to_pack(self, pack, push=False):
        # This function adds records that will be pushed to (or updated from) the cloud, to the SenML pack.
        # NOTE: When pushing records to the cloud (push==True) only fully initialized records are added to
        # the pack. And when updating records from the cloud (push==False), partially initialized records
        # are allowed in the pack, so they can be initialized from the cloud.
        # NOTE: all initialized sub-records are added to the pack whether they changed their state since the
        # last update or not, because the cloud currently does not support partial objects updates.
        if isinstance(self._value, dict):
            if not push or self.initialized:
                for r in self._value.values():
                    pack.add(r)
        elif not push or self._value is not None:
            pack.add(self)
        self.updated = False

    def senml_callback(self, record, **kwargs):
        # This function gets called after a record is updated from the cloud (from_cbor).
        # The updated flag is cleared to avoid sending the same value again to the cloud,
        # and the on_write function flag is set to so it gets called on the next run.
        self.updated = False
        self.on_write_scheduled = True

    async def run(self, client):
        while True:
            self.run_sync(client)
            await asyncio.sleep(self.interval)
            if self.backoff is not None:
                self.interval = min(self.interval * self.backoff, 5.0)

    def run_sync(self, client):
        if self.on_run is not None:
            self.on_run(client, self.args)
        if self.on_read is not None:
            self.value = self.on_read(client)
        if self.on_write is not None and self.on_write_scheduled:
            self.on_write_scheduled = False
            self.on_write(client, self if isinstance(self.value, dict) else self.value)


class ArduinoCloudClient:
    def __init__(
            self,
            device_id,
            username=None,
            password=None,
            ssl_params={},
            server=None,
            port=None,
            keepalive=10,
            ntp_server="pool.ntp.org",
            ntp_timeout=3,
            sync_mode=False
    ):
        self.tasks = {}
        self.records = {}
        self.thing_id = None
        self.keepalive = keepalive
        self.last_ping = timestamp()
        self.senmlpack = SenmlPack("", self.senml_generic_callback)
        self.ntp_server = ntp_server
        self.ntp_timeout = ntp_timeout
        self.async_mode = not sync_mode
        self.connected = False

        # Convert args to bytes if they are passed as strings.
        if isinstance(device_id, str):
            device_id = bytes(device_id, "utf-8")
        if username is not None and isinstance(username, str):
            username = bytes(username, "utf-8")
        if password is not None and isinstance(password, str):
            password = bytes(password, "utf-8")

        self.device_topic = b"/a/d/" + device_id + b"/e/i"
        self.command_topic = b"/a/d/" + device_id + b"/c/up"

        # Update RTC from NTP server on MicroPython.
        self.update_systime()

        # If no server/port were passed in args, set the default server/port
        # based on authentication type.
        if server is None:
            server = _DEFAULT_SERVER
        if port is None:
            port = _DEFAULT_PORT[0] if password is None else _DEFAULT_PORT[1]

        # Create MQTT client.
        self.mqtt = MQTTClient(
            device_id, server, port, ssl_params, username, password, keepalive, self.mqtt_callback
        )

        # Add internal objects initialized by the cloud.
        for name in ["thing_id", "tz_offset", "tz_dst_until"]:
            self.register(name, value=None)

    def __getitem__(self, key):
        if isinstance(self.records[key].value, dict):
            return self.records[key]
        return self.records[key].value

    def __setitem__(self, key, value):
        self.records[key].value = value

    def __contains__(self, key):
        return key in self.records

    def get(self, key, default=None):
        if key in self and self[key] is not None:
            return self[key]
        return default

    def update_systime(self, server=None, timeout=None):
        try:
            import ntptime
            ntptime.host = self.ntp_server if server is None else server
            ntptime.timeout = self.ntp_timeout if timeout is None else timeout
            ntptime.settime()
            logging.info("RTC time set from NTP.")
        except ImportError:
            pass    # No ntptime module.
        except Exception as e:
            if log_level_enabled(logging.ERROR):
                logging.error(f"Failed to set RTC time from NTP: {e}.")

    def create_task(self, name, coro, *args, **kwargs):
        if callable(coro):
            coro = coro(*args)
        try:
            asyncio.get_event_loop()
            self.tasks[name] = asyncio.create_task(coro)
            if log_level_enabled(logging.INFO):
                logging.info(f"task: {name} created.")
        except Exception:
            # Defer task creation until there's a running event loop.
            self.tasks[name] = coro

    def create_topic(self, topic, inout):
        return bytes(f"/a/t/{self.thing_id}/{topic}/{inout}", "utf-8")

    def register(self, aiotobj, coro=None, **kwargs):
        if isinstance(aiotobj, str):
            if kwargs.get("value", None) is None and kwargs.get("on_read", None) is not None:
                kwargs["value"] = kwargs.get("on_read")(self)
            aiotobj = ArduinoCloudObject(aiotobj, **kwargs)

        # Register the ArduinoCloudObject
        self.records[aiotobj.name] = aiotobj

        # Check if object needs to be initialized from the cloud.
        if not aiotobj.initialized and "r:m" not in self.records:
            self.register("r:m", value="getLastValues")

        # Create a task for this object if it has any callbacks.
        if self.async_mode and aiotobj.runnable:
            self.create_task(aiotobj.name, aiotobj.run, self)

    def senml_generic_callback(self, record, **kwargs):
        # This callback catches all unknown/umatched sub/records that were not part of the pack.
        rname, sname = record.name.split(":") if ":" in record.name else [record.name, None]
        if rname in self.records:
            if log_level_enabled(logging.INFO):
                logging.info(f"Ignoring cloud initialization for record: {record.name}")
        else:
            if log_level_enabled(logging.WARNING):
                logging.warning(f"Unkown record found: {record.name} value: {record.value}")

    def mqtt_callback(self, topic, message):
        if log_level_enabled(logging.DEBUG):
            logging.debug(f"mqtt topic: {topic[-8:]}... message: {message[:8]}...")
        self.senmlpack.clear()
        for record in self.records.values():
            # If the object is uninitialized, updates are always allowed even if it's a read-only
            # object. Otherwise, for initialized objects, updates are only allowed if the object
            # is writable (on_write function is set) and the value is received from the out topic.
            if not record.initialized or (record.on_write is not None and b"shadow" not in topic):
                record.add_to_pack(self.senmlpack)
        self.senmlpack.from_cbor(message)
        self.senmlpack.clear()

    def ts_expired(self, ts, last_ts_ms, interval_s):
        return last_ts_ms == 0 or (ts - last_ts_ms) > int(interval_s * 1000)

    def poll_records(self):
        ts = timestamp_ms()
        try:
            for record in self.records.values():
                if record.runnable and self.ts_expired(ts, record.last_poll, record.interval):
                    record.run_sync(self)
                    record.last_poll = ts
        except Exception as e:
            self.records.pop(record.name)
            if log_level_enabled(logging.ERROR):
                logging.error(f"task: {record.name} raised exception: {str(e)}.")

    def poll_connect(self, aiot=None, args=None):
        logging.info("Connecting to Arduino IoT cloud...")
        try:
            self.mqtt.connect()
        except Exception as e:
            if log_level_enabled(logging.WARNING):
                logging.warning(f"Connection failed {e}, retrying...")
            return

        if self.thing_id is None:
            self.mqtt.subscribe(self.device_topic, qos=1)
        else:
            self.mqtt.subscribe(self.create_topic("e", "i"))

        if self.async_mode:
            if self.thing_id is None:
                self.register("discovery", on_run=self.poll_discovery, interval=0.500)
            self.register("mqtt_task", on_run=self.poll_mqtt, interval=1.0)
            raise DoneException()
        self.connected = True

    def poll_discovery(self, aiot=None, args=None):
        self.mqtt.check_msg()
        if self.records.get("thing_id").value is not None:
            self.thing_id = self.records.pop("thing_id").value
            if not self.thing_id:  # Empty thing ID should not happen.
                raise Exception("Device is not linked to a Thing ID.")

            self.topic_out = self.create_topic("e", "o")
            self.mqtt.subscribe(self.create_topic("e", "i"))

            if lastval_record := self.records.pop("r:m", None):
                lastval_record.add_to_pack(self.senmlpack)
                self.mqtt.subscribe(self.create_topic("shadow", "i"), qos=1)
                self.mqtt.publish(self.create_topic("shadow", "o"), self.senmlpack.to_cbor(), qos=1)

            if hasattr(cbor2, "dumps"):
                # Push library version and mode.
                libv = "%s-%s" % (__version__, "async" if self.async_mode else "sync")
                # Note we have to add the tag manually because python-ecosys's cbor2 doesn't suppor CBORTags.
                self.mqtt.publish(self.command_topic, b"\xda\x00\x01\x07\x00" + cbor2.dumps([libv]), qos=1)
            logging.info("Device configured via discovery protocol.")
            if self.async_mode:
                raise DoneException()

    def poll_mqtt(self, aiot=None, args=None):
        self.mqtt.check_msg()
        if self.thing_id is not None:
            self.senmlpack.clear()
            for record in self.records.values():
                if record.updated:
                    record.add_to_pack(self.senmlpack, push=True)
            if len(self.senmlpack._data):
                logging.debug("Pushing records to Arduino IoT cloud:")
                if log_level_enabled(logging.DEBUG):
                    for record in self.senmlpack._data:
                        logging.debug(f"  ==> record: {record.name} value: {str(record.value)[:48]}...")
                self.mqtt.publish(self.topic_out, self.senmlpack.to_cbor(), qos=1)
                self.last_ping = timestamp()
            elif self.keepalive and (timestamp() - self.last_ping) > self.keepalive:
                self.mqtt.ping()
                self.last_ping = timestamp()
                logging.debug("No records to push, sent a ping request.")

    async def run(self, interval, backoff):
        # Creates tasks from coros here manually before calling
        # gather, so we can keep track of tasks in self.tasks dict.
        for name, coro in self.tasks.items():
            self.create_task(name, coro)

        # Create connection task.
        self.register("connection_task", on_run=self.poll_connect, interval=interval, backoff=backoff)

        while True:
            task_except = None
            try:
                await asyncio.gather(*self.tasks.values(), return_exceptions=False)
                break   # All tasks are done, not likely.
            except KeyboardInterrupt as e:
                raise e
            except Exception as e:
                task_except = e
                pass    # import traceback; traceback.print_exc()

            for name in list(self.tasks):
                task = self.tasks[name]
                try:
                    if task.done():
                        self.tasks.pop(name)
                        self.records.pop(name, None)
                        if isinstance(task_except, DoneException) and log_level_enabled(logging.INFO):
                            logging.info(f"task: {name} complete.")
                        elif task_except is not None and log_level_enabled(logging.ERROR):
                            logging.error(f"task: {name} raised exception: {str(task_except)}.")
                        if name == "mqtt_task":
                            self.register(
                                "connection_task",
                                on_run=self.poll_connect,
                                interval=interval,
                                backoff=backoff
                            )
                        break   # Break after the first task is removed.
                except (CancelledError, InvalidStateError):
                    pass

    def start(self, interval=1.0, backoff=1.2):
        if self.async_mode:
            asyncio.run(self.run(interval, backoff))
            return

        last_conn_ms = 0
        last_disc_ms = 0

        while True:
            ts = timestamp_ms()
            if not self.connected and self.ts_expired(ts, last_conn_ms, interval):
                self.poll_connect()
                if last_conn_ms != 0:
                    interval = min(interval * backoff, 5.0)
                last_conn_ms = ts

            if self.connected and self.thing_id is None and self.ts_expired(ts, last_disc_ms, 0.250):
                self.poll_discovery()
                last_disc_ms = ts

            if self.connected and self.thing_id is not None:
                break
            self.poll_records()

    def update(self):
        if self.async_mode:
            raise RuntimeError("This function can't be called in asyncio mode.")

        if not self.connected:
            try:
                self.start()
            except Exception as e:
                raise e

        self.poll_records()

        try:
            self.poll_mqtt()
        except Exception as e:
            self.connected = False
            if log_level_enabled(logging.WARNING):
                logging.warning(f"Connection lost {e}")