Kye Gomez
2 months ago
parent
bbb8f4e458
commit
834b9c5599
@ -1,96 +0,0 @@
|
||||
import base64
|
||||
import requests
|
||||
from struct import unpack
|
||||
|
||||
|
||||
class SolanaTokenInfoFetcher:
|
||||
def __init__(self, rpc_url="https://api.mainnet-beta.solana.com"):
|
||||
"""
|
||||
Initializes the token info fetcher.
|
||||
|
||||
Args:
|
||||
rpc_url (str): The Solana RPC URL.
|
||||
"""
|
||||
self.rpc_url = rpc_url
|
||||
|
||||
def make_rpc_request(self, method, params):
|
||||
"""
|
||||
Makes an RPC request to the Solana JSON RPC API.
|
||||
|
||||
Args:
|
||||
method (str): The RPC method to call.
|
||||
params (list): The parameters for the method.
|
||||
|
||||
Returns:
|
||||
dict: The JSON response from the RPC call.
|
||||
"""
|
||||
headers = {"Content-Type": "application/json"}
|
||||
payload = {
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"method": method,
|
||||
"params": params,
|
||||
}
|
||||
try:
|
||||
response = requests.post(self.rpc_url, json=payload, headers=headers)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Error making RPC request: {e}")
|
||||
return None
|
||||
|
||||
def get_token_info(self, mint_address):
|
||||
"""
|
||||
Fetches token metadata (name, symbol) from the token mint account.
|
||||
|
||||
Args:
|
||||
mint_address (str): The token mint address.
|
||||
|
||||
Returns:
|
||||
dict: A dictionary containing the token name and symbol.
|
||||
"""
|
||||
params = [
|
||||
mint_address,
|
||||
{"encoding": "base64"}
|
||||
]
|
||||
response = self.make_rpc_request("getAccountInfo", params)
|
||||
|
||||
if not response or "result" not in response or not response["result"]["value"]:
|
||||
print(f"Failed to fetch data for mint address: {mint_address}")
|
||||
return None
|
||||
|
||||
# Decode the account data
|
||||
account_data = response["result"]["value"]["data"][0]
|
||||
decoded_data = base64.b64decode(account_data)
|
||||
|
||||
# Extract token name and symbol
|
||||
try:
|
||||
token_name = decoded_data[0:32].decode("utf-8").rstrip("\x00")
|
||||
token_symbol = decoded_data[32:44].decode("utf-8").rstrip("\x00")
|
||||
decimals = unpack("<B", decoded_data[44:45])[0]
|
||||
return {
|
||||
"name": token_name,
|
||||
"symbol": token_symbol,
|
||||
"decimals": decimals
|
||||
}
|
||||
except Exception as e:
|
||||
print(f"Error decoding token metadata: {e}")
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Replace with the token mint address you want to query
|
||||
mint_address = "4zMMC9srt5Ri5X14GAgXhaHii3GnPAEERYPJgZJDncDU" # Example: USDC -> 4zMMC9srt5Ri5X14GAgXhaHii3GnPAEERYPJgZJDncDU
|
||||
|
||||
# Initialize the token info fetcher
|
||||
fetcher = SolanaTokenInfoFetcher()
|
||||
|
||||
# Fetch and display token information
|
||||
token_info = fetcher.get_token_info(mint_address)
|
||||
|
||||
if token_info:
|
||||
print(f"Token Name: {token_info['name']}")
|
||||
print(f"Token Symbol: {token_info['symbol']}")
|
||||
print(f"Decimals: {token_info['decimals']}")
|
||||
else:
|
||||
print("Failed to fetch token information.")
|
||||
@ -1,6 +1,17 @@
|
||||
from litellm import encode
|
||||
|
||||
import subprocess
|
||||
|
||||
def count_tokens(text: str, model: str = "gpt-4o") -> int:
|
||||
"""Count the number of tokens in the given text."""
|
||||
try:
|
||||
from litellm import encode
|
||||
except ImportError:
|
||||
subprocess.run(["pip", "install", "litellm"])
|
||||
from litellm import encode
|
||||
|
||||
|
||||
return len(encode(model=model, text=text))
|
||||
|
||||
|
||||
|
||||
# if __name__ == "__main__":
|
||||
# print(count_tokens("Hello, how are you?"))
|
||||
|
||||
@ -0,0 +1,65 @@
|
||||
from swarms.utils.auto_download_check_packages import auto_check_and_download_package, check_and_install_package
|
||||
|
||||
def test_check_and_install_package_pip():
|
||||
result = check_and_install_package("numpy", package_manager="pip")
|
||||
print(f"Test result for 'numpy' installation using pip: {result}")
|
||||
assert result, "Failed to install or verify 'numpy' using pip"
|
||||
|
||||
def test_check_and_install_package_conda():
|
||||
result = check_and_install_package("numpy", package_manager="conda")
|
||||
print(f"Test result for 'numpy' installation using conda: {result}")
|
||||
assert result, "Failed to install or verify 'numpy' using conda"
|
||||
|
||||
def test_check_and_install_specific_version():
|
||||
result = check_and_install_package("numpy", package_manager="pip", version="1.21.0")
|
||||
print(f"Test result for specific version of 'numpy' installation using pip: {result}")
|
||||
assert result, "Failed to install or verify specific version of 'numpy' using pip"
|
||||
|
||||
def test_check_and_install_with_upgrade():
|
||||
result = check_and_install_package("numpy", package_manager="pip", upgrade=True)
|
||||
print(f"Test result for 'numpy' upgrade using pip: {result}")
|
||||
assert result, "Failed to upgrade 'numpy' using pip"
|
||||
|
||||
def test_auto_check_and_download_single_package():
|
||||
result = auto_check_and_download_package("scipy", package_manager="pip")
|
||||
print(f"Test result for 'scipy' installation using pip: {result}")
|
||||
assert result, "Failed to install or verify 'scipy' using pip"
|
||||
|
||||
def test_auto_check_and_download_multiple_packages():
|
||||
packages = ["scipy", "pandas"]
|
||||
result = auto_check_and_download_package(packages, package_manager="pip")
|
||||
print(f"Test result for multiple packages installation using pip: {result}")
|
||||
assert result, f"Failed to install or verify one or more packages in {packages} using pip"
|
||||
|
||||
def test_auto_check_and_download_multiple_packages_with_versions():
|
||||
packages = ["numpy:1.21.0", "pandas:1.3.0"]
|
||||
result = auto_check_and_download_package(packages, package_manager="pip")
|
||||
print(f"Test result for multiple packages with versions installation using pip: {result}")
|
||||
assert result, f"Failed to install or verify one or more packages in {packages} with specific versions using pip"
|
||||
|
||||
# Example of running tests
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
test_check_and_install_package_pip()
|
||||
print("test_check_and_install_package_pip passed")
|
||||
|
||||
test_check_and_install_package_conda()
|
||||
print("test_check_and_install_package_conda passed")
|
||||
|
||||
test_check_and_install_specific_version()
|
||||
print("test_check_and_install_specific_version passed")
|
||||
|
||||
test_check_and_install_with_upgrade()
|
||||
print("test_check_and_install_with_upgrade passed")
|
||||
|
||||
test_auto_check_and_download_single_package()
|
||||
print("test_auto_check_and_download_single_package passed")
|
||||
|
||||
test_auto_check_and_download_multiple_packages()
|
||||
print("test_auto_check_and_download_multiple_packages passed")
|
||||
|
||||
test_auto_check_and_download_multiple_packages_with_versions()
|
||||
print("test_auto_check_and_download_multiple_packages_with_versions passed")
|
||||
|
||||
except AssertionError as e:
|
||||
print(f"Test failed: {str(e)}")
|
||||
Loading…
Reference in new issue