Add scripts command group

src/lvmnps/actor/commands/__init__.py

@@ -12,4 +12,5 @@
 
 from .onoff import cycle, off, on
 from .refresh import refresh
+from .scripts import scripts
 from .status import status

src/lvmnps/actor/commands/scripts.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2023-11-23
+# @Filename: scripts.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+from . import lvmnps_command_parser
+
+
+if TYPE_CHECKING:
+    from src.lvmnps.actor.actor import NPSCommand
+
+
+__all__ = ["scripts"]
+
+
+def validate_nps(command: NPSCommand):
+    """Checks that the NPS implements scripting."""
+
+    nps = command.actor.nps
+
+    if nps.implementations.get("scripting", False) is False:
+        command.fail("Scripting not allowed for this NPS.")
+        return False
+
+    return True
+
+
+@lvmnps_command_parser.group()
+def scripts():
+    """Handles user scripts."""
+
+    pass
+
+
+@scripts.command()
+@click.argument("SCRIPT", type=str, nargs=-1)
+async def run(command: NPSCommand, script: tuple[str, ...]):
+    """Runs a user script.
+
+    The first argument is expected to be the name of the script. Additional
+    argument will be considered arguments to pass to the script function.
+
+    """
+
+    if not validate_nps(command):
+        return
+
+    nps = command.actor.nps
+
+    script_name = script[0]
+    script_args = script[1:]
+
+    try:
+        thread_id = await nps.run_script(script_name, *script_args)
+    except Exception as err:
+        return command.fail(f"Failed executing script {script_name!r}: {err}")
+
+    return command.finish(
+        script={
+            "name": script_name,
+            "args": list(script_args),
+            "running": True,
+            "thread_id": thread_id,
+        }
+    )
+
+
+@scripts.command()
+@click.argument("THREAD", type=int, required=False)
+async def stop(command: NPSCommand, thread: int | None):
+    """Runs a user script.
+
+    If the thread number is not provided, stops all running scripts.
+
+    """
+
+    if not validate_nps(command):
+        return
+
+    nps = command.actor.nps
+
+    try:
+        await nps.stop_script(thread_num=thread)
+    except Exception:
+        return command.fail("Failed stopping scripts.")
+
+    return command.finish()
+
+
+@scripts.command(name="list")
+async def list_scripts(command: NPSCommand):
+    """Lists running scripts."""
+
+    if not validate_nps(command):
+        return
+
+    nps = command.actor.nps
+
+    threads = await nps.list_running_scripts()
+
+    scripts = [
+        {"name": threads[id_], "args": [], "running": True, "thread_id": id_}
+        for id_ in threads
+    ]
+
+    return command.finish(scripts=scripts)

src/lvmnps/actor/schema.json

@@ -1,6 +1,19 @@
 {
   "$schema": "http://json-schema.org/draft-07/schema",
-  "definitions": {},
+  "definitions": {
+    "script": {
+      "type": "object",
+      "description": "Information about a user script",
+      "properties": {
+        "name": { "type": "string" },
+        "args": { "type": "array", "items": { "type": "string" } },
+        "running": { "type": "boolean" },
+        "thread_id": { "type": "integer" }
+      },
+      "required": ["name", "args", "running", "thread_id"],
+      "additionalProperties": false
+    }
+  },
   "type": "object",
   "properties": {
     "nps_type": { "type": "string", "description": "Type of NPS switch" },
@@ -35,7 +48,9 @@
         "additionalProperties": true,
         "description": "Properties of each outlet"
       }
-    }
+    },
+    "script": { "$ref": "#/definitions/script" },
+    "scripts": { "type": "array", "items": { "$ref": "#/definitions/script" } }
   },
   "additionalProperties": false
 }

src/lvmnps/nps/core.py

@@ -11,7 +11,7 @@
 import abc
 import asyncio
 
-from typing import Any, Sequence
+from typing import Any, Sequence, TypedDict
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -63,10 +63,17 @@ async def off(self):
 OutletArgType = OutletModel | int | str | Sequence[str | int | OutletModel]
 
 
+class ImplementationsDict(TypedDict):
+    """Dictionary of NPS implementations."""
+
+    scripting: bool
+
+
 class NPSClient(abc.ABC):
     """Base NPS client."""
 
     nps_type: str
+    implementations: ImplementationsDict = {"scripting": False}
 
     def __init__(self):
         self.outlets: dict[str, OutletModel] = {}
@@ -181,3 +188,18 @@ async def cycle(
         switched_outlets = await self.set_state(outlets, on=True)
 
         return switched_outlets
+
+    async def run_script(self, name: str, *args, **kwargs) -> int:
+        """Runs a user script."""
+
+        raise NotImplementedError()
+
+    async def stop_script(self, thread_num: int | None = None):
+        """Stops a running script."""
+
+        raise NotImplementedError()
+
+    async def list_running_scripts(self) -> dict[int, str]:
+        """Returns a list of running scripts."""
+
+        raise NotImplementedError()

src/lvmnps/nps/implementations/dli.py

@@ -85,6 +85,7 @@ def __post_init__(self):
         self.outlet: dict[str, DLIOutletModel] = {}
 
         self.nps_type = "dli"
+        self.implementations = {"scripting": True}
 
     async def setup(self):
         """Sets up the power supply, setting any required configuration options."""
@@ -261,3 +262,18 @@ async def stop_script(self, thread_num: int | None = None):
             )
 
             self._validate_response(response, 200)
+
+    async def list_running_scripts(self) -> dict[int, str]:
+        """Returns a mapping of running thread number to script name."""
+
+        threads: dict[int, str] = {}
+
+        async with self.api_client as client:
+            response = await client.get(url="/script/threads/")
+            self._validate_response(response, 200)
+
+        for thread, description in response.json().items():
+            script_name = description["label"].split(" ")[0]
+            threads[int(thread)] = script_name
+
+        return threads