zkSNARK Utilities

Functions for ezkl circuit setup, witness generation, proof generation, and I/O serialization. Source: src/zkSNARK.py.

---

Setup

setup_prediction_zksnark

setup_prediction_zksnark(h, P, dy, CPCT_cov, y, jac_fh, jac_Ch, Kt, KCR_status, nonce_h, nonce_P, nonce_dy, nonce_CPCT_cov, nonce_KCR_status, nonce_kalman_diff, nonce_kalman_threshold, prediction_model, model_path, settings_path, compiled_model_path, vk_path, pk_path, py_run_args, data_path)

Exports the prediction model to ONNX and runs the full ezkl setup pipeline to produce a verification key (VK) and proving key (PK) for the prediction circuit.

Steps performed: ONNX export → ezkl.gen_settings() → ezkl.compile_circuit() → ezkl.setup(). Asserts that all output artefacts exist before returning.

Parameters:

Name	Type	Description	Default
h, P, dy, CPCT_cov, y, jac_fh, jac_Ch, Kt, KCR_status	Tensor	Representative input tensors used to trace the ONNX graph.	required
nonce_kalman_diff, nonce_kalman_threshold	Tensor	Nonce tensors appended to each circuit input for ezkl Poseidon hashing.	required
prediction_model	Module	The PredictionWrapper forward method to export.	required
model_path	str	Output path for the ONNX model file.	required
settings_path	str	Output path for the ezkl settings JSON.	required
compiled_model_path	str	Output path for the compiled circuit.	required
vk_path	str	Output path for the verification key.	required
pk_path	str	Output path for the proving key.	required
py_run_args		ezkl PyRunArgs object specifying circuit configuration (scale, etc.).	required
data_path	str	Path to the input data JSON (used only for reference).	required

Returns:

Type	Description
	Tuple[float, float, float, float]: (model_export_time, settings_generation_time, compile_circuit_time, circuit_setup_time) in seconds.

Source code in src/zkSNARK.py

def setup_prediction_zksnark(h, P,dy,CPCT_cov,y,jac_fh,jac_Ch,Kt,KCR_status,
                             nonce_h,nonce_P,nonce_dy,nonce_CPCT_cov,nonce_KCR_status,nonce_kalman_diff,nonce_kalman_threshold,
                             prediction_model, model_path,settings_path, compiled_model_path, vk_path, pk_path,py_run_args,data_path):
    """
    Exports the prediction model to ONNX and runs the full ezkl setup pipeline to
    produce a verification key (VK) and proving key (PK) for the prediction circuit.

    Steps performed: ONNX export → ezkl.gen_settings() → ezkl.compile_circuit() →
    ezkl.setup(). Asserts that all output artefacts exist before returning.

    Args:
        h, P, dy, CPCT_cov, y, jac_fh, jac_Ch, Kt, KCR_status (torch.Tensor):
            Representative input tensors used to trace the ONNX graph.
        nonce_h, nonce_P, nonce_dy, nonce_CPCT_cov, nonce_KCR_status,
        nonce_kalman_diff, nonce_kalman_threshold (torch.Tensor):
            Nonce tensors appended to each circuit input for ezkl Poseidon hashing.
        prediction_model (torch.nn.Module): The PredictionWrapper forward method to export.
        model_path (str): Output path for the ONNX model file.
        settings_path (str): Output path for the ezkl settings JSON.
        compiled_model_path (str): Output path for the compiled circuit.
        vk_path (str): Output path for the verification key.
        pk_path (str): Output path for the proving key.
        py_run_args: ezkl PyRunArgs object specifying circuit configuration (scale, etc.).
        data_path (str): Path to the input data JSON (used only for reference).

    Returns:
        Tuple[float, float, float, float]:
            (model_export_time, settings_generation_time, compile_circuit_time, circuit_setup_time)
            in seconds.
    """
    #Data Dump Order -> [h_array, P_array, dy_array, CPCT_cov_array, y_array, jac_fh_array, jac_Ch_array, Kt_array]

    model_export_time = time.time()
    torch.onnx.export(prediction_model, (h,P,dy,CPCT_cov,y,jac_fh,jac_Ch,Kt,KCR_status,nonce_h,nonce_P,nonce_dy,nonce_CPCT_cov,nonce_KCR_status,nonce_kalman_diff,nonce_kalman_threshold,), model_path,
                      input_names=["h","P","dy","CPCT_cov","y","jac_fh","jac_Ch","Kt"], output_names=["out"],
                      do_constant_folding=True, opset_version=15, export_params=True)
    model_export_time = time.time()-model_export_time

    settings_generation_time = time.time()
    res = ezkl.gen_settings(model_path, settings_path, py_run_args=py_run_args)
    assert res == True
    settings_generation_time = time.time() - settings_generation_time

    compile_circuit_time = time.time()
    res = ezkl.compile_circuit(model_path, compiled_model_path, settings_path)
    assert res == True
    compile_circuit_time = time.time() - compile_circuit_time

    circuit_setup_time = time.time()
    res = ezkl.setup(compiled_model_path, vk_path, pk_path,)
    circuit_setup_time = time.time() - circuit_setup_time

    assert res == True
    assert os.path.isfile(vk_path)
    assert os.path.isfile(pk_path)
    assert os.path.isfile(settings_path)

    return model_export_time,settings_generation_time,compile_circuit_time,circuit_setup_time

---

setup_hypothesis_zksnark

setup_hypothesis_zksnark(dy, CPCT_cov, U, S, KCR_status, hypothesis_model, model_path, settings_path, compiled_model_path, vk_path, pk_path, py_run_args)

Exports the hypothesis test model to ONNX and runs the full ezkl setup pipeline to produce a VK and PK for the hypothesis circuit.

Steps performed: ONNX export → ezkl.gen_settings() → ezkl.compile_circuit() → ezkl.setup(). The UCL is baked into the HypothesisTest model as a constant, so no CDF lookup table inputs are required.

Parameters:

Name	Type	Description	Default
dy	list[float]	DVAL-summed innovation vector (flat).	required
CPCT_cov	list[float]	DVAL-summed innovation covariance (flat).	required
U	Tensor	Left singular vectors of CPCT_cov.	required
S	Tensor	Diagonal matrix of inverse square-root singular values.	required
KCR_status	list[float]	Kalman check status carry-in (flat).	required
hypothesis_model	Module	The HypothesisTest module to export.	required
model_path	str	Output path for the ONNX model file.	required
settings_path	str	Output path for the ezkl settings JSON.	required
compiled_model_path	str	Output path for the compiled circuit.	required
vk_path	str	Output path for the verification key.	required
pk_path	str	Output path for the proving key.	required
py_run_args		ezkl PyRunArgs object specifying circuit configuration.	required

Returns:

Type	Description
	Tuple[float, float, float, float]: (model_export_time, settings_generation_time, compile_circuit_time, circuit_setup_time) in seconds.

Source code in src/zkSNARK.py

def setup_hypothesis_zksnark(dy, CPCT_cov,U,S, KCR_status, hypothesis_model, model_path, settings_path, compiled_model_path, vk_path, pk_path, py_run_args):
    """
    Exports the hypothesis test model to ONNX and runs the full ezkl setup pipeline
    to produce a VK and PK for the hypothesis circuit.

    Steps performed: ONNX export → ezkl.gen_settings() → ezkl.compile_circuit() →
    ezkl.setup(). The UCL is baked into the HypothesisTest model as a constant, so no
    CDF lookup table inputs are required.

    Args:
        dy (list[float]): DVAL-summed innovation vector (flat).
        CPCT_cov (list[float]): DVAL-summed innovation covariance (flat).
        U (torch.Tensor): Left singular vectors of CPCT_cov.
        S (torch.Tensor): Diagonal matrix of inverse square-root singular values.
        KCR_status (list[float]): Kalman check status carry-in (flat).
        hypothesis_model (torch.nn.Module): The HypothesisTest module to export.
        model_path (str): Output path for the ONNX model file.
        settings_path (str): Output path for the ezkl settings JSON.
        compiled_model_path (str): Output path for the compiled circuit.
        vk_path (str): Output path for the verification key.
        pk_path (str): Output path for the proving key.
        py_run_args: ezkl PyRunArgs object specifying circuit configuration.

    Returns:
        Tuple[float, float, float, float]:
            (model_export_time, settings_generation_time, compile_circuit_time, circuit_setup_time)
            in seconds.
    """
    dy_tensor = torch.tensor(dy)
    CPCT_cov_tensor = torch.tensor(CPCT_cov)
    U_tensor = torch.tensor(U)
    S_tensor = torch.tensor(S)
    KCR_status_tensor = torch.tensor(KCR_status)

    # torch.onnx.export(hypothesis_model, (dy_tensor, CPCT_cov_tensor,U_tensor,S_tensor,xs_tensor,cdf_values_tensor,KCR_status_tensor), model_path,
    #                   input_names=["dy", "CPCT_cov", "U", "S", "xs", "cdf_vals","kcr_status_check"], output_names=["out"], opset_version=15,
    #                   do_constant_folding=True, export_params=True)
    model_export_time = time.time()
    torch.onnx.export(hypothesis_model,(dy_tensor, CPCT_cov_tensor, U_tensor, S_tensor, KCR_status_tensor),
                      model_path,input_names=["dy", "CPCT_cov", "U", "S", "kcr_status_check"],output_names=["out"], opset_version=15,
                      do_constant_folding=True, export_params=True)
    model_export_time = time.time() - model_export_time

    settings_generation_time = time.time()
    res = ezkl.gen_settings(model_path, settings_path, py_run_args=py_run_args)
    assert res == True
    settings_generation_time = time.time() - settings_generation_time

    compile_circuit_time = time.time()
    res = ezkl.compile_circuit(model_path, compiled_model_path, settings_path)
    assert res == True
    compile_circuit_time = time.time() - compile_circuit_time

    circuit_setup_time = time.time()
    res = ezkl.setup(compiled_model_path, vk_path, pk_path, )
    circuit_setup_time = time.time() - circuit_setup_time

    assert res == True
    assert os.path.isfile(vk_path)
    assert os.path.isfile(pk_path)
    assert os.path.isfile(settings_path)

    return model_export_time,settings_generation_time,compile_circuit_time,circuit_setup_time

---

Prove & Witness

generate_prediction_zksnark

generate_prediction_zksnark(data_path, settings_path, compiled_model_path, witness_path, proof_path, pk_path, gen_proof=True)

Generates the witness and optionally the zkSNARK proof for the prediction circuit.

Downloads the SRS if needed, runs ezkl.gen_witness() to produce the witness file, then optionally runs ezkl.mock() (for sanity checking) and ezkl.prove() to produce a Plonk proof. Prints Poseidon hashes of all circuit inputs and outputs regardless of whether a proof is generated, and parses the witness to return the DVAL-summed residuals and covariance for passing into the hypothesis circuit.

Parameters:

Name	Type	Description	Default
data_path	str	Path to the JSON file containing serialized circuit inputs.	required
settings_path	str	Path to the ezkl settings JSON (used for SRS download).	required
compiled_model_path	str	Path to the compiled prediction circuit.	required
witness_path	str	Output path for the generated witness JSON.	required
proof_path	str	Output path for the generated proof file.	required
pk_path	str	Path to the proving key.	required
gen_proof	bool	If True, generates a proof; if False, only generates the witness.	True

Returns:

Type	Description
	Tuple[list, list, list, float, float]: - dy: DVAL-summed innovation vector as a flat float list. - CPCT_cov: DVAL-summed innovation covariance as a flat float list. - KCR_status: Kalman check status as a flat float list. - witness_generation_time: Seconds taken for witness generation. - prove_time: Seconds taken for proof generation (0 if gen_proof=False).

Source code in src/zkSNARK.py

def generate_prediction_zksnark(data_path, settings_path, compiled_model_path, witness_path, proof_path, pk_path,gen_proof=True):
    """
    Generates the witness and optionally the zkSNARK proof for the prediction circuit.

    Downloads the SRS if needed, runs ezkl.gen_witness() to produce the witness file,
    then optionally runs ezkl.mock() (for sanity checking) and ezkl.prove() to produce
    a Plonk proof. Prints Poseidon hashes of all circuit inputs and outputs regardless
    of whether a proof is generated, and parses the witness to return the DVAL-summed
    residuals and covariance for passing into the hypothesis circuit.

    Args:
        data_path (str): Path to the JSON file containing serialized circuit inputs.
        settings_path (str): Path to the ezkl settings JSON (used for SRS download).
        compiled_model_path (str): Path to the compiled prediction circuit.
        witness_path (str): Output path for the generated witness JSON.
        proof_path (str): Output path for the generated proof file.
        pk_path (str): Path to the proving key.
        gen_proof (bool): If True, generates a proof; if False, only generates the witness.

    Returns:
        Tuple[list, list, list, float, float]:
            - dy: DVAL-summed innovation vector as a flat float list.
            - CPCT_cov: DVAL-summed innovation covariance as a flat float list.
            - KCR_status: Kalman check status as a flat float list.
            - witness_generation_time: Seconds taken for witness generation.
            - prove_time: Seconds taken for proof generation (0 if gen_proof=False).
    """
    print(f"data_path:{data_path}")
    print(f"settings_path:{settings_path}")
    print(f"compiled_model_path:{compiled_model_path}")
    print(f"witness_path:{witness_path}")
    print(f"proof_path:{proof_path}")
    print(f"pk_path:{pk_path}")

    witness_generation_time = time.time()
    # compile witness
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    res = loop.run_until_complete(async_srs(settings_path))
    #print(res)
    res = loop.run_until_complete(async_compile(data_path, compiled_model_path, witness_path))
    #print(res)
    loop.close()
    assert os.path.isfile(witness_path)
    witness_generation_time = time.time()-witness_generation_time

    with open(witness_path, 'r') as file:
        prediction_output = json.load(file)

    prove_time=0
    if gen_proof:
        # print(f"----------------------\tPredictionMockingFunction\t---------------------------------")
        print_section_header("PredictionMockingFunction")
        result = ezkl.mock(witness_path, compiled_model_path)
        print("Mock result:", result)
        print_section_header("PredictionProvingFunction")
        #print(f"----------------------\tPredictionProvingFunction\t---------------------------------")
        # generate proof
        prove_time = time.time()
        res = ezkl.prove(witness_path, compiled_model_path, pk_path, proof_path, "single",)
        prove_time = time.time() - prove_time
        print_section_header("StatePredictionResults")
        #h, P, delta_y, CPCT_cov, y, jac_fh, jac_Ch, Kt, KCR_status,
        #nonce_h,nonce_P,nonce_dy,nonce_CPCT_cov,nonce_KCR_status,nonce_kalman_diff,nonce_kalman_threshold,
        print(f"INPUT:  hash(current_h):                        {res['instances'][0][0]}")
        print(f"INPUT:  hash(current_P):                        {res['instances'][0][1]}")
        print(f"INPUT:  hash(current_residual):                 {res['instances'][0][2]}")
        print(f"INPUT:  hash(current_residual_covariance):      {res['instances'][0][3]}")
        print(f"INPUT:  hash(current_y):                        {res['instances'][0][4]}")
        print(f"INPUT:  hash(current_jac_fh):                   {res['instances'][0][5]}")
        print(f"INPUT:  hash(current_jac_Ch):                   {res['instances'][0][6]}")
        print(f"INPUT:  hash(current_Kt):                       {res['instances'][0][7]}")
        print(f"INPUT:  hash(current_KCR_status):               {res['instances'][0][8]}")
        print(f"OUTPUT: hash(predicted_h):                      {res['instances'][0][16]}")
        print(f"OUTPUT: hash(predicted_P):                      {res['instances'][0][17]}")
        print(f"OUTPUT: hash(predicted_residual):               {res['instances'][0][18]}")
        print(f"OUTPUT: hash(predicted_residual_covariance):    {res['instances'][0][19]}")
        print(f"OUTPUT: hash(predicted_KCR_status):             {res['instances'][0][20]}")
        assert os.path.isfile(proof_path)

    else:
        print(f"INPUT:  hash(current_h):                        {prediction_output['processed_inputs']['poseidon_hash'][0]}")
        print(f"INPUT:  hash(current_P):                        {prediction_output['processed_inputs']['poseidon_hash'][1]}")
        print(f"INPUT:  hash(current_residual):                 {prediction_output['processed_inputs']['poseidon_hash'][2]}")
        print(f"INPUT:  hash(current_residual_covariance):      {prediction_output['processed_inputs']['poseidon_hash'][3]}")
        print(f"INPUT:  hash(current_y):                        {prediction_output['processed_inputs']['poseidon_hash'][4]}")
        print(f"INPUT:  hash(current_jac_fh):                   {prediction_output['processed_inputs']['poseidon_hash'][5]}")
        print(f"INPUT:  hash(current_jac_Ch):                   {prediction_output['processed_inputs']['poseidon_hash'][6]}")
        print(f"INPUT:  hash(current_Kt):                       {prediction_output['processed_inputs']['poseidon_hash'][7]}")
        print(f"INPUT:  hash(current_KCR_status):               {prediction_output['processed_inputs']['poseidon_hash'][8]}")
        print(f"OUTPUT: hash(predicted_h):                      {prediction_output['processed_outputs']['poseidon_hash'][0]}")
        print(f"OUTPUT: hash(predicted_P):                      {prediction_output['processed_outputs']['poseidon_hash'][1]}")
        print(f"OUTPUT: hash(predicted_residual):               {prediction_output['processed_outputs']['poseidon_hash'][2]}")
        print(f"OUTPUT: hash(predicted_residual_covariance):    {prediction_output['processed_outputs']['poseidon_hash'][3]}")
        print(f"OUTPUT: hash(predicted_KCR_status):             {prediction_output['processed_outputs']['poseidon_hash'][4]}")

    dy_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][2]
    CPCT_cov_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][3]
    KCR_status_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][4]

    # h_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][0]
    # h = list(map(float, list(prediction_output ["pretty_elements"]["rescaled_outputs"][0])))
    # dy = list(map(float, list (prediction_output ["pretty_elements"]["rescaled_outputs"][1])))
    # CPCT_cov = list(map(float, list (prediction_output ["pretty_elements"]["rescaled_outputs"][2])))
    # outputs = [["-0.0625","-0.078125","-0.046875","-0.046875"],
    #            ["0.4375","0.6640625","-0.3515625","0.1875","0.609375"],
    #            ["6.84375","1.34375","1.2578125","1.1328125","0.984375","1.34375","6.6640625","1.421875","1.3046875","0.921875","1.2578125","1.421875","7","1.375","0.8671875","1.1328125","1.3046875","1.375","6.7109375","0.984375","0.984375","0.921875","0.859375","0.984375","6.1171875"]]

    dy = [float(x) for x in dy_outputs]
    CPCT_cov = [float(x) for x in CPCT_cov_outputs]
    KCR_status = [float(x) for x in KCR_status_outputs]

    return dy, CPCT_cov, KCR_status, witness_generation_time,prove_time

