Merge pull request #34 from thezeroalpha/sdls

Add methods for SDLS API interaction

Author: fqqb

yamcs-client/src/yamcs/client/links/client.py

@@ -9,6 +9,7 @@
 from yamcs.client.model import Link
 from yamcs.protobuf.cop1 import cop1_pb2
 from yamcs.protobuf.links import links_pb2
+from yamcs.protobuf.sdls import sdls_pb2
 
 __all__ = [
     "Cop1Subscription",
@@ -231,9 +232,7 @@ def get_cop1_status(self) -> Cop1Status:
         message.ParseFromString(response.content)
         return Cop1Status(message)
 
-    def create_cop1_subscription(
-        self, on_data: Callable[[Cop1Status], None], timeout: float = 60
-    ) -> Cop1Subscription:
+    def create_cop1_subscription(self, on_data: Callable[[Cop1Status], None], timeout: float = 60) -> Cop1Subscription:
         """
         Create a new subscription for receiving status of the COP1 link.
 
@@ -257,13 +256,52 @@ def create_cop1_subscription(
         # Represent subscription as a future
         subscription = Cop1Subscription(manager)
 
-        wrapped_callback = functools.partial(
-            _wrap_callback_parse_cop1_status, subscription, on_data
-        )
+        wrapped_callback = functools.partial(_wrap_callback_parse_cop1_status, subscription, on_data)
 
         manager.open(wrapped_callback)
 
         # Wait until a reply or exception is received
         subscription.reply(timeout=timeout)
 
         return subscription
+
+    def sdls_get_ctr(self, spi: int) -> int:
+        """
+        Get the sequence counter associated with a given `spi` (Security Parameter Index) on this link.
+        """
+        response = self.ctx.get_proto(f"/sdls/{self._instance}/{self._link}/{spi}/seq")
+        message = sdls_pb2.GetSeqCtrResponse()
+        message.ParseFromString(response.content)
+        return int.from_bytes(message.seq, byteorder="big")
+
+    def sdls_set_ctr(self, spi: int, new_seq: int):
+        """
+        Set the sequence counter associated with a given `spi` (Security Parameter Index) on this link to `new_seq`.
+        """
+        body = sdls_pb2.SetSeqCtrRequestBody()
+        byte_length = (new_seq.bit_length() + 7) // 8
+        body.seq = new_seq.to_bytes(byte_length, byteorder="big")
+        self.ctx.put_proto(f"/sdls/{self._instance}/{self._link}/{spi}/seq", data=body.SerializeToString())
+
+    def sdls_set_key(self, spi: int, key: bytes):
+        """
+        Update the `key` associated with a given `spi` (Security Parameter Index) on this link.
+        """
+        self.ctx.put_proto(f"/sdls/{self._instance}/{self._link}/{spi}/key", data=key)
+
+    def sdls_set_spi(self, vc_id: int, spi: int):
+        """
+        Update the `spi` used for SDLS on the VC `vc_id` on this link.
+        """
+        body = sdls_pb2.SetSpiRequestBody()
+        body.spi = spi
+        self.ctx.put_proto(f"/sdls/{self._instance}/{self._link}/{vc_id}/spi", data=body.SerializeToString())
+
+    def sdls_set_spis(self, vc_id: int, spis: list[int]):
+        """
+        Update the `spis` used for SDLS on the VC `vc_id` on this link.
+        Only valid for incoming links (i.e., downlink).
+        """
+        body = sdls_pb2.SetSpisRequestBody()
+        body.spis[:] = spis
+        self.ctx.put_proto(f"/sdls/{self._instance}/{self._link}/{vc_id}/spis", data=body.SerializeToString())