Generated on: 2024-07-19 14:07:36
Last active
July 19, 2024 18:10
-
-
Save talalashraf/f11efab6303c8cb7185c62ca0830889b to your computer and use it in GitHub Desktop.
Consensus params on block size for cosmos chains
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import os | |
| import json | |
| import requests | |
| import logging | |
| from datetime import datetime | |
| # Set up logging | |
| logging.basicConfig(level=logging.INFO, | |
| format='%(asctime)s - %(levelname)s - %(message)s', | |
| handlers=[ | |
| logging.FileHandler("script.log"), | |
| logging.StreamHandler() | |
| ]) | |
| def get_consensus_params(rpc_url): | |
| endpoint = f"{rpc_url}/consensus_params" | |
| try: | |
| logging.info(f"Requesting consensus params from {endpoint}") | |
| response = requests.get(endpoint, timeout=10) | |
| if response.status_code == 200: | |
| data = response.json() | |
| result = data.get('result', {}) | |
| consensus_params = result.get('consensus_params', {}) | |
| block = consensus_params.get('block', {}) | |
| max_bytes = block.get('max_bytes') | |
| max_gas = block.get('max_gas') | |
| logging.info(f"Successfully retrieved consensus params: max_bytes={max_bytes}, max_gas={max_gas}") | |
| return max_bytes, max_gas | |
| else: | |
| logging.warning(f"Failed to get consensus params. Status code: {response.status_code}") | |
| except requests.RequestException as e: | |
| logging.error(f"Request failed for {endpoint}: {str(e)}") | |
| return None, None | |
| def format_bytes(bytes_value): | |
| try: | |
| mb_value = int(bytes_value) / (1024 * 1024) | |
| return f"{mb_value:.0f}M" | |
| except (ValueError, TypeError): | |
| return bytes_value # Return original value if conversion fails | |
| def main(): | |
| base_dir = '.' # Assuming the script is run from the top-level directory | |
| results = [] | |
| logging.info("Starting to process chain directories") | |
| for chain_dir in os.listdir(base_dir): | |
| chain_path = os.path.join(base_dir, chain_dir) | |
| if os.path.isdir(chain_path): | |
| chain_json_path = os.path.join(chain_path, 'chain.json') | |
| if os.path.exists(chain_json_path): | |
| logging.info(f"Processing {chain_json_path}") | |
| with open(chain_json_path, 'r') as f: | |
| chain_data = json.load(f) | |
| chain_name = chain_data.get('chain_name', 'Unknown') | |
| rpc_endpoints = chain_data.get('apis', {}).get('rpc', []) | |
| for rpc in rpc_endpoints: | |
| rpc_url = rpc.get('address') | |
| if rpc_url: | |
| max_bytes, max_gas = get_consensus_params(rpc_url) | |
| if max_bytes is not None and max_gas is not None: | |
| results.append({ | |
| 'chain_name': chain_name, | |
| 'rpc_url': rpc_url, | |
| 'max_bytes': format_bytes(max_bytes), | |
| 'max_gas': max_gas | |
| }) | |
| logging.info(f"Added results for {chain_name}") | |
| break # Stop after first successful request for this chain | |
| else: | |
| logging.warning(f"No valid RPC URL found for {chain_name}") | |
| # Write results to blockparams.md | |
| logging.info("Writing results to blockparams.md") | |
| with open('blockparams.md', 'w') as f: | |
| f.write("# Block Parameters for Cosmos Chains\n\n") | |
| f.write(f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n") | |
| f.write("| Chain Name | RPC URL | Max Bytes | Max Gas |\n") | |
| f.write("|------------|---------|-----------|--------|\n") | |
| for result in results: | |
| f.write(f"| {result['chain_name']} | {result['rpc_url']} | {result['max_bytes']} | {result['max_gas']} |\n") | |
| logging.info("Script execution completed. Results written to blockparams.md") | |
| if __name__ == "__main__": | |
| logging.info("Script started") | |
| main() | |
| logging.info("Script finished") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment