Setting Weights

Setting weights is a crucial part of the Bittensor network’s consensus mechanism. With Fiber, it’s very easy, and works using just one function:

def set_node_weights(
    substrate_interface: SubstrateInterface,
    keypair: Keypair,
    node_ids: list[int],
    node_weights: list[float],
    netuid: int,
    validator_node_id: int,
    version_key: int = 0,
    wait_for_inclusion: bool = False,
    wait_for_finalization: bool = False,
    max_attempts: int = 1,
) -> bool:
    node_ids_formatted, node_weights_formatted = _normalize_and_quantize_weights(node_ids, node_weights)
    rpc_call = substrate_interface.compose_call(
        call_module="SubtensorModule",
        call_function="set_weights",
        call_params={
            "dests": node_ids_formatted,
            "weights": node_weights_formatted,
            "netuid": netuid,
            "version_key": version_key,
        },
    )

    extrinsic_to_send = substrate_interface.create_signed_extrinsic(call=rpc_call, keypair=keypair, era={"period": 5})

    weights_can_be_set = False
    for attempt in range(1, max_attempts + 1):
        if not can_set_weights(substrate_interface, netuid, validator_node_id):
            logger.info(logger.info(f"Skipping attempt {attempt}/{max_attempts}. Too soon to set weights. Will wait 30 secs..."))
            time.sleep(30)
            continue
        else:
            weights_can_be_set = True
            break

    if not weights_can_be_set:
        logger.error("No attempt to set weightsmade. Perhaps it is too soon to set weights!")
        return False

    logger.info("Attempting to set weights...")

    success, error_message = _send_extrinsic(
        substrate_interface=substrate_interface,
        extrinsic_to_send=extrinsic_to_send,
        wait_for_inclusion=wait_for_inclusion,
        wait_for_finalization=wait_for_finalization,
    )

    if not wait_for_finalization and not wait_for_inclusion:
        logger.info("Not waiting for finalization or inclusion to set weights. Returning immediately.")
        return success

    if success:
        if wait_for_finalization:
            logger.info("✅ Successfully set weights and finalized")
        elif wait_for_inclusion:
            logger.info("✅ Successfully set weights and included")
        else:
            logger.info("✅ Successfully set weights")
    else:
        logger.error(f"❌ Failed to set weights: {error_message}")

    return success