Merge pull request #23 from anedyaio/development

Bug Fixes and Commands

Author: anedyasystems

examples/mqtt.py

@@ -1,28 +1,42 @@
 import anedya
 import time
 
-config = anedya.default_config()
-config.connection_mode = anedya.ConnectionMode.MQTT
-config.set_deviceid("<SET-DEVICE-ID>")
-config.set_connection_key("<SET-CONNECTION-KEY>")
+client = None
+tr = None
 
-# Create a client
-client = anedya.AnedyaClient(config)
 
-time.sleep(1)
-# Client is created, now connect with the MQTT server
-client.connect()
-time.sleep(2)
-print(client._mqttclient.is_connected())
+def main():
+    config = anedya.default_config()
+    config.connection_mode = anedya.ConnectionMode.MQTT
+    config.set_deviceid("67719273-7cfe-4726-a846-72ca86340916")
+    config.set_connection_key("7346b841bc8cf7fe39555ae19654612b")
+    config.set_on_command(callback=on_command_callback)
 
-# Publish the datapoint
-data = anedya.DataPoints()
+    # Create a client
+    global client
+    client = anedya.AnedyaClient(config)
 
-dp1 = anedya.FloatData(variable='temperature', timestamp_milli=int(time.time_ns() / 1000000), value=10)
+    time.sleep(1)
+    # Client is created, now connect with the MQTT server
+    client.connect()
+    time.sleep(2)
+    print(client._mqttclient.is_connected())
 
-data.append(dp1)
+    # Publish the datapoint
+    
+    input("Press Enter to continue after sending command...")
+    tr.wait_to_complete()
+    print("Disconnecting")
+    client.disconnect()
 
-client.submit_data(data)
 
-time.sleep(10)
-client.disconnect()
+def on_command_callback(cmdinput: anedya.CommandDetails):
+    print(f"Received command from platform: {cmdinput.command}")
+    # Change the status of the command
+    global tr
+    tr = client.update_command_status(command=cmdinput, status=anedya.CommandStatus.RECEIVED, callback_mode=True)
+    return
+
+
+if __name__ == '__main__':
+    main()

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = anedya-dev-sdk
-version = 0.1.6
+version = 0.1.7
 url = https://github.com/anedyaio/anedya-dev-sdk-python
 author = Anedya Systems
 author_email = support@anedya.io

src/anedya/__init__.py

@@ -1,12 +1,16 @@
 # src/__init__.py
 
 from .config import AnedyaConfig, ConnectionMode, default_config
-from .models import DataPoints, FloatData
+from .models import DataPoints, FloatData, GeoData, CommandDetails
+from .client.commandsUpdate import CommandStatus
 from .anedya import AnedyaClient
 
 __all__ = ['AnedyaConfig',
            'ConnectionMode',
            'default_config',
            'FloatData',
            'DataPoints',
-           'AnedyaClient']
+           'AnedyaClient',
+           'CommandDetails',
+           'GeoData',
+           'CommandStatus']

src/anedya/anedya.py

@@ -105,6 +105,7 @@ def disconnect(self):
     from .client.submitData import submit_data
     from .client.submitLogs import submit_logs
     from .client.commandsUpdate import update_command_status
+    from .client.commandsNext import next_command
     from .client.timeSync import get_time
     from .client.mqttHandlers import _onconnect_handler, _ondisconnect_handler
     from .client.callbacks import _error_callback, _response_callback, _command_callback

src/anedya/client/commandsNext.py

