rpc_wrappers: data_logger_read: added

src/infuse_iot/rpc_wrappers/coap_download.py

@@ -6,6 +6,7 @@
 import infuse_iot.generated.rpc_definitions as defs
 from infuse_iot.commands import InfuseRpcCommand
 from infuse_iot.generated.rpc_definitions import rpc_enum_file_action
+from infuse_iot.util.ctypes import UINT32_MAX
 
 
 class coap_download(InfuseRpcCommand, defs.coap_download):
@@ -79,8 +80,6 @@ class request(ctypes.LittleEndianStructure):
             ]
             _pack_ = 1
 
-        UINT32_MAX = 2**32 - 1
-
         return request(
             self.server,
             self.port,
@@ -92,8 +91,6 @@ class request(ctypes.LittleEndianStructure):
         )
 
     def request_json(self):
-        UINT32_MAX = 2**32 - 1
-
         return {
             "server_address": self.server.decode("utf-8"),
             "server_port": str(self.port),

src/infuse_iot/rpc_wrappers/data_logger_read.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+
+import binascii
+import os
+
+import infuse_iot.generated.rpc_definitions as defs
+from infuse_iot.commands import InfuseRpcCommand
+from infuse_iot.util.ctypes import UINT32_MAX
+
+
+class data_logger_read(InfuseRpcCommand, defs.data_logger_read):
+    RPC_DATA_RECEIVE = True
+
+    @classmethod
+    def add_parser(cls, parser):
+        logger = parser.add_mutually_exclusive_group(required=True)
+        logger.add_argument("--onboard", action="store_true", help="Onboard flash logger")
+        logger.add_argument("--removable", action="store_true", help="Removable flash logger (SD)")
+        parser.add_argument("--start", type=int, default=0, help="First logger block to read (default 0)")
+        parser.add_argument("--last", type=int, default=UINT32_MAX, help="Last logger block to read (default all)")
+
+    def __init__(self, args):
+        self.infuse_id = args.id
+        self.start = args.start
+        self.last = args.last
+        if args.onboard:
+            self.logger = defs.rpc_enum_data_logger.FLASH_ONBOARD
+        elif args.removable:
+            self.logger = defs.rpc_enum_data_logger.FLASH_REMOVABLE
+        else:
+            raise NotImplementedError
+        self.expected_offset = 0
+        self.output = b""
+
+    def request_struct(self):
+        return self.request(self.logger, self.start, self.last)
+
+    def request_json(self):
+        return {"logger": self.logger.name, "start_block": self.start, "last_block": self.last}
+
+    def data_recv_cb(self, offset: int, data: bytes) -> None:
+        if offset != self.expected_offset:
+            missing = offset - self.expected_offset
+            print(f"Missed {missing:d} bytes from offset 0x{self.expected_offset:08x}")
+            self.output += b"\x00" * missing
+
+        self.output += data
+        # Next expected offset
+        self.expected_offset = offset + len(data)
+
+    def handle_response(self, return_code, response):
+        if return_code != 0:
+            print(f"Failed to read data logger ({os.strerror(-return_code)})")
+            return
+
+        if response.sent_len != len(self.output):
+            print(f"Unexpected received length ({response.sent_len} != {len(self.output)})")
+            return
+
+        if response.sent_crc != binascii.crc32(self.output):
+            print(f"Unexpected received length ({response.sent_crc:08x} != {binascii.crc32(self.output)}:08x)")
+            return
+
+        output_file = f"{self.infuse_id:016x}_{self.logger.name}.bin"
+        with open(output_file, "wb") as f:
+            f.write(self.output)
+        print(f"Wrote {response.sent_len:d} bytes to {output_file}")

src/infuse_iot/rpc_wrappers/data_logger_state.py

@@ -4,6 +4,7 @@
 
 import infuse_iot.generated.rpc_definitions as defs
 from infuse_iot.commands import InfuseRpcCommand
+from infuse_iot.util.time import humanised_seconds
 
 
 class data_logger_state(InfuseRpcCommand, defs.data_logger_state):
@@ -29,7 +30,7 @@ def request_json(self):
 
     def handle_response(self, return_code, response):
         if return_code != 0:
-            print(f"Failed to query current time ({os.strerror(-return_code)})")
+            print(f"Failed to query data logger state ({os.strerror(-return_code)})")
             return
 
         def sizeof_fmt(num, suffix="B"):
@@ -48,5 +49,17 @@ def sizeof_fmt(num, suffix="B"):
         print(f"{self.logger.name}")
         print(f"\t     Logged: {sizeof_fmt(total_logged)}")
         print(f"\t     Blocks: {r.current_block}/{r.logical_blocks} ({percent:.0f}%)")
-        print(f"\t Block Rate: {block_rate:.2f} blocks/sec")
-        print(f"\t  Byte Rate: {sizeof_fmt(byte_rate)}/sec")
+        if byte_rate == 0.0:
+            print("\t Block Rate: N/A")
+            print("\t  Byte Rate: N/A")
+        elif byte_rate < 0.1:
+            print(f"\t Block Rate: {1/block_rate:.2f} sec/block")
+            print(f"\t  Byte Rate: {1/byte_rate:.2f} sec/byte")
+        else:
+            print(f"\t Block Rate: {block_rate:.2f} blocks/sec")
+            print(f"\t  Byte Rate: {sizeof_fmt(byte_rate)}/sec")
+        if r.bytes_logged > 0:
+            physical_wrap_time = r.physical_blocks / block_rate
+            logical_fill_time = r.logical_blocks / block_rate
+            print(f"\t   Phy Wrap: {humanised_seconds(int(physical_wrap_time))}")
+            print(f"\t   Log Fill: {humanised_seconds(int(logical_fill_time))}")

src/infuse_iot/tools/localhost/index.html

@@ -41,6 +41,10 @@ <h2>Table Control</h2>
         // Initialize the Tabulator table
         const dataTable = new Tabulator("#data-table", {
             layout: "fitDataTable",
+            pagination:"local",
+            paginationSize:100,
+            paginationSizeSelector:[10, 25, 50, 100],
+            paginationCounter:"rows",
         });
         const shownTdfTable = new Tabulator("#tdf-show", {
             layout: "fitDataTable",

src/infuse_iot/util/ctypes.py

@@ -5,6 +5,19 @@
 
 from typing_extensions import Any, Self
 
+UINT8_MAX = 2**8 - 1
+UINT16_MAX = 2**16 - 1
+UINT32_MAX = 2**32 - 1
+UINT64_MAX = 2**64 - 1
+INT8_MIN = -(2**7)
+INT16_MIN = -(2**15)
+INT32_MIN = -(2**31)
+UINT64_MIN = -(2**63)
+INT8_MAX = 2**7 - 1
+INT16_MAX = 2**15 - 1
+INT32_MAX = 2**31 - 1
+INT64_MAX = 2**63 - 1
+
 
 def bytes_to_uint8(b: bytes):
     return (len(b) * ctypes.c_uint8)(*b)
@@ -42,7 +55,7 @@ def vla_from_buffer_copy(cls, source, offset=0) -> Self:
         return base
 
     def iter_fields(self, prefix: str = "") -> Generator[tuple[str, Any], None, None]:
-        for field_name, _field_type in self._fields_:
+        for field_name, _field_type in self._fields_:  # type: ignore
             val = getattr(self, field_name)
             if isinstance(val, VLACompatLittleEndianStruct):
                 yield from val.iter_fields(f"{field_name}.")

src/infuse_iot/util/time.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+
+
+def humanised_seconds(seconds: int, units: int = 2) -> str:
+    """Convert a seconds count to human readable units"""
+    scales = [365 * 24 * 60 * 60, 7 * 24 * 60 * 60, 24 * 60 * 60, 60 * 60, 60, 1]
+    postfix = ["year", "week", "day", "hour", "minute", "second"]
+
+    vals = []
+    for scale in scales:
+        vals.append(seconds // scale)
+        seconds -= scale * vals[-1]
+
+    def construct(val, postfix):
+        return f"{val} {postfix}{'' if val == 1 else 's'}"
+
+    for idx, val in enumerate(vals):
+        if val == 0:
+            continue
+        units = min(units, len(postfix) - idx)
+        final = []
+        while units > 0:
+            final.append(construct(vals[idx], postfix[idx]))
+            idx += 1
+            units -= 1
+
+        return ", ".join(final)
+    return "0 seconds"

tests/util/test_ctypes.py

@@ -0,0 +1,60 @@
+import ctypes
+import os
+
+from infuse_iot.util.ctypes import VLACompatLittleEndianStruct
+
+assert "TOXTEMPDIR" in os.environ, "you must run these tests using tox"
+
+
+class VLABase(VLACompatLittleEndianStruct):
+    _fields_ = [
+        ("first", ctypes.c_uint32),
+    ]
+    vla_field = ("vla", 0 * ctypes.c_uint32)
+
+
+class VLANested(VLACompatLittleEndianStruct):
+    _fields_ = [
+        ("first", ctypes.c_uint32),
+    ]
+    vla_field = ("vla", VLABase)
+
+
+class VLANone(VLACompatLittleEndianStruct):
+    _fields_ = [
+        ("first", ctypes.c_uint32),
+        ("second", ctypes.c_uint32),
+    ]
+
+
+def test_vla_compat_struct():
+    b = b"".join(x.to_bytes(4, "little") for x in range(2, 10))
+
+    base = VLABase.vla_from_buffer_copy(b)
+    assert base.first == 2
+    assert len(base.vla) == 7
+    for idx, val in enumerate(base.vla):
+        assert val == idx + 3
+
+    nested = VLANested.vla_from_buffer_copy(b)
+    assert nested.first == 2
+    assert nested.vla.first == 3
+    assert len(nested.vla.vla) == 6
+    for idx, val in enumerate(nested.vla.vla):
+        assert val == idx + 4
+
+    none = VLANone.vla_from_buffer_copy(b)
+    assert none.first == 2
+    assert none.second == 3
+
+    unaligned = b"\x00" * 31
+    try:
+        VLABase.vla_from_buffer_copy(unaligned)
+        raise AssertionError()
+    except TypeError:
+        pass
+    try:
+        VLANested.vla_from_buffer_copy(unaligned)
+        raise AssertionError()
+    except TypeError:
+        pass

tests/util/test_time.py

@@ -0,0 +1,18 @@
+import os
+
+from infuse_iot.util.time import humanised_seconds
+
+assert "TOXTEMPDIR" in os.environ, "you must run these tests using tox"
+
+
+def test_humanised_seconds():
+    for units in range(1, 6):
+        # Test 0 case
+        assert humanised_seconds(0, units) == "0 seconds"
+        # Test a range of magnitudes
+        seconds = 1
+        for _time_step in range(10):
+            result = humanised_seconds(seconds, units)
+            assert isinstance(result, str)
+            print(f"{seconds:10}: {result}")
+            seconds *= 10