next mass update

umnp/protocol/common/timestamp.py

+import sys
+import time
+
 try:
     import datetime
 except ImportError:
     from umnp.protocol.compat.datetime import datetime
 
 
+if sys.implementation == "micropython":
+    # noinspection PyUnresolvedReferences
+    import machine
+
+
+"""
+        rtc = machine.RTC()
+        now = rtc.datetime()
+        year, month, day, hour, minute, second, second_fraction = now
+"""
+
+
+def ts_utc_now_second() -> int:
+    if sys.implementation == "micropython":
+        rtc = machine.RTC()
+        # https://docs.micropython.org/en/latest/library/time.html#module-time
+        # Micropython's time.time() returns the number of seconds, as an integer, since the Epoch
+        return time.time()
+
+    else:
+        ts = datetime.datetime.now(tz=datetime.timezone.utc).timestamp()
+        return int(round(ts))
+
+
 class TimeStamp:
     def __init__(self, when=None):
         if when:
-            self._ts = int(round(datetime.datetime.now(datetime.timezone.utc).timestamp()))
+            self._ts = ts_utc_now_second()
         else:
             self._ts = when
 
@@ -29,5 +56,3 @@ def valid_timestamp(when: TimeStamp | None) -> TimeStamp:
         raise ValueError("Expected None TimeStamp")
 
     return when
-
-

tools/fs/__init__.py → umnp/tools/fs/__init__.py

 import os
 
 
-def fs_entries_recurse(current_directory="", total_path=""):
+def human_readable_time(timestamp: int):
+    t = time.localtime(timestamp)
+    year, month, day = t[0:3]
+    hour, minute, second = t[3:6]
+    return f"{year}-{month:02d}-{day:02d} {hour:02d}:{minute:02d}:{second:02d}"
+
+
+def fs_entries_recurse(current_directory="", total_path="/"):
     files = []
     dirs = []
     all_entries = []
 
-    for fstats in os.ilistdir(current_directory):
+    for fstats in os.ilistdir(total_path):
         fn = fstats[0]
         ft = fstats[1]
-        all_entries.append(total_path + "/" + fn)
+        if total_path.endswith("/"):
+            target = total_path + fn
+        else:
+            target = total_path + "/" + fn
+        all_entries.append(target)
         if ft == 0x4000:
             dirs.append(total_path + "/" + fn)
 
-            f, current_directory, a = fs_entries_recurse(fn, total_path + "/" + fn)
+            f, current_directory, a = fs_entries_recurse(fn, target)
             files.extend(f)
             dirs.extend(current_directory)
             all_entries.extend(a)
         else:
-            files.append(total_path + "/" + fn)
+            files.append(target)
     return files, dirs, all_entries
 
 
 class FileSystemInformation:
     def __init__(self):
         self._size_cache = {}
+        self._atime_cache = {}
         self._files = []
         self._directories = []
         self._all_entries = []
@@ -38,8 +50,7 @@ class FileSystemInformation:
         self._size_total_kb = None
         self._longest_path = 0
         self.update_file_system_info()
-
-        self._files, self._directories, self._all_entries = fs_entries_recurse()
+        self._files, self._directories, self._all_entries = fs_entries_recurse("")
         for fn in self._all_entries:
             self._longest_path = max(self._longest_path, len(fn))
 
@@ -50,17 +61,22 @@ class FileSystemInformation:
         self._directories = []
         self._all_entries = []
 
-    def get_size(self, path):
-        if path in self._size_cache:
-            return self._size_cache[path]
+    def get_atime(self, path):
+        stats = os.stat(path)
+        return stats[7]
 
-        mode, inode, device_id, link_count, uid, gid, size, atime, mtime, ctime = os.stat(path)
+    def get_size(self, path):
+        # mode, inode, device_id, link_count, uid, gid, size, atime, mtime, ctime
+        mode, _, _, _, _, _, size, _, _, _ = os.stat(path)
         b = 0
         if mode == 0x8000:
             b = size
         self._size_cache[path] = b
         return b
 
+    def is_dir(self, path):
+        return os.stat(path)[0] == 0x4000
+
     def update_file_system_info(self):
         if len(self._vfs_stats) == 0:
             self._vfs_stats = os.statvfs("/")
