pack command

hart_protocol/_parsing.py

@@ -1,9 +1,9 @@
-from collections import namedtuple
+import struct
 
 
 
 
-def parse(response: bytes) -> namedtuple:
+def parse(response: bytes) -> dict:
     out = dict()
     out["full_response"]: bytes = response
     if response[0] & 0x80:  # long address
@@ -12,9 +12,8 @@ def parse(response: bytes) -> namedtuple:
     else:  # short address
         out["address"]: bytes = response[1]
         response = response[2:]
-    command = response[0]
-    bytecount = response[1]
-    out["status"] = response[2:4]
+    command, bytecount, status = struct.unpack_from(">BBL", response)
+    out["status"] = status
     data = response[4:4+bytecount]
     out["command"]: int = command
     out["command_name"]: str = f"hart_command_{command}"
@@ -30,4 +29,7 @@ def parse(response: bytes) -> namedtuple:
         out["software_revision_level"]: bytes = data[6]
         out["hardware_revision_level"]: bytes = data[7]
         out["device_id"]: bytes = data[9:12]
+    if command in [1]:
+        out["command_name"] = "read_primary_variable"
+        out["primary_variable"] = struct.unpack_from(">f", data)
     return out

hart_protocol/tools.py

@@ -16,6 +16,25 @@ def calculate_long_address(manufacturer_id: int, manufacturer_device_type: int,
     return out.to_bytes(5, "big")
 
 
+def pack_command(address, command_id, data=None):
+    if type(address) == bytes:
+        address = int.from_bytes(address, "big")
+    if type(command_id) == int:
+        command_id = command_id.to_bytes(1, "big")
+    command = b"\xFF\xFF\xFF\xFF\xFF"  # preamble
+    command += b"\x82"  # start charachter
+    command += (549755813888 | address).to_bytes(5, "big")
+    command += command_id
+    if data is None:
+        command += b"\x00"  # byte count
+    else:
+        command += len(data).to_bytes(1, "big")  # byte count
+        command += data  # data
+    command += calculate_checksum(command[5:])
+    print(command)
+    return command
+
+
 def pack_ascii(string: str) -> bytes:
     chars = [c.encode() for c in string]
     chars = [ord(c) & 0b0011_1111 for c in chars]

hart_protocol/universal.py

@@ -2,32 +2,12 @@
 
 
 def read_unique_identifier(address: bytes) -> bytes:
-    command = b"\xFF\xFF\xFF\xFF\xFF"  # preamble
-    command += b"\x82"  # start charachter
-    command += b"\x80\x00\x00\x00\x00"  # address of sender
-    command += b"\x00"  # command ID (0)
-    command += b"\x00"  # byte count
-    command += tools.calculate_checksum(command[5:])
-    return command
+    return tools.pack_command(address, 0)
 
 
 def read_primary_variable(address: bytes) -> bytes:
-    command = b"\xFF\xFF\xFF\xFF\xFF"  # preamble
-    command += b"\x82"  # start charachter
-    command += b"\x80\x00\x00\x00\x00"  # address of sender
-    command += b"\x01"  # command ID (1)
-    command += b"\x00"  # byte count
-    command += tools.calculate_checksum(command[5:])
-    return command
+    return tools.pack_command(address, 1)
 
 
-def read_unique_identifier_associated_with_tag(tag: bytes) -> bytes:
-    assert len(tag) == 6
-    command = b"\xFF\xFF\xFF\xFF\xFF"  # preamble
-    command += b"\x82"  # start charachter
-    command += b"\x80\x00\x00\x00\x00"  # address of sender
-    command += b"\x0B"  # command ID (11)
-    command += b"\x06"  # byte count
-    command += tag
-    command += tools.calculate_checksum(command[5:])
-    return command
+def read_unique_identifier_associated_with_tag(tag: bytes, *, address:int=0) -> bytes:
+    return tools.pack_command(address, 11, tag)