fix: format code and do some cleanup

third_party/pyth/p2w_autoattest.py

@@ -1,23 +1,22 @@
 #!/usr/bin/env python3
 
 # This script sets up a simple loop for periodical attestation of Pyth data
-from pyth_utils import *
-
-from http.client import HTTPConnection
-from http.server import HTTPServer, BaseHTTPRequestHandler
-
 import json
 import os
 import re
-import subprocess
-import time
+import sys
 import threading
+import time
+from http.client import HTTPConnection
+from http.server import BaseHTTPRequestHandler, HTTPServer
 
+from pyth_utils import *
 
 P2W_ADDRESS = "P2WH424242424242424242424242424242424242424"
 P2W_ATTEST_INTERVAL = float(os.environ.get("P2W_ATTEST_INTERVAL", 5))
 P2W_OWNER_KEYPAIR = os.environ.get(
-    "P2W_OWNER_KEYPAIR", f"/usr/src/solana/keys/p2w_owner.json")
+    "P2W_OWNER_KEYPAIR", f'{"/usr/src/solana/keys/p2w_owner.json"}'
+)
 P2W_ATTESTATIONS_PORT = int(os.environ.get("P2W_ATTESTATIONS_PORT", 4343))
 
 PYTH_ACCOUNTS_HOST = "pyth"
@@ -29,6 +28,7 @@ ATTESTATIONS = {
     "pendingSeqnos": [],
 }
 