@@ -0,0 +1,127 @@
+from ..models import CommandDetails, AnedyaEncoder
+from ..config import ConnectionMode
+from ..errors import AnedyaTxFailure, AnedyaInvalidConfig
+import uuid
+import json
+import base64
+import random
+import string
+import datetime
+
+
+def next_command(self, timeout: float | None = None) -> tuple[CommandDetails, bool]:
+    """
+    Get the next command from the server.
+    Returns:
+        CommandDetails: The next command.
+
+    Raises:
+        AnedyaTxFailure: If the transaction fails.
+    """
+    if self._config is None:
+        raise AnedyaInvalidConfig('Configuration not provided')
+    if self._config.connection_mode == ConnectionMode.HTTP:
+        return _next_command_http(self, timeout=timeout)
+    elif self._config.connection_mode == ConnectionMode.MQTT:
+        return _next_command_mqtt(self, timeout=timeout)
+    else:
+        raise AnedyaInvalidConfig('Invalid connection mode')
+
+
+def _next_command_http(self, timeout: float | None = None) -> tuple[CommandDetails, bool]:
+    if self._config._testmode:
+        url = "https://device.stageapi.anedya.io/v1/submitData"
+    else:
+        url = self._baseurl + "/v1/submitData"
+    d = _next_command_req("req_" + ''.join(random.choices(string.ascii_letters + string.digits, k=16)))
+    r = self._httpsession.post(url, data=d.encodeJSON(), timeout=timeout)
+    # print(r.json())
+    try:
+        jsonResponse = r.json()
+        payload = json.loads(jsonResponse)
+        if payload['success'] is not True:
+            raise AnedyaTxFailure(payload['error'], payload['errCode'])
+        # Payload has success now create a CommandDetails object and return
+        command = CommandDetails()
+        command.id = uuid.UUID(payload["commandId"])
+        command.command = payload["command"]
+        command.status = payload["status"]
+        command.type = payload["datatype"]
+        if self.type == "string":
+            self.data = payload["data"]
+        elif self.type == "binary":
+            base64_bytes = payload["data"].encode("ascii")
+            data_bytes = base64.b64decode(base64_bytes)
+            self.data = data_bytes
+        command.updated = datetime.datetime.fromtimestamp(payload["updated"] / 1000)
+        command.issued = datetime.datetime.fromtimestamp(payload["issued"] / 1000)
+    except ValueError:
+        raise AnedyaTxFailure(message="Invalid JSON response")
+    return command, payload['nextavailable']
+
+
+def _next_command_mqtt(self, timeout: float | None = None) -> tuple[CommandDetails, bool]:
+    # Create and register a transaction
+    tr = self._transactions.create_transaction()
+    # Encode the payload
+    d = _next_command_req(tr.get_id())
+    payload = d.encodeJSON()
+    # Publish the message
+    print(payload)
+    topic_prefix = "$anedya/device/" + str(self._config._deviceID)
+    print(topic_prefix + "/commands/updateStatus/json")
+    msginfo = self._mqttclient.publish(topic=topic_prefix + "/commands/updateStatus/json",
+                                       payload=payload, qos=1)
+    try:
+        msginfo.wait_for_publish(timeout=timeout)
+    except ValueError:
+        raise AnedyaTxFailure(message="Publish queue full")
+    except RuntimeError as err:
+        raise AnedyaTxFailure(message=str(err))
+    except Exception as err:
+        raise AnedyaTxFailure(message=str(err))
+    # Wait for transaction to complete
+    tr.wait_to_complete()
+    # Transaction completed
+    # Get the data from the transaction
+    data = tr.get_data()
+    # Clear transaction
+    self._transactions.clear_transaction(tr)
+    # Check if transaction is successful or not
+    if data['success'] is not True:
+        raise AnedyaTxFailure(data['error'], data['errCode'])
+    try:
+        if data['available'] is not True:
+            return None, None
+        command = CommandDetails()
+        command.id = uuid.UUID(payload["commandId"])
+        command.command = payload["command"]
+        command.status = payload["status"]
+        command.type = payload["datatype"]
+        if self.type == "string":
+            self.data = payload["data"]
+        elif self.type == "binary":
+            base64_bytes = payload["data"].encode("ascii")
+            data_bytes = base64.b64decode(base64_bytes)
+            self.data = data_bytes
+        command.updated = datetime.datetime.fromtimestamp(payload["updated"] / 1000)
+        command.issued = datetime.datetime.fromtimestamp(payload["issued"] / 1000)
+    except ValueError:
+        raise AnedyaTxFailure(message="Invalid JSON response")
+    return command, payload['nextavailable']
+
+
+class _next_command_req:
+    def __init__(self, req_id: str):
+        self.reqID = req_id
+        return
+
+    def toJSON(self):
+        dict = {
+            "reqId": self.reqID,
+        }
+        return dict
+
+    def encodeJSON(self):
+        data = json.dumps(self, cls=AnedyaEncoder)
+        return data

src/anedya/client/commandsUpdate.py

@@ -3,47 +3,45 @@
 import random
 import base64
 from ..models import CommandDetails, AnedyaEncoder
-from enum import Enum
+from ..transaction import Transaction
 from ..errors import AnedyaInvalidConfig, AnedyaInvalidType, AnedyaTxFailure
 from ..config import ConnectionMode
