45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
import base64
|
|
import os
|
|
import time
|
|
from typing import Dict, Any, List, Tuple
|
|
|
|
|
|
def b64(s: bytes) -> str:
|
|
return base64.b64encode(s).decode("ascii")
|
|
|
|
|
|
def ub64(s: str) -> bytes:
|
|
return base64.b64decode(s.encode("ascii"))
|
|
|
|
|
|
def make_random_bytes(size: int) -> bytes:
|
|
return os.urandom(size)
|
|
|
|
|
|
def monotonic_ms() -> int:
|
|
return int(time.monotonic() * 1000)
|
|
|
|
|
|
def assert_dict_has_keys(d: Dict[str, Any], keys: List[str]) -> None:
|
|
missing = [k for k in keys if k not in d]
|
|
assert not missing, f"Missing keys: {missing}; present: {list(d.keys())}"
|
|
|
|
|
|
def chunk_bytes(data: bytes, chunk_size: int) -> List[bytes]:
|
|
out: List[bytes] = []
|
|
for i in range(0, len(data), chunk_size):
|
|
out.append(data[i:i+chunk_size])
|
|
if len(data) == 0:
|
|
out.append(b"")
|
|
return out
|
|
|
|
|
|
def approx_eq_bytes(a: bytes, b: bytes, msg: str = "") -> None:
|
|
assert a == b, msg or f"bytes mismatch: len(a)={len(a)} len(b)={len(b)}"
|
|
|
|
|
|
def measure_throughput(op_name: str, size_bytes: int, elapsed_s: float) -> Tuple[float, str]:
|
|
if elapsed_s <= 0:
|
|
return float("inf"), f"{op_name}: {size_bytes} bytes in {elapsed_s:.6f}s = inf B/s"
|
|
thr = size_bytes / elapsed_s
|
|
return thr, f"{op_name}: {size_bytes} bytes in {elapsed_s:.6f}s = {thr:.2f} B/s" |