TagoTiP service for sending device commands

docs/source/Services/TagoTiP_Type.rst

@@ -0,0 +1,19 @@
+**TagoTiP Type**
+===================
+
+
+.. _TagoTiPCommand:
+
+TagoTiPCommand
+--------------
+
+    **Attributes:**
+
+        | **serial**: str
+        | Serial of the target device
+
+        | **protocol**: str
+        | Protocol used to deliver the command
+
+        | **body**: str
+        | Command payload sent to the device

docs/source/Services/index.rst

@@ -197,6 +197,32 @@ Send SMS to phone number
                     "message": "Test",
                 }
             )
+=======
+TagoTiP
+=======
+
+Send a command to a device through TagoTiP
+
+Requires a Service Authorization token in the module token.
+
+    **Parameters:**
+
+        | **command**: :ref:`TagoTiPCommand`
+        | Command object with serial, protocol and body
+
+.. code-block::
+    :caption: **Example:**
+
+            from tagoio_sdk import Services
+
+            services = Services({"token": "your-service-authorization-token"})
+            services.tagotip.cmd(
+                {
+                    "serial": "mqtt1",
+                    "protocol": "mqtt",
+                    "body": "reboot-now",
+                }
+            )
 
 
 .. toctree::
@@ -207,6 +233,7 @@ Send SMS to phone number
     Notification_Type
     Pdf_Type
     SMS_Type
+    TagoTiP_Type
 
 
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "tagoio-sdk"
-version = "5.1.4"
+version = "5.1.5"
 description = "Official Python SDK for TagoIO"
 authors = [{name = "TagoIO Inc.", email = "contact@tago.io"}]
 license = {text = "Apache-2.0"}

src/tagoio_sdk/modules/Services/Services.py

@@ -9,6 +9,7 @@
 from tagoio_sdk.modules.Services.Notification import Notification
 from tagoio_sdk.modules.Services.PDF import PDFService
 from tagoio_sdk.modules.Services.SMS import SMS
+from tagoio_sdk.modules.Services.TagoTiP import TagoTiP
 from tagoio_sdk.params import get_params
 
 
@@ -23,3 +24,4 @@ def __init__(self, params: Optional[GenericModuleParams] = None):
         self.Notification = Notification(self.params)
         self.Attachment = Attachment(self.params)
         self.PDF = PDFService(self.params)
+        self.tagotip = TagoTiP(self.params)

src/tagoio_sdk/modules/Services/TagoTiP.py

@@ -0,0 +1,50 @@
+from typing import TypedDict
+
+from tagoio_sdk.common.tagoio_module import TagoIOModule
+
+
+class TagoTiPCommand(TypedDict):
+    serial: str
+    """
+    Serial of the target device
+    """
+    protocol: str
+    """
+    Protocol used to deliver the command
+    """
+    body: str
+    """
+    Command payload sent to the device
+    """
+
+
+class TagoTiP(TagoIOModule):
+    def cmd(self, command: TagoTiPCommand) -> str:
+        """
+        Send a command to a device through TagoTiP.
+
+        Requires a Service Authorization token in the module `token`.
+
+        :param TagoTiPCommand command: Command object with serial, protocol and body
+
+        Example:
+            >>> from tagoio_sdk import Services
+            >>> services = Services({"token": "your-service-authorization-token"})
+            >>> services.tagotip.cmd({
+            ...     "serial": "mqtt1",
+            ...     "protocol": "mqtt",
+            ...     "body": "reboot-now",
+            ... })
+        """
+        result = self.doRequest(
+            {
+                "path": "/tip/cmd",
+                "method": "POST",
+                "body": {
+                    "serial": command["serial"],
+                    "protocol": command["protocol"],
+                    "body": command["body"],
+                },
+            }
+        )
+        return result

tests/Services/test_tagotip.py

@@ -0,0 +1,32 @@
+import os
+
+from requests_mock.mocker import Mocker
+
+from tagoio_sdk.modules.Services.Services import Services
+
+
+os.environ["T_ANALYSIS_TOKEN"] = "your_token_value"
+
+
+def testTagoTiPCmd(requests_mock: Mocker) -> None:
+    """Test cmd method of TagoTiP service."""
+    requests_mock.post(
+        "https://api.tago.io/tip/cmd", json={"status": True, "result": "ok"}
+    )
+
+    services = Services({"token": "your-service-authorization-token"})
+
+    result = services.tagotip.cmd(
+        {
+            "serial": "mqtt1",
+            "protocol": "mqtt",
+            "body": "reboot-now",
+        }
+    )
+
+    assert result == "ok"
+    assert requests_mock.last_request.json() == {
+        "serial": "mqtt1",
+        "protocol": "mqtt",
+        "body": "reboot-now",
+    }