---

generate_hypothesis_zksnark

generate_hypothesis_zksnark(data_path, settings_path, compiled_model_path, witness_path, proof_path, pk_path, gen_proof=True)

Generates the witness and optionally the zkSNARK proof for the hypothesis circuit.

Downloads the SRS if needed, runs ezkl.gen_witness() to produce the witness file, then optionally runs ezkl.prove() to generate a Plonk proof. Prints Poseidon hashes of all circuit inputs (residual, covariance, U, S, KCR status) in both modes.

Parameters:

Name	Type	Description	Default
data_path	str	Path to the JSON file containing serialized circuit inputs.	required
settings_path	str	Path to the ezkl settings JSON (used for SRS download).	required
compiled_model_path	str	Path to the compiled hypothesis circuit.	required
witness_path	str	Output path for the generated witness JSON.	required
proof_path	str	Output path for the generated proof file.	required
pk_path	str	Path to the proving key.	required
gen_proof	bool	If True, generates a proof; if False, only generates the witness.	True

Returns:

Type	Description
	Tuple[float, float]: - witness_generation_time: Seconds taken for witness generation. - prove_time: Seconds taken for proof generation (0 if gen_proof=False).

Source code in src/zkSNARK.py

def generate_hypothesis_zksnark(data_path, settings_path, compiled_model_path, witness_path, proof_path, pk_path,gen_proof=True):
    """
    Generates the witness and optionally the zkSNARK proof for the hypothesis circuit.

    Downloads the SRS if needed, runs ezkl.gen_witness() to produce the witness file,
    then optionally runs ezkl.prove() to generate a Plonk proof. Prints Poseidon hashes
    of all circuit inputs (residual, covariance, U, S, KCR status) in both modes.

    Args:
        data_path (str): Path to the JSON file containing serialized circuit inputs.
        settings_path (str): Path to the ezkl settings JSON (used for SRS download).
        compiled_model_path (str): Path to the compiled hypothesis circuit.
        witness_path (str): Output path for the generated witness JSON.
        proof_path (str): Output path for the generated proof file.
        pk_path (str): Path to the proving key.
        gen_proof (bool): If True, generates a proof; if False, only generates the witness.

    Returns:
        Tuple[float, float]:
            - witness_generation_time: Seconds taken for witness generation.
            - prove_time: Seconds taken for proof generation (0 if gen_proof=False).
    """
    print(f"data_path:{data_path}")
    print(f"settings_path:{settings_path}")
    print(f"compiled_model_path:{compiled_model_path}")
    print(f"witness_path:{witness_path}")
    print(f"pk_path:{pk_path}")

    witness_generation_time = time.time()
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    res = loop.run_until_complete(async_srs(settings_path))
    res = loop.run_until_complete(async_compile(data_path, compiled_model_path, witness_path))
    loop.close()
    assert os.path.isfile(witness_path)
    witness_generation_time = time.time() - witness_generation_time

    with open(witness_path, 'r') as file:
        hypothesis_output = json.load(file)

    prove_time=0

    if gen_proof:
        print_section_header("HypothesisProvingFunction")
        prove_time = time.time()
        res = ezkl.prove(witness_path, compiled_model_path, pk_path, proof_path, "single", )
        prove_time = time.time() - prove_time

        print_section_header("HypothesisTestResults")
        print(f"INPUT:  hash(predicted_residual):             {res['instances'][0][0]}")
        print(f"INPUT:  hash(predicted_residual_covariance):  {res['instances'][0][1]}")
        print(f"INPUT:  hash(predicted_U):                    {res['instances'][0][2]}")
        print(f"INPUT:  hash(predicted_S):                    {res['instances'][0][3]}")
        print(f"INPUT:  hash(kcr_status_check):               {res['instances'][0][4]}")

        assert os.path.isfile(proof_path)

    else:
        print_section_header("HypothesisTestResults")
        print(f"INPUT:  hash(predicted_residual):             {hypothesis_output['processed_inputs']['poseidon_hash'][0]}")
        print(f"INPUT:  hash(predicted_residual_covariance):  {hypothesis_output['processed_inputs']['poseidon_hash'][1]}")
        print(f"INPUT:  hash(predicted_U):                    {hypothesis_output['processed_inputs']['poseidon_hash'][2]}")
        print(f"INPUT:  hash(predicted_S):                    {hypothesis_output['processed_inputs']['poseidon_hash'][3]}")
        print(f"INPUT:  hash(kcr_status_check):               {hypothesis_output['processed_inputs']['poseidon_hash'][4]}")

    return witness_generation_time,prove_time