+from ..models import CommandStatus
 
 
-class CommandStatus(Enum):
-    PENDING = "pending"
-    RECEIVED = "received"
-    PROCESSING = "processing"
-    SUCCESS = "success"
-    FAILURE = "failure"
-    INVALIDATED = "invalidated"
-
-
-def update_command_status(self, command: CommandDetails, status: CommandStatus, ackdata: str | bytes | None = None, acktype: str = "string", timeout: float | None = None) -> None:
+def update_command_status(self, command: CommandDetails, status: CommandStatus, ackdata: str | bytes | None = None, acktype: str = "string", timeout: float | None = None, callback_mode: bool = False) -> None | Transaction:
     """
-    Update status of a command
+    Update status of a command. Not thread safe.
 
     Args:
         command (CommandDetails): Command object of which status needs to be updated
         status (CommandStatus): New status of the command
         ackdata (str | bytes | None, optional): Data to be submitted along with acknowledgement. Maximum 1 kB of data is allowed. Defaults to None.
         acktype (str, optional): Specify the type of data submitted. Defaults to "string".
         timeout (float | None, optional): Time out in seconds for the request. In production setup it is advisable to use a timeout or else your program can get stuck indefinitely. Defaults to None.
+        callback_mode (bool, optional): When using MQTT connection, it is not possible to publish the message from inside a callback. In such scenario, you can set callback_mode to True. Instead of publishing request right away, the function
+        schedules the message to be published after the callback is completed. Also function returns a transaction object. Which can be used to check whether transaction has finished or not and what is the status of transaction.
+        Use this method only when only single transaction is happening at a time. In the case where multiple transactions happening simultaneously, we suggest avoiding call to this function from within callback.Defaults to False.
 
     Raises:
         AnedyaInvalidConfig: Invalid configuration
         AnedyaInvalidType: Invalid datatype is specified
         AnedyaTxFailure: Transaction failure
+
+    Returns:
+        None | Transaction: Returns transaction object if callback_mode is True. returns None otherwise
     """
     if self._config is None:
         raise AnedyaInvalidConfig('Configuration not provided')
     if self._config.connection_mode == ConnectionMode.HTTP:
         return _update_command_status_http(self, command=command, status=status, timeout=timeout)
     elif self._config.connection_mode == ConnectionMode.MQTT:
-        return _update_command_status_mqtt(self, command=command, status=status, timeout=timeout)
+        return _update_command_status_mqtt(self, command=command, status=status, timeout=timeout, callback_mode=callback_mode)
     else:
         raise AnedyaInvalidConfig('Invalid connection mode')
 
 
-def _update_command_status_http(self, command: CommandDetails, status: CommandStatus, ackdata: str | tuple | None = None, acktype: str = "string", timeout: float | None = None) -> None:
+def _update_command_status_http(self, command: CommandDetails, status: CommandStatus, ackdata: str | bytes | None = None, acktype: str = "string", timeout: float | None = None) -> None:
     if self._config._testmode:
         url = "https://device.stageapi.anedya.io/v1/submitData"
     else:
@@ -61,24 +59,29 @@ def _update_command_status_http(self, command: CommandDetails, status: CommandSt
     return
 
 
-def _update_command_status_mqtt(self, command: CommandDetails, status: CommandStatus, ackdata: str | tuple | None = None, acktype: str = "string", timeout: float | None = None) -> None:
+def _update_command_status_mqtt(self, command: CommandDetails, status: CommandStatus, ackdata: str | bytes | None = None, acktype: str = "string", timeout: float | None = None, callback_mode: bool = False) -> None:
     # Create and register a transaction
     tr = self._transactions.create_transaction()
     # Encode the payload
     d = _UpdateCommandStatusReq(tr.get_id(), command=command, status=status, ackdata=ackdata, acktype=acktype)
     payload = d.encodeJSON()
     # Publish the message
-    # print(payload)
+    print(payload)
     topic_prefix = "$anedya/device/" + str(self._config._deviceID)
-    # print(topic_prefix + "/submitData/json")
+    print(topic_prefix + "/commands/updateStatus/json")
     msginfo = self._mqttclient.publish(topic=topic_prefix + "/commands/updateStatus/json",
                                        payload=payload, qos=1)
+    if callback_mode:
+        # Can not block in callback mode
+        return tr
     try:
         msginfo.wait_for_publish(timeout=timeout)
     except ValueError:
         raise AnedyaTxFailure(message="Publish queue full")
     except RuntimeError as err:
         raise AnedyaTxFailure(message=str(err))
+    except Exception as err:
+        raise AnedyaTxFailure(message=str(err))
     # Wait for transaction to complete
     tr.wait_to_complete()
     # Transaction completed
@@ -110,6 +113,9 @@ def __init__(self, reqId: str, command: CommandDetails, status: CommandStatus, a
             self.acktype = "binary"
         else:
             raise AnedyaInvalidType('Invalid acktype')
+        if ackdata is None:
+            self.ackdata = ""
+            self.acktype = "string"
 
     def toJSON(self):
         dict = {

src/anedya/client/mqttHandlers.py

@@ -21,7 +21,7 @@ def _onconnect_handler(self, client, userdata, flags, reason_code, properties):
         self._mqttclient.subscribe(
             topic=topic_prefix + "/response", qos=0)
         self._mqttclient.subscribe(
-            topic=topic_prefix + "/command", qos=0)
+            topic=topic_prefix + "/commands", qos=0)
         # Define all Callbacks for error and response
         # Callback for errors
         print("Adding Callbacks")