Add method to run a file action

Author: fqqb

yamcs-client/src/yamcs/filetransfer/client.py

@@ -1,7 +1,8 @@
 import functools
 import json
-from typing import Any, Callable, Iterable, List, Mapping, Optional
+from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional
 
+from google.protobuf import json_format, struct_pb2
 from yamcs.core.context import Context
 from yamcs.core.exceptions import NotFound
 from yamcs.core.futures import WebSocketSubscriptionFuture
@@ -260,20 +261,40 @@ def get_filelist(
         message.ParseFromString(response.content)
         return RemoteFileListing(message)
 
-    def pause_transfer(self, id: str):
+    def pause_transfer(self, id: str) -> None:
         url = f"/filetransfer/{self._instance}/{self._service}/transfers/{id}:pause"
         self.ctx.post_proto(url)
 
-    def resume_transfer(self, id: str):
+    def resume_transfer(self, id: str) -> None:
         url = f"/filetransfer/{self._instance}/{self._service}/transfers/{id}:resume"
         self.ctx.post_proto(url)
 
-    def cancel_transfer(self, id: str):
+    def cancel_transfer(self, id: str) -> None:
         url = f"/filetransfer/{self._instance}/{self._service}/transfers/{id}:cancel"
         self.ctx.post_proto(url)
 
+    def run_file_action(
+        self,
+        file: str,
+        action: str,
+        message: Optional[Mapping[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        req = filetransfer_pb2.RunFileActionRequest()
+        req.file = file
+        req.action = action
+        if message:
+            req.message.update(message)
+
+        url = f"/filetransfer/{self._instance}/{self._service}/files:runFileAction"
+        response = self.ctx.post_proto(url, data=req.SerializeToString())
+        response_message = struct_pb2.Struct()
+        response_message.ParseFromString(response.content)
+        return json_format.MessageToDict(response_message)
+
     def create_transfer_subscription(
-        self, on_data: Optional[Callable[[Transfer], None]] = None, timeout: float = 60
+        self,
+        on_data: Optional[Callable[[Transfer], None]] = None,
+        timeout: float = 60,
     ) -> TransferSubscription:
         options = filetransfer_pb2.SubscribeTransfersRequest()
         options.instance = self._instance

yamcs-client/src/yamcs/filetransfer/model.py

@@ -1,12 +1,16 @@
 import datetime
 import threading
-from typing import TYPE_CHECKING, Any, Callable, List, Mapping, Optional
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Mapping, Optional
 
 from yamcs.core.helpers import parse_server_time
 from yamcs.protobuf.filetransfer import filetransfer_pb2
 
 if TYPE_CHECKING:
-    from yamcs.filetransfer.client import ServiceClient
+    from yamcs.filetransfer.client import (
+        FileListSubscription,
+        ServiceClient,
+        TransferSubscription,
+    )
 
 
 class Service:
@@ -170,29 +174,54 @@ def get_filelist(
             options=options,
         )
 
-    def pause_transfer(self, id: str):
+    def pause_transfer(self, id: str) -> None:
         """
         Pauses a transfer
         """
         self._service_client.pause_transfer(id)
 
-    def resume_transfer(self, id: str):
+    def resume_transfer(self, id: str) -> None:
         """
         Resume a transfer
         """
         self._service_client.resume_transfer(id)
 
-    def cancel_transfer(self, id: str):
+    def cancel_transfer(self, id: str) -> None:
         """
         Cancel a transfer
         """
         self._service_client.cancel_transfer(id)
 
+    def run_file_action(
+        self,
+        file: str,
+        action: str,
+        message: Optional[Mapping[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Run an action on a remote file.
+
+        Available actions depend on the specific file transfer implementation that
+        is in use use.
+
+        .. versionadded:: 1.9.6
+
+        :param file:
+            Remote file identifier
+        :param action:
+            Action identifier
+        :param message:
+            Action message
+        :return:
+            Action result (if the action returns anything)
+        """
+        return self._service_client.run_file_action(file, action, message)
+
     def create_transfer_subscription(
         self,
         on_data: Optional[Callable[["Transfer"], None]] = None,
         timeout: float = 60,
-    ):
+    ) -> "TransferSubscription":
         """
         Create a new transfer subscription.
 
@@ -213,7 +242,7 @@ def create_filelist_subscription(
         self,
         on_data: Optional[Callable[["RemoteFileListing"], None]] = None,
         timeout: float = 60,
-    ):
+    ) -> "FileListSubscription":
         """
         Create a new filelist subscription.
 

yamcs-client/src/yamcs/link/client.py

@@ -1,6 +1,7 @@
 import functools
-from typing import Callable, Mapping, Optional
+from typing import Any, Callable, Dict, Mapping, Optional
 
+from google.protobuf import json_format, struct_pb2
 from yamcs.core.context import Context
 from yamcs.core.futures import WebSocketSubscriptionFuture
 from yamcs.core.subscriptions import WebSocketSubscriptionManager
@@ -106,21 +107,30 @@ def disable_link(self):
         url = f"/links/{self._instance}/{self._link}:disable"
         self.ctx.post_proto(url, data=req.SerializeToString())
 
-    def run_action(self, action: str, message: Optional[Mapping] = None):
+    def run_action(
+        self,
+        action: str,
+        message: Optional[Mapping[str, Any]] = None,
+    ) -> Dict[str, Any]:
         """
         Runs the given action for this link
 
         :param action:
             action identifier
         :param message:
             action message
+        :return:
+            Action result (if the action returns anything)
         """
         req = links_pb2.RunActionRequest()
         if message:
             req.message.update(message)
 
         url = f"/links/{self._instance}/{self._link}/actions/{action}"
-        self.ctx.post_proto(url, data=req.message.SerializeToString())
+        response = self.ctx.post_proto(url, data=req.message.SerializeToString())
+        response_message = struct_pb2.Struct()
+        response_message.ParseFromString(response.content)
+        return json_format.MessageToDict(response_message)
 
     def get_cop1_config(self) -> Cop1Config:
         """