---

I/O Serialization

dump_state_estimate

dump_state_estimate(h_se, P_se, dy_se, CPCT_cov_se, y_se, jac_fh_se, jac_Ch_se, Kt_se, KCR_se, nonce_h_se, nonce_P_se, nonce_dy_se, nonce_CPCT_cov_se, nonce_KCR_status_se, nonce_kalman_diff_se, nonce_kalman_threshold_se, prediction_data_path, reset_residual)

Serializes prediction circuit inputs to a JSON file in the format expected by ezkl.

Converts all input tensors (state, covariance, residuals, Jacobians, Kalman gain, nonce columns) to flat float lists via convert_tensor_to_float_array, then writes them as the input_data array to prediction_data_path. If reset_residual=True, zeros out dy, CPCT_cov, and KCR_status before writing (used at the start of each new DVAL interval).

Parameters:

Name	Type	Description	Default
h_se, P_se, dy_se, CPCT_cov_se, y_se, jac_fh_se, jac_Ch_se, Kt_se, KCR_se		Sequences of tensors for each circuit input field.	required
nonce_kalman_diff_se, nonce_kalman_threshold_se		Sequences of nonce tensors for each circuit input field.	required
prediction_data_path	str	Output path for the JSON input data file.	required
reset_residual	bool	If True, zero out dy, CPCT_cov, and KCR_status arrays before writing (marks the start of a fresh DVAL accumulation interval).	required

Source code in src/zkSNARK.py

def dump_state_estimate(h_se,P_se,dy_se,CPCT_cov_se,y_se,jac_fh_se,jac_Ch_se,Kt_se,KCR_se,
                        nonce_h_se,nonce_P_se,nonce_dy_se,nonce_CPCT_cov_se,nonce_KCR_status_se,nonce_kalman_diff_se,nonce_kalman_threshold_se,
                        prediction_data_path,reset_residual):
    """
    Serializes prediction circuit inputs to a JSON file in the format expected by ezkl.

    Converts all input tensors (state, covariance, residuals, Jacobians, Kalman gain,
    nonce columns) to flat float lists via convert_tensor_to_float_array, then writes
    them as the input_data array to prediction_data_path. If reset_residual=True,
    zeros out dy, CPCT_cov, and KCR_status before writing (used at the start of each
    new DVAL interval).

    Args:
        h_se, P_se, dy_se, CPCT_cov_se, y_se, jac_fh_se, jac_Ch_se, Kt_se, KCR_se:
            Sequences of tensors for each circuit input field.
        nonce_h_se, nonce_P_se, nonce_dy_se, nonce_CPCT_cov_se, nonce_KCR_status_se,
        nonce_kalman_diff_se, nonce_kalman_threshold_se:
            Sequences of nonce tensors for each circuit input field.
        prediction_data_path (str): Output path for the JSON input data file.
        reset_residual (bool): If True, zero out dy, CPCT_cov, and KCR_status arrays
            before writing (marks the start of a fresh DVAL accumulation interval).
    """
    h_array = convert_tensor_to_float_array(h_se)
    P_array = convert_tensor_to_float_array(P_se)
    dy_array = convert_tensor_to_float_array(dy_se)
    CPCT_cov_array = convert_tensor_to_float_array(CPCT_cov_se)
    y_array = convert_tensor_to_float_array(y_se)
    jac_fh_array = convert_tensor_to_float_array(jac_fh_se)
    jac_Ch_array = convert_tensor_to_float_array(jac_Ch_se)
    Kt_array = convert_tensor_to_float_array(Kt_se)
    KCR_status_array = convert_tensor_to_float_array(KCR_se)

    nonce_h_array = convert_tensor_to_float_array(nonce_h_se)
    nonce_P_array = convert_tensor_to_float_array(nonce_P_se)
    nonce_CPCT_cov_array = convert_tensor_to_float_array(nonce_CPCT_cov_se)
    nonce_dy_array = convert_tensor_to_float_array(nonce_dy_se)
    nonce_KCR_status_array = convert_tensor_to_float_array(nonce_KCR_status_se)
    nonce_kalman_diff_se = convert_tensor_to_float_array(nonce_kalman_diff_se)
    nonce_kalman_threshold_se = convert_tensor_to_float_array(nonce_kalman_threshold_se)

    if reset_residual:
        #print(f"----------------------\tResettingResidual\t---------------------------------")
        print_section_header("ResettingResidual")
        dy_array = [0.0 for x in dy_array]
        CPCT_cov_array = [0.0 for x in CPCT_cov_array]
        KCR_status_array = [1.0 for x in KCR_status_array]

    # print("h_se.shape:", h_se.shape, "len(h_array):", len(h_array))
    # print("P_se.shape:", P_se.shape, "len(P_array):", len(P_array))
    # print("dy_se.shape:", dy_se.shape, "len(dy_array):", len(dy_array))
    # print("CPCT_cov_se.shape:", CPCT_cov_se.shape, "len(CPCT_cov_array):", len(CPCT_cov_array))
    # print("y_se.shape:", y_se.shape, "len(y_array):", len(y_array))
    # print("jac_fh_se.shape:", jac_fh_se.shape, "len(jac_fh_array):", len(jac_fh_array))
    # print("jac_Ch_se.shape:", jac_Ch_se.shape, "len(jac_Ch_array):", len(jac_Ch_array))
    # print("Kt_se.shape:", Kt_se.shape, "len(Kt_array):", len(Kt_array))

    input_data_dict = dict(input_data=[h_array, P_array, dy_array, CPCT_cov_array, y_array, jac_fh_array, jac_Ch_array, Kt_array,KCR_status_array,
                                       nonce_h_array,nonce_P_array,nonce_dy_array,nonce_CPCT_cov_array,nonce_KCR_status_array,nonce_kalman_diff_se,nonce_kalman_threshold_se])
    json.dump(input_data_dict, open(prediction_data_path, 'w'))

