Enables Python checkers black and ruff and applies "nix fmt"

Author: bladyjoker

antithesis/entrypoint.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env -S python3 -u
 
-# This file serves as the client's entrypoint. It: 
+# This file serves as the client's entrypoint. It:
 # 1. Confirms that all nodes in the cluster are available
 # 2. Signals "setupComplete" using the Antithesis SDK
 
@@ -15,7 +15,7 @@
 print("Client [entrypoint]: running...")
 
 # Here is the python format for setup_complete. At this point, our system is fully initialized and ready to test.
-setup_complete({"Message":"sim-rs cluster is healthy"})
+setup_complete({"Message": "sim-rs cluster is healthy"})
 
 # sleep infinity
 time.sleep(31536000)

crypto-benchmarks.rs/demo/ui/server.py

@@ -1,4 +1,12 @@
-from flask import Flask, send_from_directory, jsonify, render_template, abort, redirect, url_for
+from flask import (
+    Flask,
+    send_from_directory,
+    jsonify,
+    render_template,
+    abort,
+    redirect,
+    url_for,
+)
 import json
 import subprocess
 from pathlib import Path
@@ -22,7 +30,9 @@ def committee(run):
         abort(500, f"extract_committee.py not found at {script}")
 
     try:
-        out = subprocess.check_output(["python3", str(script), str(rdir)], cwd=ROOT, text=True)
+        out = subprocess.check_output(
+            ["python3", str(script), str(rdir)], cwd=ROOT, text=True
+        )
         data = json.loads(out)
         return jsonify(data)
     except subprocess.CalledProcessError as e:
@@ -48,11 +58,10 @@ def registry(run):
                     pools.append({"id": pid, "stake": s})
                     total_stake += s
 
-    return jsonify({
-        "pools": pools,
-        "total_pools": len(pools),
-        "total_stake": total_stake
-    })
+    return jsonify(
+        {"pools": pools, "total_pools": len(pools), "total_stake": total_stake}
+    )
+
 
 @app.route("/demo/<run>")
 def demo_for_run(run):
@@ -63,6 +72,7 @@ def demo_for_run(run):
         return send_from_directory(str(run_dir), "demo.json")
     abort(404, f"demo.json not found in {run_dir}")
 
+
 @app.route("/demo/<run>/<path:filename>")
 def demo_asset(run, filename):
     """Serve auxiliary files (eid.txt, ebhash.txt, etc.) from the run directory."""
@@ -72,15 +82,18 @@ def demo_asset(run, filename):
         return send_from_directory(str(run_dir), filename)
     abort(404, f"{filename} not found in {run_dir}")
 
+
 # === UI endpoint ===
 @app.route("/ui")
 def ui():
     return render_template("index.html")
 
+
 # Small helper route to redirect `/` to `/ui`
 @app.route("/")
 def root():
     return redirect(url_for("ui"))
 
+
 if __name__ == "__main__":
     app.run(host="0.0.0.0", port=5050, debug=True)

pre-commit-hooks.nix

@@ -1,3 +1,4 @@
+# Sourced by build.nix
 # https://github.com/cachix/git-hooks.nix?tab=readme-ov-file#hooks
 {
   # Nix
@@ -8,4 +9,8 @@
   # Haskell
   fourmolu.enable = true;
   hlint.enable = true;
+
+  # Python
+  black.enable = true;
+  ruff.enable = true;
 }

scripts/trace-translator/trace-translator.py

@@ -5,12 +5,15 @@
 in the format expected by the Leios trace verifier.
 """
 
-import json, sys
+import json
+import sys
 from datetime import datetime, timezone
 
+
 def log_message(message, time):
     print(json.dumps({"message": message, "time_s": time.total_seconds()}))
 
+
 fmt = "%Y-%m-%dT%H:%M:%S.%fZ"
 last_at = None
 last_SLC_slot = None
@@ -23,43 +26,39 @@ def log_message(message, time):
         exit(127)
 
     if last_at is None:
-        last_at = obj['at']
+        last_at = obj["at"]
 
-    curr_at = obj['at']
-    time = datetime.strptime(curr_at, fmt).replace(tzinfo=timezone.utc) \
-           -                                                            \
-           datetime.strptime(last_at, fmt).replace(tzinfo=timezone.utc)
+    curr_at = obj["at"]
+    time = datetime.strptime(curr_at, fmt).replace(
+        tzinfo=timezone.utc
+    ) - datetime.strptime(last_at, fmt).replace(tzinfo=timezone.utc)
     last_at = curr_at
 
-    if obj['ns'] == "Forge.Loop.AdoptedBlock":
+    if obj["ns"] == "Forge.Loop.AdoptedBlock":
         message = {
-            "type":             "RBGenerated",
-            "producer":         obj['host'],
-            "slot":             last_SLC_slot,
-            "id":               obj['data']['blockHash'],
-            "endorsement":      None,
-            "parent":           None, # FIXME: This is not available
-            "size":             obj['data']['blockSize'],
-            "tx_payload_bytes": None, # FIXME: This is not available
-            }
-    elif obj['ns'] == "BlockFetch.Client.CompletedBlockFetch":
+            "type": "RBGenerated",
+            "producer": obj["host"],
+            "slot": last_SLC_slot,
+            "id": obj["data"]["blockHash"],
+            "endorsement": None,
+            "parent": None,  # FIXME: This is not available
+            "size": obj["data"]["blockSize"],
+            "tx_payload_bytes": None,  # FIXME: This is not available
+        }
+    elif obj["ns"] == "BlockFetch.Client.CompletedBlockFetch":
         message = {
-            "type":      "RBReceived",
-            "recipient": obj['host'],
-            "id":        obj['data']['block']
-            }
+            "type": "RBReceived",
+            "recipient": obj["host"],
+            "id": obj["data"]["block"],
+        }
     else:
-        if obj['ns'] == "Forge.Loop.StartLeadershipCheck":
+        if obj["ns"] == "Forge.Loop.StartLeadershipCheck":
             if last_SLC_slot is None:
-                last_SLC_slot = obj['data']['slot']
+                last_SLC_slot = obj["data"]["slot"]
             else:
-                message = {
-                    "type": "Slot",
-                    "node": obj['host'],
-                    "slot": last_SLC_slot
-                    }
+                message = {"type": "Slot", "node": obj["host"], "slot": last_SLC_slot}
                 log_message(message, time)
-                last_SLC_slot = obj['data']['slot']
+                last_SLC_slot = obj["data"]["slot"]
         continue
 
     log_message(message, time)