xlite_toolkit/func_defs.py at main · tryiou/xlite_toolkit · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
import json
import requests
import time
import platform
import os
import subprocess
import sys
import psutil


def is_xlite_daemon_running():
    binary_name = 'xlite-daemon.exe' if platform.system() == 'Windows' else \
        'xlite-daemon-linux64' if platform.system() == 'Linux' else 'xlite-daemon-osx64'
    for proc in psutil.process_iter(['name']):
        if proc.info['name'] == binary_name:
            return True
    return False


def rpc_call(method, params, url, rpc_user, rpc_password, rpc_port, display=False):
    max_retries = 4
    retry_delay = 5
    timeout = 5  # Default timeout in seconds

    url = url + ':' + str(rpc_port) + '/'
    payload = json.dumps({'method': method, 'params': params})
    headers = {"Content-Type": "application/json"}
    auth = (rpc_user, rpc_password)

    if display:
        print('rpc_call(', method, ',', params, ')', url)

    for retry in range(max_retries):
        try:
            response = requests.request('POST', url, data=payload, headers=headers, auth=auth, timeout=timeout)
            response.raise_for_status()  # Raise an exception for non-2xx status codes
            response_json = response.json()

            return response_json

        except requests.exceptions.HTTPError as e:
            print("HTTP Error:", e)

        except requests.exceptions.ConnectionError as e:
            print("Connection Error:", e)

        except Exception as e:
            print("Error:", e)

        if retry < max_retries - 1:
            print('Retrying in {} seconds...'.format(retry_delay))
            time.sleep(retry_delay)

    print('Max retries reached. Unable to complete the request.')
    return None


def get_settings_folder():
    system = platform.system()
    if system == 'Windows':
        return os.path.expandvars('%APPDATA%\\CloudChains\\settings')
    elif system == 'Darwin':  # MacOS
        return os.path.expanduser('~/Library/Application Support/CloudChains/settings')
    elif system == 'Linux':
        return os.path.expanduser('~/.config/CloudChains/settings')
    else:
        raise Exception('Unsupported operating system: ' + system)


def run_bin(xlite_pass):
    print('Starting xlite-daemon')
    script_dir = os.path.dirname(os.path.abspath(__file__))
    system = platform.system()

    if system == 'Windows':
        binary_name = 'xlite-daemon-win64.exe'
    elif system == 'Linux':
        binary_name = 'xlite-daemon-linux64'
    else:
        binary_name = 'xlite-daemon-osx64'

    binary_path = os.path.join(script_dir, binary_name)

    if not os.path.exists(binary_path):
        print(f"Error: '{binary_name}' binary file not found.")
        sys.exit(1)

    arguments = []
    os.environ['WALLET_PASSWORD'] = xlite_pass['xlite_pass']
    # Execute the binary and detach it
    process = execute_binary(binary_path, arguments)
    time.sleep(1)
    return process


def execute_binary(binary_path, arguments):
    try:
        # Start the process and detach it
        process = subprocess.Popen([binary_path] + arguments, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                   stdin=subprocess.DEVNULL, start_new_session=True)
        return process
    except Exception as e:
        # Handle any exceptions that occur during the execution
        print('Error:', e)
        return None


def kill_bin(process):
    print('Closing xlite-daemon')
    process.terminate()
    #    process.send_signal(subprocess.signal.SIGINT)
    process.wait()