example script

ryzen7_script.py

@@ -0,0 +1,62 @@
+import math
+
+from i7lib.client import run
+from i7lib.timeutils import now, to_datetime
+
+INPUTS = ["ryzen7-cpu", "ryzen7-ram", "ryzen7-disk"]
+ID_LOAD_INDEX = 54
+NAME_LOAD_INDEX = "ryzen7-load-index"
+LOG_EVERY = 60
+
+tick = 0
+
+
+def log(msg):
+    stamp = to_datetime(now()).strftime("%Y-%m-%d %H:%M:%S")
+    print(f"[{stamp}] {msg}")
+
+
+def read_inputs(plant):
+    values = []
+    for name in INPUTS:
+        value = plant.state.get(name)
+        if value is None or math.isnan(value):
+            return None
+        values.append(value)
+    return values
+
+
+def report_missing(plant):
+    missing = [name for name in INPUTS if name not in plant.state]
+    if missing:
+        log(f"WARNING: not exposed to this job: {', '.join(missing)}")
+
+
+def write_load_index(plant, load_index):
+    batch = plant.new_write_batch()
+    batch.add(ID_LOAD_INDEX, load_index, NAME_LOAD_INDEX)
+    plant.send(batch)
+
+
+def logic(plant):
+    global tick
+
+    if tick == 0:
+        report_missing(plant)
+    tick += 1
+
+    values = read_inputs(plant)
+    if values is None:
+        log("skipping write: an input is missing or not yet reporting")
+        return
+
+    load_index = max(values)
+    write_load_index(plant, load_index)
+
+    if tick % LOG_EVERY == 1:
+        readings = " ".join(f"{n}={v:.2f}" for n, v in zip(INPUTS, values))
+        log(f"{readings} -> {NAME_LOAD_INDEX}={load_index:.2f}")
+
+
+if __name__ == "__main__":
+    run(logic, interval_ms=1000, offset_ms=200)