+
 class P2WAutoattestStatusEndpoint(BaseHTTPRequestHandler):
     """
     A dumb endpoint for last attested price metadata.
@@ -49,56 +49,87 @@ class P2WAutoattestStatusEndpoint(BaseHTTPRequestHandler):
         self.wfile.write(data)
         self.wfile.flush()
 
+
 def serve_attestations():
     """
     Run a barebones HTTP server to share Pyth2wormhole attestation history
     """
-    server_address = ('', P2W_ATTESTATIONS_PORT)
+    server_address = ("", P2W_ATTESTATIONS_PORT)
     httpd = HTTPServer(server_address, P2WAutoattestStatusEndpoint)
     httpd.serve_forever()
 
 
 # Get actor pubkeys
 P2W_OWNER_ADDRESS = sol_run_or_die(
-    "address", ["--keypair", P2W_OWNER_KEYPAIR], capture_output=True).stdout.strip()
+    "address", ["--keypair", P2W_OWNER_KEYPAIR], capture_output=True
+).stdout.strip()
 PYTH_OWNER_ADDRESS = sol_run_or_die(
-    "address", ["--keypair", PYTH_PROGRAM_KEYPAIR], capture_output=True).stdout.strip()
+    "address", ["--keypair", PYTH_PROGRAM_KEYPAIR], capture_output=True
+).stdout.strip()
 
 
 # Top up pyth2wormhole owner
-sol_run_or_die("airdrop", [
-    str(SOL_AIRDROP_AMT),
-    "--keypair",  P2W_OWNER_KEYPAIR,
-    "--commitment", "finalized",
-], capture_output=True)
+sol_run_or_die(
+    "airdrop",
+    [
+        str(SOL_AIRDROP_AMT),
+        "--keypair",
+        P2W_OWNER_KEYPAIR,
+        "--commitment",
+        "finalized",
+    ],
+    capture_output=True,
+)
 
 # Initialize pyth2wormhole
-init_result = run_or_die([
-    "pyth2wormhole-client",
-    "--log-level", "4",
-    "--p2w-addr", P2W_ADDRESS,
-    "--rpc-url", SOL_RPC_URL,
-    "--payer", P2W_OWNER_KEYPAIR,
-    "init",
-    "--wh-prog", WORMHOLE_ADDRESS,
-    "--owner", P2W_OWNER_ADDRESS,
-    "--pyth-owner", PYTH_OWNER_ADDRESS,
-], capture_output=True, die=False)
+init_result = run_or_die(
+    [
+        "pyth2wormhole-client",
+        "--log-level",
+        "4",
+        "--p2w-addr",
+        P2W_ADDRESS,
+        "--rpc-url",
+        SOL_RPC_URL,
+        "--payer",
+        P2W_OWNER_KEYPAIR,
+        "init",
+        "--wh-prog",
+        WORMHOLE_ADDRESS,
+        "--owner",
+        P2W_OWNER_ADDRESS,
+        "--pyth-owner",
+        PYTH_OWNER_ADDRESS,
+    ],
+    capture_output=True,
+    die=False,
+)
 
 if init_result.returncode != 0:
     print("NOTE: pyth2wormhole-client init failed, retrying with set_config")
-    run_or_die([
-        "pyth2wormhole-client",
-        "--log-level", "4",
-        "--p2w-addr", P2W_ADDRESS,
-        "--rpc-url", SOL_RPC_URL,
-        "--payer", P2W_OWNER_KEYPAIR,
-        "set-config",
-        "--owner", P2W_OWNER_KEYPAIR,
-        "--new-owner", P2W_OWNER_ADDRESS,
-        "--new-wh-prog", WORMHOLE_ADDRESS,
-        "--new-pyth-owner", PYTH_OWNER_ADDRESS,
-    ], capture_output=True)
+    run_or_die(
+        [
+            "pyth2wormhole-client",
+            "--log-level",
+            "4",
+            "--p2w-addr",
+            P2W_ADDRESS,
+            "--rpc-url",
+            SOL_RPC_URL,
+            "--payer",
+            P2W_OWNER_KEYPAIR,
+            "set-config",
+            "--owner",
+            P2W_OWNER_KEYPAIR,
+            "--new-owner",
+            P2W_OWNER_ADDRESS,
+            "--new-wh-prog",
+            WORMHOLE_ADDRESS,
+            "--new-pyth-owner",
+            PYTH_OWNER_ADDRESS,
+        ],
+        capture_output=True,
+    )
 
 # Retrieve current price/product pubkeys from the pyth publisher
 conn = HTTPConnection(PYTH_ACCOUNTS_HOST, PYTH_ACCOUNTS_PORT)
@@ -119,17 +150,27 @@ price_addr = pyth_accounts["price"]
 product_addr = pyth_accounts["product"]
 
 nonce = 0
-attest_result = run_or_die([
-    "pyth2wormhole-client",
-    "--log-level", "4",
-    "--p2w-addr", P2W_ADDRESS,
-    "--rpc-url", SOL_RPC_URL,
-    "--payer", P2W_OWNER_KEYPAIR,
-    "attest",
-    "--price", price_addr,
-    "--product", product_addr,
-    "--nonce", str(nonce),
-], capture_output=True)
+attest_result = run_or_die(
+    [
+        "pyth2wormhole-client",
+        "--log-level",
+        "4",
+        "--p2w-addr",
+        P2W_ADDRESS,
+        "--rpc-url",
+        SOL_RPC_URL,
+        "--payer",
+        P2W_OWNER_KEYPAIR,
+        "attest",
+        "--price",
+        price_addr,
+        "--product",
+        product_addr,
+        "--nonce",
+        str(nonce),
+    ],
+    capture_output=True,
+)
 
 print("p2w_autoattest ready to roll.")
 print(f"ACCOUNTS: {pyth_accounts}")
@@ -147,17 +188,27 @@ seqno_regex = re.compile(r"^Sequence number: (\d+)")
 
 nonce = 1
 while True:
-    attest_result = run_or_die([
-        "pyth2wormhole-client",
-        "--log-level", "4",
-        "--p2w-addr", P2W_ADDRESS,
-        "--rpc-url", SOL_RPC_URL,
-        "--payer", P2W_OWNER_KEYPAIR,
-        "attest",
-        "--price", price_addr,
-        "--product", product_addr,
-        "--nonce", str(nonce),
-    ], capture_output=True)
+    attest_result = run_or_die(
+        [
+            "pyth2wormhole-client",
+            "--log-level",
+            "4",
+            "--p2w-addr",
+            P2W_ADDRESS,
+            "--rpc-url",
+            SOL_RPC_URL,
+            "--payer",
+            P2W_OWNER_KEYPAIR,
+            "attest",
+            "--price",
+            price_addr,
+            "--product",
+            product_addr,
+            "--nonce",
+            str(nonce),
+        ],
+        capture_output=True,
+    )
     matches = seqno_regex.match(attest_result.stdout)
 
     if matches is not None:
@@ -167,9 +218,9 @@ while True:
         ATTESTATIONS["pendingSeqnos"].append(seqno)
 
     else:
-        print(f"Warning: Could not get sequence number")
+        print(f'{"Warning: Could not get sequence number"}')
 
     time.sleep(P2W_ATTEST_INTERVAL)
     nonce += 1
 
-readiness_thread.join()
+    readiness_thread.join()

third_party/pyth/pyth_utils.py

@@ -6,10 +6,12 @@ import subprocess
 PYTH = os.environ.get("PYTH", "./pyth")
 PYTH_KEY_STORE = os.environ.get("PYTH_KEY_STORE", "/home/pyth/.pythd")
 PYTH_PROGRAM_KEYPAIR = os.environ.get(
-    "PYTH_PROGRAM_KEYPAIR", f"{PYTH_KEY_STORE}/publish_key_pair.json")
+    "PYTH_PROGRAM_KEYPAIR", f"{PYTH_KEY_STORE}/publish_key_pair.json"
+)
 PYTH_PROGRAM_SO_PATH = os.environ.get("PYTH_PROGRAM_SO", "../target/oracle.so")
 PYTH_PUBLISHER_KEYPAIR = os.environ.get(
-    "PYTH_PUBLISHER_KEYPAIR", f"{PYTH_KEY_STORE}/publish_key_pair.json")
+    "PYTH_PUBLISHER_KEYPAIR", f"{PYTH_KEY_STORE}/publish_key_pair.json"
+)
 PYTH_PUBLISHER_INTERVAL = float(os.environ.get("PYTH_PUBLISHER_INTERVAL", "5"))
 
 SOL_AIRDROP_AMT = 100
@@ -24,7 +26,7 @@ def run_or_die(args, die=True, **kwargs):
     """
     Opinionated subprocess.run() call with fancy logging
     """
-    args_readable = ' '.join(args)
+    args_readable = " ".join(args)
     print(f"CMD RUN\t{args_readable}", file=sys.stderr)
     sys.stderr.flush()
     ret = subprocess.run(args, text=True, **kwargs)
@@ -54,23 +56,21 @@ def pyth_run_or_die(subcommand, args=[], debug=False, confirm=True, **kwargs):
     Pyth boilerplate in front of run_or_die
     """
     return run_or_die(
-        [PYTH, subcommand]
-        + args
-        + (["-d"] if debug else [])
+        [PYTH, subcommand] + args + (["-d"] if debug else [])
         # Note: not all pyth subcommands accept -n
         + ([] if confirm else ["-n"])
         + ["-k", PYTH_KEY_STORE]
         + ["-r", SOL_RPC_HOST]
-        + ["-c", "finalized"], **kwargs)
+        + ["-c", "finalized"],
+        **kwargs,
+    )
 
 
 def sol_run_or_die(subcommand, args=[], **kwargs):
     """
     Solana boilerplate in front of run_or_die
     """
-    return run_or_die(["solana", subcommand]
-                      + args
-                      + ["--url", SOL_RPC_URL], **kwargs)
+    return run_or_die(["solana", subcommand] + args + ["--url", SOL_RPC_URL], **kwargs)
 
 
 class ReadinessTCPHandler(socketserver.StreamRequestHandler):
@@ -83,6 +83,8 @@ def readiness():
     """
     Accept connections from readiness probe
     """
-    with socketserver.TCPServer(("0.0.0.0", READINESS_PORT), ReadinessTCPHandler) as srv:
+    with socketserver.TCPServer(
+        ("0.0.0.0", READINESS_PORT), ReadinessTCPHandler
+    ) as srv:
         srv.serve_forever()
     # run_or_die(["nc", "-k", "-l", "-p", READINESS_PORT])