---

extract_state_estimate

extract_state_estimate(SD, MD, prediction_witness_path)

Parses prediction circuit outputs from a witness JSON and returns them as tensors.

Reads the rescaled outputs from 'pretty_elements', converts them to floats, and reshapes each into the padded tensor format used as inputs for the next prediction circuit call (i.e., with one trailing nonce column/row).

Parameters:

Name	Type	Description	Default
SD	int	State space dimension.	required
MD	int	Measurement space dimension.	required
prediction_witness_path	str	Path to the prediction witness JSON file.	required

Returns:

Name	Type	Description
Tuple		h (torch.Tensor): Updated hidden state (1, 1, SD+1). P (torch.Tensor): Updated state covariance (1, SD, SD+1). dy (torch.Tensor): DVAL-summed residual (1, 1, MD+1). CPCT_cov (torch.Tensor): DVAL-summed innovation covariance (1, MD, MD+1). KCR_status (torch.Tensor): Kalman check status (1, 1, 2). KCR_status_array (list[float]): KCR status as a flat float list. kalman_diff_array (list[float]): Kalman diff as a flat float list. kalman_threshold_array (list[float]): Kalman threshold as a flat float list.

Source code in src/zkSNARK.py

def extract_state_estimate(SD,MD,prediction_witness_path,):
    """
    Parses prediction circuit outputs from a witness JSON and returns them as tensors.

    Reads the rescaled outputs from 'pretty_elements', converts them to floats, and
    reshapes each into the padded tensor format used as inputs for the next prediction
    circuit call (i.e., with one trailing nonce column/row).

    Args:
        SD (int): State space dimension.
        MD (int): Measurement space dimension.
        prediction_witness_path (str): Path to the prediction witness JSON file.

    Returns:
        Tuple:
            - h (torch.Tensor): Updated hidden state (1, 1, SD+1).
            - P (torch.Tensor): Updated state covariance (1, SD, SD+1).
            - dy (torch.Tensor): DVAL-summed residual (1, 1, MD+1).
            - CPCT_cov (torch.Tensor): DVAL-summed innovation covariance (1, MD, MD+1).
            - KCR_status (torch.Tensor): Kalman check status (1, 1, 2).
            - KCR_status_array (list[float]): KCR status as a flat float list.
            - kalman_diff_array (list[float]): Kalman diff as a flat float list.
            - kalman_threshold_array (list[float]): Kalman threshold as a flat float list.
    """
    with open(prediction_witness_path, 'r') as file:
        prediction_output = json.load(file)

    h_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][0]
    P_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][1]
    dy_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][2]
    CPCT_cov_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][3]
    KCR_status_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][4]
    kalman_diff_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][5]
    kalman_threshold_outputs = prediction_output["pretty_elements"]["rescaled_outputs"][6]

    h = [float(x) for x in h_outputs]
    P = [float(x) for x in P_outputs]
    dy = [float(x) for x in dy_outputs]
    CPCT_cov = [float(x) for x in CPCT_cov_outputs]
    KCR_status_array = [float(x) for x in KCR_status_outputs]
    kalman_diff_array = [float(x) for x in kalman_diff_outputs]
    kalman_threshold_array = [float(x) for x in kalman_threshold_outputs]

    h = torch.tensor(h).reshape(1, 1, SD+1)
    P = torch.tensor(P).reshape(1, SD, SD+1)
    dy = torch.tensor(dy).reshape(1, 1, MD+1)
    CPCT_cov = torch.tensor(CPCT_cov).reshape(1, MD, MD+1)
    # kalman_diff = torch.tensor(kalman_diff_array).reshape(1, MD, SD)
    # kalman_threshold = torch.tensor(kalman_threshold_array).reshape(1, MD, SD)
    KCR_status = torch.tensor(KCR_status_array).reshape(1, 1, 2)

    return h, P, dy, CPCT_cov, KCR_status, KCR_status_array, kalman_diff_array, kalman_threshold_array

---

dump_hypothesis_test_inputs

dump_hypothesis_test_inputs(dy, CPCT_cov, U, S, KCR_status, data_path)

Serializes hypothesis circuit inputs to a JSON file in the format expected by ezkl.

Flattens U and S and writes the input_data dict (containing dy, CPCT_cov, U, S, KCR_status) to data_path. The UCL is baked into the compiled circuit, so no CDF lookup table inputs are included.

Parameters:

Name	Type	Description	Default
dy	list[float]	DVAL-summed innovation vector (flat).	required
CPCT_cov	list[float]	DVAL-summed innovation covariance (flat).	required
U	Tensor	Left singular vectors of CPCT_cov.	required
S	Tensor	Diagonal inverse square-root singular value matrix.	required
KCR_status	list[float]	Kalman check status as a flat float list.	required
data_path	str	Output path for the JSON input data file.	required

Source code in src/zkSNARK.py

def dump_hypothesis_test_inputs(dy, CPCT_cov,U,S,KCR_status,data_path):
    """
    Serializes hypothesis circuit inputs to a JSON file in the format expected by ezkl.

    Flattens U and S and writes the input_data dict (containing dy, CPCT_cov, U, S,
    KCR_status) to data_path. The UCL is baked into the compiled circuit, so no CDF
    lookup table inputs are included.

    Args:
        dy (list[float]): DVAL-summed innovation vector (flat).
        CPCT_cov (list[float]): DVAL-summed innovation covariance (flat).
        U (torch.Tensor): Left singular vectors of CPCT_cov.
        S (torch.Tensor): Diagonal inverse square-root singular value matrix.
        KCR_status (list[float]): Kalman check status as a flat float list.
        data_path (str): Output path for the JSON input data file.
    """
    U = U.flatten().tolist()
    S = S.flatten().tolist()
    #hypothesis_input_data = dict(input_data=[dy, CPCT_cov,U,S,xs, cdf_values,KCR_status])
    hypothesis_input_data = dict(input_data=[dy, CPCT_cov, U, S, KCR_status])
    json.dump(hypothesis_input_data, open(data_path, 'w'))

