Cosmostat Init Commit

files/api/LinkedList.py

@@ -0,0 +1,98 @@
+
+##############################
+# linked list classes
+# written by the intern
+##############################
+
+import time
+
+# single node in a singly linked list
+class Node:
+    __slots__ = ("value", "next", "timestamp")
+
+    def __init__(self, value):
+        self.value = value
+        self.timestamp = time.time()
+        self.next = None
+
+# small, bounded history implemented with a singly linked list
+class ValueHistory:
+    def __init__(self, maxlen: int):
+        if maxlen <= 0:
+            raise ValueError("maxlen must be a positive integer")
+        self.maxlen = maxlen
+        self.head: Node | None = None   # oldest entry
+        self.tail: Node | None = None   # newest entry
+        self.size = 0
+
+    # Append a new value to the history, dropping the oldest if needed
+    def add(self, value):
+        new_node = Node(value)
+
+        # link it after the current tail
+        if self.tail is None:          # empty list
+            self.head = self.tail = new_node
+        else:
+            self.tail.next = new_node
+            self.tail = new_node
+
+        self.size += 1
+
+        # 2. enforce the size bound
+        if self.size > self.maxlen:
+            # drop the head (oldest item)
+            assert self.head is not None   # for the type checker
+            self.head = self.head.next
+            self.size -= 1
+
+            # If the list became empty, also reset tail
+            if self.head is None:
+                self.tail = None
+
+    # Return the history as a Python dict list (oldest → newest)
+    def get_history(self, count: int | None = None):
+        if count is None:
+            count = self.maxlen        
+        out = []
+        cur = self.head
+        counter = 0
+        while cur is not None and counter < count:
+            counter += 1
+            out.append(
+                {
+                    "timestamp": cur.timestamp,
+                    "value": cur.value
+                }
+            )
+            cur = cur.next
+        return out
+    
+    # Return oldest timestamp
+    def get_first_timestamp(self):
+        if self.head is not None:
+            return self.head.timestamp
+        else:
+            return time.time()
+    
+    # Return current data
+    def get_current_value(self):
+        if self.tail is not None:
+            return self.tail.value
+        else:
+            return 0
+
+    # ------------------------------------------------------------------
+    #  Convenience methods
+    # ------------------------------------------------------------------
+    def __len__(self):
+        return self.size
+
+    def __iter__(self):
+        """Iterate over values from oldest to newest."""
+        cur = self.head
+        while cur is not None:
+            yield cur.value
+            cur = cur.next
+
+    def __repr__(self):
+        return f"BoundedHistory(maxlen={self.maxlen}, data={self.get()!r})"