Merge pull request #44 from oresat/edl-os-cmd-fix

Fix up edl_cmd_shell for os commands

pyproject.toml

@@ -19,7 +19,7 @@ classifiers = [
 ]
 dependencies = [
     "bitstring",
-    "cfdp-py",
+    "cfdp-py==0.1.2",
     "dataclasses-json",
     "oresat-configs",
     "oresat-olaf>=3.5.0",

requirements.txt

@@ -1,7 +1,7 @@
 bitstring
 black
 build
-cfdp-py
+cfdp-py==0.1.2
 dataclasses-json
 isort
 oresat-configs

scripts/edl_cmd_shell.py

@@ -221,11 +221,14 @@ def help_sdo_write(self):
         print("  <index> is the index or object name")
         print("  <subindex> is the subindex or object name")
         print("  <value> is value to write")
+        print()
+        print("If <index>.<subindex> is of type DOMAIN, <value> will be interpreted as a filename")
+        print("and the contents will be written.")
 
     def do_sdo_write(self, arg: str):
         """Do the sdo_write command."""
 
-        args = arg.split(" ")
+        args = arg.split(" ", maxsplit=3)
         if len(args) != 4:
             self.help_sdo_write()
             return
@@ -284,8 +287,21 @@ def do_sdo_write(self, arg: str):
             value = float(args[3])
         elif obj.data_type == canopen.objectdictionary.VISIBLE_STRING:
             value = args[3]
+        elif obj.data_type == canopen.objectdictionary.DOMAIN:
+            try:
+                with open(args[3], "rb") as f:
+                    value = f.read()
+            except FileNotFoundError as e:
+                print(f"{e.__class__.__name__}: {e}")
+                return
+        elif obj.data_type == canopen.objectdictionary.OCTET_STRING:
+            try:
+                value = bytes.fromhex(args[3])
+            except ValueError:
+                value = args[3].encode("ascii")
+
         else:
-            print("invaid")
+            print(f"invalid OD obj type {obj} 0x{obj.data_type:X}")
             return
 
         raw = obj.encode_raw(value)

scripts/edl_file_upload.py

@@ -142,6 +142,7 @@ def run(self):
             packet = EdlPacket(payload, self._sequence_number, SRC_DEST_ORESAT)
             message = packet.pack(self._hmac_key)
             uplink.send(message)
+            print("Current sequence number:", self._sequence_number)
             self._sequence_number += 1
             time.sleep(self._delay)