---

extract_hypothesis_test_outputs

extract_hypothesis_test_outputs(hypothesis_proof_path, gen_proof=True)

Parses hypothesis circuit outputs from either a proof file or a witness file.

When gen_proof=True, reads public outputs from 'pretty_public_inputs' (proof format). When gen_proof=False, reads from 'pretty_elements' (witness format). Extracts the six circuit outputs in order: alarm status, SVD check, KCR check, chi-squared statistic, SVD diff, and SVD threshold.

Parameters:

Name	Type	Description	Default
hypothesis_proof_path	str	Path to the proof JSON (gen_proof=True) or witness JSON (gen_proof=False).	required
gen_proof	bool	Determines which JSON key to use for output extraction.	True

Returns:

Type	Description
	Tuple[list, list, list, list, list, list]: (alarm_status_output, SVD_check_output, KCR_check_output, chi_square_test_statistic_output, svd_diff_output, svd_threshold_output) Each is a list of rescaled string values from ezkl's pretty output.

Source code in src/zkSNARK.py

def extract_hypothesis_test_outputs(hypothesis_proof_path,gen_proof=True):
    """
    Parses hypothesis circuit outputs from either a proof file or a witness file.

    When gen_proof=True, reads public outputs from 'pretty_public_inputs' (proof format).
    When gen_proof=False, reads from 'pretty_elements' (witness format). Extracts the
    six circuit outputs in order: alarm status, SVD check, KCR check, chi-squared
    statistic, SVD diff, and SVD threshold.

    Args:
        hypothesis_proof_path (str): Path to the proof JSON (gen_proof=True) or
            witness JSON (gen_proof=False).
        gen_proof (bool): Determines which JSON key to use for output extraction.

    Returns:
        Tuple[list, list, list, list, list, list]:
            (alarm_status_output, SVD_check_output, KCR_check_output,
             chi_square_test_statistic_output, svd_diff_output, svd_threshold_output)
            Each is a list of rescaled string values from ezkl's pretty output.
    """
    with open(hypothesis_proof_path, 'r') as file:
        hypothesis_output = json.load(file)
    if gen_proof:
        # alpha_output = hypothesis_output["pretty_elements"]["rescaled_outputs"][0]
        alarm_status_output = hypothesis_output['pretty_public_inputs']["rescaled_outputs"][0]
        SVD_check_output = hypothesis_output['pretty_public_inputs']["rescaled_outputs"][1]
        KCR_check_output = hypothesis_output['pretty_public_inputs']["rescaled_outputs"][2]
        chi_square_test_statistic_output = hypothesis_output['pretty_public_inputs']["rescaled_outputs"][3]
        svd_diff_output = hypothesis_output['pretty_public_inputs']["rescaled_outputs"][4]
        svd_threshold_output = hypothesis_output['pretty_public_inputs']["rescaled_outputs"][5]

    else:
        # alpha_output = hypothesis_output["pretty_elements"]["rescaled_outputs"][0]
        alarm_status_output = hypothesis_output['pretty_elements']["rescaled_outputs"][0]
        SVD_check_output = hypothesis_output['pretty_elements']["rescaled_outputs"][1]
        KCR_check_output = hypothesis_output['pretty_elements']["rescaled_outputs"][2]
        chi_square_test_statistic_output = hypothesis_output['pretty_elements']["rescaled_outputs"][3]
        svd_diff_output = hypothesis_output['pretty_elements']["rescaled_outputs"][4]
        svd_threshold_output = hypothesis_output['pretty_elements']["rescaled_outputs"][5]

    return alarm_status_output, SVD_check_output, KCR_check_output, chi_square_test_statistic_output, svd_diff_output, svd_threshold_output

---

dump_check_results

dump_check_results(filename, svd_results)

Writes SVD and Kalman check results to a JSON file.

Parameters:

Name	Type	Description	Default
filename	str	Output path for the JSON file.	required
svd_results	dict	Dictionary of check result data to serialize.	required

Source code in src/zkSNARK.py

def dump_check_results(filename,svd_results):
    """
    Writes SVD and Kalman check results to a JSON file.

    Args:
        filename (str): Output path for the JSON file.
        svd_results (dict): Dictionary of check result data to serialize.
    """
    with open(filename, "w") as f:
        json.dump(svd_results, f, indent=2)

---

Utilities

convert_tensor_to_float_array

convert_tensor_to_float_array(h_tensor)

Flattens a sequence of tensors into a single Python list of floats.

Detaches each tensor from the computation graph, moves to CPU, flattens, concatenates, and converts to a list. Used to serialize circuit inputs and outputs into the JSON format expected by ezkl.

Parameters:

Name	Type	Description	Default
h_tensor	Sequence[Tensor]	A list or tuple of tensors to flatten.	required

Returns:

Type	Description
	list[float]: Concatenated flat list of all tensor values.

Source code in src/zkSNARK.py

def convert_tensor_to_float_array(h_tensor):
    """
    Flattens a sequence of tensors into a single Python list of floats.

    Detaches each tensor from the computation graph, moves to CPU, flattens,
    concatenates, and converts to a list. Used to serialize circuit inputs and
    outputs into the JSON format expected by ezkl.

    Args:
        h_tensor (Sequence[torch.Tensor]): A list or tuple of tensors to flatten.

    Returns:
        list[float]: Concatenated flat list of all tensor values.
    """
    h_flat_tensor = [t.detach().cpu().flatten() for t in h_tensor]
    h_flat_tensor = torch.cat(h_flat_tensor, dim=0)
    h = h_flat_tensor.tolist()
    return h

