Update SDLS API: set counter, modify SPIs

Author: Alex Balgavy

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

@@ -12,8 +12,8 @@
 from yamcs.protobuf.sdls import sdls_pb2
 
 __all__ = [
-    "Cop1Subscription",
-    "LinkClient",
+    'Cop1Subscription',
+    'LinkClient',
 ]
 
 
@@ -92,7 +92,7 @@ def get_info(self) -> Link:
         """
         Get info on this link.
         """
-        response = self.ctx.get_proto(f"/links/{self._instance}/{self._link}")
+        response = self.ctx.get_proto(f'/links/{self._instance}/{self._link}')
         message = links_pb2.LinkInfo()
         message.ParseFromString(response.content)
         return Link(message)
@@ -102,15 +102,15 @@ def enable_link(self):
         Enables this link.
         """
         req = links_pb2.EnableLinkRequest()
-        url = f"/links/{self._instance}/{self._link}:enable"
+        url = f'/links/{self._instance}/{self._link}:enable'
         self.ctx.post_proto(url, data=req.SerializeToString())
 
     def disable_link(self):
         """
         Disables this link.
         """
         req = links_pb2.DisableLinkRequest()
-        url = f"/links/{self._instance}/{self._link}:disable"
+        url = f'/links/{self._instance}/{self._link}:disable'
         self.ctx.post_proto(url, data=req.SerializeToString())
 
     def run_action(
@@ -132,7 +132,7 @@ def run_action(
         if message:
             req.message.update(message)
 
-        url = f"/links/{self._instance}/{self._link}/actions/{action}"
+        url = f'/links/{self._instance}/{self._link}/actions/{action}'
         response = self.ctx.post_proto(url, data=req.message.SerializeToString())
         response_message = struct_pb2.Struct()
         response_message.ParseFromString(response.content)
@@ -142,7 +142,7 @@ def get_cop1_config(self) -> Cop1Config:
         """
         Gets the COP1 configuration for a data link.
         """
-        response = self.ctx.get_proto(f"/cop1/{self._instance}/{self._link}/config")
+        response = self.ctx.get_proto(f'/cop1/{self._instance}/{self._link}/config')
         message = cop1_pb2.Cop1Config()
         message.ParseFromString(response.content)
         return Cop1Config(message)
@@ -167,7 +167,7 @@ def update_cop1_config(
         if t1 is not None:
             req.t1 = int(round(1000 * t1))
 
-        url = f"/cop1/{self._instance}/{self._link}/config"
+        url = f'/cop1/{self._instance}/{self._link}/config'
         response = self.ctx.patch_proto(url, data=req.SerializeToString())
 
         message = cop1_pb2.Cop1Config()
@@ -183,7 +183,7 @@ def disable_cop1(self, bypass_all: bool = True):
         """
         req = cop1_pb2.DisableRequest()
         req.setBypassAll = bypass_all
-        url = f"/cop1/{self._instance}/{self._link}:disable"
+        url = f'/cop1/{self._instance}/{self._link}:disable'
         self.ctx.post_proto(url, data=req.SerializeToString())
 
     def initialize_cop1(
@@ -212,22 +212,22 @@ def initialize_cop1(
         if v_r is not None:
             req.vR = v_r
 
-        url = f"/cop1/{self._instance}/{self._link}:initialize"
+        url = f'/cop1/{self._instance}/{self._link}:initialize'
         self.ctx.post_proto(url, data=req.SerializeToString())
 
     def resume_cop1(self):
         """
         Resume COP1.
         """
         req = cop1_pb2.ResumeRequest()
-        url = f"/cop1/{self._instance}/{self._link}:resume"
+        url = f'/cop1/{self._instance}/{self._link}:resume'
         self.ctx.post_proto(url, data=req.SerializeToString())
 
     def get_cop1_status(self) -> Cop1Status:
         """
         Retrieve the COP1 status.
         """
-        response = self.ctx.get_proto(f"/cop1/{self._instance}/{self._link}/status")
+        response = self.ctx.get_proto(f'/cop1/{self._instance}/{self._link}/status')
         message = cop1_pb2.Cop1Status()
         message.ParseFromString(response.content)
         return Cop1Status(message)
@@ -251,7 +251,7 @@ def create_cop1_subscription(self, on_data: Callable[[Cop1Status], None], timeou
         options.instance = self._instance
         options.link = self._link
 
-        manager = WebSocketSubscriptionManager(self.ctx, topic="cop1", options=options)
+        manager = WebSocketSubscriptionManager(self.ctx, topic='cop1', options=options)
 
         # Represent subscription as a future
         subscription = Cop1Subscription(manager)
@@ -269,19 +269,38 @@ 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")
+        response = self.ctx.get_proto(f'/sdls/{self._instance}/{self._link}/{spi}/seq')
         message = sdls_pb2.GetSeqCtrResponse()
         message.ParseFromString(response.content)
-        return message.seq
+        return int.from_bytes(message.seq, byteorder='big')
 
-    def sdls_reset_ctr(self, spi: int):
+    def sdls_set_ctr(self, spi: int, new_seq: int):
         """
-        Reset the sequence counter associated with a given `spi` (Security Parameter Index) on this link.
+        Set the sequence counter associated with a given `spi` (Security Parameter Index) on this link to `new_seq`.
         """
-        self.ctx.delete_proto(f"/sdls/{self._instance}/{self._link}/{spi}/seq")
+        byte_length = (new_seq.bit_length() + 7) // 8
+        seq_bytes = new_seq.to_bytes(byte_length, byteorder='big')
+        self.ctx.put_proto(f'/sdls/{self._instance}/{self._link}/{spi}/seq', data=seq_bytes)
 
     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)
+        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())