@@ -77,7 +93,13 @@ class FileSystemInformation:
     def print_fs_tree(self):
         for fn in self._all_entries:
             spacer = " " * (self._longest_path + 2 - len(fn))
+
             size = self.get_size(fn)
+            is_dir = self.is_dir(fn)
             size_spacer = " " * (6 - len(str(size)))
+            a_time = " " * 19
+            if not self.is_dir(fn):
+                t = self.get_atime(fn)
+                a_time = human_readable_time(t)
 
-            print(f"{fn}{spacer} {size_spacer}{size} bytes")
+            print(f"{fn}{spacer} {size_spacer}{size} bytes -- {a_time}")

umnp/tools/list-files.py

+import os
+import time
+
+
+def human_readable_time(timestamp: int):
+    t = time.localtime(timestamp)
+    year, month, day = t[0:3]
+    hour, minute, second = t[3:6]
+    return f"{year}-{month:02d}-{day:02d}T{hour:02d}:{minute:02d}:{second:02d}"
+
+
+def fs_entries_recurse(current_directory="", total_path="/"):
+    files = []
+    dirs = []
+    all_entries = []
+
+    for fstats in os.ilistdir(total_path):
+        fn = fstats[0]
+        ft = fstats[1]
+        if total_path.endswith("/"):
+            target = total_path + fn
+        else:
+            target = total_path + "/" + fn
+        all_entries.append(target)
+        if ft == 0x4000:
+            dirs.append(total_path + "/" + fn)
+
+            f, current_directory, a = fs_entries_recurse(fn, target)
+            files.extend(f)
+            dirs.extend(current_directory)
+            all_entries.extend(a)
+        else:
+            files.append(target)
+    return files, dirs, all_entries
+
+
+class FileSystemInformation:
+    def __init__(self):
+        self._size_cache = {}
+        self._atime_cache = {}
+        self._files = []
+        self._directories = []
+        self._all_entries = []
+        self._vfs_stats = []
+        self._usage = None
+        self._size_free_kb = None
+        self._blocks_free = None
+        self._max_fn_len = None
+        self._blocks_used = None
+        self._block_size = None
+        self._size_total_kb = None
+        self._longest_path = 0
+        self.update_file_system_info()
+        self._files, self._directories, self._all_entries = fs_entries_recurse("")
+        for fn in self._all_entries:
+            self._longest_path = max(self._longest_path, len(fn))
+
+    def invalidate(self):
+        self._size_cache = {}
+        self._vfs_stats = []
+        self._files = []
+        self._directories = []
+        self._all_entries = []
+
+    def get_atime(self, path):
+        stats = os.stat(path)
+        return stats[7]
+
+    def get_size(self, path):
+        # mode, inode, device_id, link_count, uid, gid, size, atime, mtime, ctime
+        mode, _, _, _, _, _, size, _, _, _ = os.stat(path)
+        b = 0
+        if mode == 0x8000:
+            b = size
+        self._size_cache[path] = b
+        return b
+
+    def is_dir(self, path):
+        return os.stat(path)[0] == 0x4000
+
+    def update_file_system_info(self):
+        if len(self._vfs_stats) == 0:
+            self._vfs_stats = os.statvfs("/")
+
+        b_size, _, b_count, b_count_free, _, _, _, _, _, max_fn_len = self._vfs_stats
+        self._usage = 1 - b_count_free / b_count
+        self._size_free_kb = b_size * b_count_free / 1024
+        self._blocks_free = b_count_free
+        self._blocks_used = b_count - b_count_free
+        self._block_size = b_size
+        self._size_total_kb = b_size * b_count / 1024
+        self._max_fn_len = max_fn_len
+
+    def print_fs_tree(self):
+        for fn in self._all_entries:
+            spacer = " " * (self._longest_path + 2 - len(fn))
+
+            size = self.get_size(fn)
+            is_dir = self.is_dir(fn)
+            size_spacer = " " * (6 - len(str(size)))
+            a_time = "?" + " " * 18
+            if is_dir:
+                ft = "d"
+            else:
+                ft = "f"
+            if not self.is_dir(fn):
+                t = self.get_atime(fn)
+                a_time = human_readable_time(t)
+
+            print(f"{fn}{spacer} {size_spacer}{size} bytes {ft} {a_time}")
+
+
+def main():
+    fs = FileSystemInformation()
+    fs.print_fs_tree()
+
+
+main()