---

dump_setup_timing_csv

dump_setup_timing_csv(filename, model_export_time, settings_generation_time, compile_circuit_time, circuit_setup_time)

Appends a single-row timing record for the zkSNARK setup phase to a CSV file.

Creates the file with a header row on first call; subsequent calls append data only.

Parameters:

Name	Type	Description	Default
filename	str	Path to the output CSV file.	required
model_export_time	float	Seconds taken to export the model to ONNX.	required
settings_generation_time	float	Seconds taken by ezkl.gen_settings().	required
compile_circuit_time	float	Seconds taken by ezkl.compile_circuit().	required
circuit_setup_time	float	Seconds taken by ezkl.setup() (VK/PK generation).	required

Source code in src/zkSNARK.py

def dump_setup_timing_csv(filename, model_export_time,
                          settings_generation_time,
                          compile_circuit_time,
                          circuit_setup_time):
    """
    Appends a single-row timing record for the zkSNARK setup phase to a CSV file.

    Creates the file with a header row on first call; subsequent calls append data only.

    Args:
        filename (str): Path to the output CSV file.
        model_export_time (float): Seconds taken to export the model to ONNX.
        settings_generation_time (float): Seconds taken by ezkl.gen_settings().
        compile_circuit_time (float): Seconds taken by ezkl.compile_circuit().
        circuit_setup_time (float): Seconds taken by ezkl.setup() (VK/PK generation).
    """
    headers = [
        "model_export_time",
        "settings_generation_time",
        "compile_circuit_time",
        "circuit_setup_time"
    ]
    values = [
        model_export_time,
        settings_generation_time,
        compile_circuit_time,
        circuit_setup_time
    ]

    write_header = not os.path.exists(filename)

    with open(filename, mode='a', newline='') as file:
        writer = csv.writer(file)
        if write_header:
            writer.writerow(headers)
        writer.writerow(values)

---

dump_proving_times

dump_proving_times(filename, nested_dict)

Appends per-period proving time metrics from a nested dict to a CSV file.

The dict maps period identifiers to dicts of metric name → value. Creates the header row on first call; subsequent calls append data rows only. Returns immediately if nested_dict is empty.

Parameters:

Name	Type	Description	Default
filename	str	Path to the output CSV file.	required
nested_dict	dict	Mapping of {period: {metric_name: value, ...}}.	required

Source code in src/zkSNARK.py

def dump_proving_times(filename, nested_dict):
    """
    Appends per-period proving time metrics from a nested dict to a CSV file.

    The dict maps period identifiers to dicts of metric name → value. Creates
    the header row on first call; subsequent calls append data rows only. Returns
    immediately if nested_dict is empty.

    Args:
        filename (str): Path to the output CSV file.
        nested_dict (dict): Mapping of {period: {metric_name: value, ...}}.
    """
    if not nested_dict:
        return

        # Flatten the data
    rows = [
        {"period": period, **metrics}
        for period, metrics in nested_dict.items()
    ]

    # Get all unique field names
    fieldnames = ["period"] + list(next(iter(nested_dict.values())).keys())

    write_header = not os.path.exists(filename)
    with open(filename, mode='a', newline='') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        if write_header:
            writer.writeheader()
        writer.writerows(rows)

---

print_section_header

print_section_header(title, width=80, char='=')

Prints a centered section header banner to stdout.

Parameters:

Name	Type	Description	Default
title	str	The section label to center in the banner.	required
width	int	Total banner width in characters. Default 80.	80
char	str	Character used to draw the top and bottom border. Default '='.	'='

Source code in src/zkSNARK.py

def print_section_header(title, width=80, char="="):
    """
    Prints a centered section header banner to stdout.

    Args:
        title (str): The section label to center in the banner.
        width (int): Total banner width in characters. Default 80.
        char (str): Character used to draw the top and bottom border. Default '='.
    """
    print(char * width)
    print(f"{title:^{width}}")
    print(char * width)

---

async_compile async

async_compile(data_path, compiled_model_path, witness_path)

Async wrapper around ezkl.gen_witness() for use in an asyncio event loop.

Parameters:

Name	Type	Description	Default
data_path	str	Path to the JSON file containing circuit input data.	required
compiled_model_path	str	Path to the compiled ezkl circuit (.compiled).	required
witness_path	str	Output path for the generated witness JSON file.	required

Returns:

Type	Description
	The result returned by ezkl.gen_witness().

Source code in src/zkSNARK.py

async def async_compile( data_path, compiled_model_path, witness_path):
    """
    Async wrapper around ezkl.gen_witness() for use in an asyncio event loop.

    Args:
        data_path (str): Path to the JSON file containing circuit input data.
        compiled_model_path (str): Path to the compiled ezkl circuit (.compiled).
        witness_path (str): Output path for the generated witness JSON file.

    Returns:
        The result returned by ezkl.gen_witness().
    """
    res = await ezkl.gen_witness(data_path, compiled_model_path, witness_path)
    return res

---

async_srs async

async_srs(settings_path)

Async wrapper around ezkl.get_srs() for downloading the structured reference string.

Parameters:

Name	Type	Description	Default
settings_path	str	Path to the ezkl settings JSON file that specifies the required SRS size.	required

Returns:

Type	Description
	The result returned by ezkl.get_srs().

Source code in src/zkSNARK.py

async def async_srs(settings_path):
    """
    Async wrapper around ezkl.get_srs() for downloading the structured reference string.

    Args:
        settings_path (str): Path to the ezkl settings JSON file that specifies
            the required SRS size.

    Returns:
        The result returned by ezkl.get_srs().
    """
    res = await ezkl.get_srs(settings_path)
    return res