Skip to content

XMRigManager

A class to manage multiple XMRig miners via their APIs.

Attributes:

Name Type Description
_miners Dict[str, XMRigAPI]

A dictionary to store miner API instances.
_api_factory Callable[..., XMRigAPI]

Factory for creating XMRigAPI instances.
_db_url str

Database URL for storing miner data.
Source code in xmrig/manager.py 
 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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class XMRigManager:
    """
    A class to manage multiple XMRig miners via their APIs.

    Attributes:
        _miners (Dict[str, XMRigAPI]): A dictionary to store miner API instances.
        _api_factory (Callable[..., XMRigAPI]): Factory for creating XMRigAPI instances.
        _db_url (str): Database URL for storing miner data.
    """

    def __init__(self, api_factory: Callable[..., XMRigAPI] = XMRigAPI, db_url: str = "sqlite:///xmrig-api.db"):
        """
        Initializes the manager with an empty collection of miners.

        Args:
            api_factory (Callable): Factory for creating XMRigAPI instances.
            db_url (str): Database URL for storing miner data.
        """
        self._miners = {}
        self._api_factory = api_factory
        self._db_url = db_url
        if self._db_url is not None:
            XMRigDatabase.init_db(self._db_url)

    def add_miner(self, miner_name: str, ip: str, port: int, access_token: Optional[str] = None, tls_enabled: bool = False) -> None:
        """
        Adds a new miner to the manager.

        Args:
            miner_name (str): A unique name for the miner.
            ip (str): IP address or domain of the XMRig API.
            port (int): Port of the XMRig API.
            access_token (Optional[str], optional): Access token for authorization. Defaults to None.
            tls_enabled (bool, optional): TLS status of the miner/API. Defaults to False.

        Raises:
            XMRigManagerError: If an error occurs while adding the miner.
        """
        try:
            if miner_name in self._miners:
                raise ValueError(f"Miner with name '{miner_name}' already exists.")
            # Use the injected factory to create the API instance
            self._miners[miner_name] = self._api_factory(miner_name, ip, port, access_token, tls_enabled, self._db_url)
            log.info(f"Miner called '{miner_name}' added to manager.")
        except Exception as e:
            raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred adding miner '{miner_name}':") from e

    def remove_miner(self, miner_name: str) -> None:
        """
        Removes a miner from the manager.

        Args:
            miner_name (str): The unique name of the miner to remove.

        Raises:
            XMRigManagerError: If an error occurs while removing the miner.
        """
        try:
            if miner_name not in self._miners:
                raise ValueError(f"Miner with name '{miner_name}' does not exist.")
            if self._db_url is not None:
                XMRigDatabase.delete_all_miner_data_from_db(miner_name, self._db_url)
            del self._miners[miner_name]
            log.info(f"Miner '{miner_name}' removed from manager.")
        except Exception as e:
            raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred removing miner '{miner_name}':") from e

    def get_miner(self, miner_name: str) -> XMRigAPI:
        """
        Retrieves a specific miner's API instance.

        Args:
            miner_name (str): The unique name of the miner.

        Returns:
            XMRigAPI: The API instance for the requested miner.

        Raises:
            XMRigManagerError: If an error occurs while retrieving the miner.
        """
        try:
            if miner_name not in self._miners:
                raise ValueError(f"Miner with name '{miner_name}' does not exist.")
            return self._miners[miner_name]
        except Exception as e:
            raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred retrieving miner '{miner_name}':") from e

    def edit_miner(self, miner_name: str, new_details: dict) -> None:
        """
        Edits the details of a miner. The following details can be edited:

        - miner_name (str): A unique name for the miner.
        - ip (str): IP address or domain of the XMRig API.
        - port (str): Port of the XMRig API.
        - access_token (Optional[str]): Access token for authorization.
        - tls_enabled (bool): TLS status of the miner/API.

        The dictionary can be in any order and can contain any number of the above keys. For example:

        full_details = {
            'miner_name': 'new_name',
            'ip': 'new_ip_or_domain_with_tld',
            'port': '1234',
            'access_token': 'new-token',
            'tls_enabled': True
        }

        partial_details = {
            'miner_name': 'new_name',
            'port': '1234'
        }

        Args:
            miner_name (str): The unique name of the miner.
            new_details (dict): The new details for the miner.

        Raises:
            XMRigManagerError: If an error occurs while editing the miner.
        """
        try:
            new_name = ""
            miner_api = self.get_miner(miner_name)
            for key, value in new_details.items():
                if key == "miner_name":
                    if value in self._miners:
                        raise ValueError(f"Miner with name '{value}' already exists.")
                    new_name = value
                    miner_api._miner_name = new_name
                    # Remove old entry and replace with new entry
                    del self._miners[miner_name]
                    self._miners[value] = miner_api
                elif key == "ip":
                    miner_api._ip = value
                elif key == "port":
                    miner_api._port = value
                elif key == "access_token":
                    miner_api.set_auth_header(value)
                elif key == "tls_enabled":
                    miner_api._tls_enabled = value
            # Get the miner API instance with the new name to edit further and then return
            miner_api = self.get_miner(new_name)
            # Check if keys "ip", "port" or "tls_enabled" are in the new_details dictionary to construct the new base URL
            if "ip" in new_details or "port" in new_details or "tls_enabled" in new_details:
                miner_api._base_url = f"http://{miner_api._ip}:{miner_api._port}"
                if miner_api._tls_enabled:
                    self._base_url = f"https://{miner_api._ip}:{miner_api._port}"
            log.info(f"Miner called '{miner_name}' successfully edited." if new_name == "" else f"Miner called '{miner_name}' successfully edited to '{new_name}'.")
            return miner_api
        except Exception as e:
            raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred editing miner '{miner_name}':") from e

    def perform_action_on_all(self, action: str) -> None:
        """
        Performs the specified action on all miners.

        Args:
            action (str): The action to perform ('pause', 'resume', 'stop', 'start').

        Raises:
            XMRigManagerError: If an error occurs while performing the action on all miners.
        """
        try:
            for miner_name, miner_api in self._miners.items():
                success = miner_api.perform_action(action)
                if success:
                    log.info(f"Action '{action}' successfully performed on '{miner_name}'.")
                else:
                    log.warning(f"Action '{action}' failed on '{miner_name}'.")
        except Exception as e:
            raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred performing action '{action}' on all miners:") from e

    def update_miners(self, endpoint: Optional[str] = None) -> bool:
        """
        Updates all miners' cached data or calls a specific endpoint on all miners.

        Args:
            endpoint (Optional[str], optional): The endpoint to call on each miner. If None, updates all cached data. Defaults to None.

        Returns:
            bool: True if successful, or False if an error occurred.

        Raises:
            XMRigManagerError: If an error occurs while updating the miners or calling the endpoint.
        """
        try:
            for miner_name, miner_api in self._miners.items():
                if endpoint:
                    response = miner_api.call_endpoint(endpoint)
                    if response:
                        log.info(f"{endpoint.capitalize()} endpoint successfully called on '{miner_name}'.")
                    else:
                        log.warning(f"Failed to call '{endpoint}' endpoint on '{miner_name}'.")
                else:
                    success = miner_api.get_all_responses()
                    if success:
                        log.info(f"Miner called '{miner_name}' successfully updated.")
                    else:
                        log.warning(f"Failed to update miner '{miner_name}'.")
            return True
        except Exception as e:
            raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred updating miners or calling endpoint '{endpoint}' on all miners:") from e

    def list_miners(self) -> List[str]:
        """
        Lists all managed miners.

        Returns:
            List[str]: A list of miner names.
        """
        return list(self._miners.keys())

__init__(api_factory=XMRigAPI, db_url='sqlite:///xmrig-api.db')

Initializes the manager with an empty collection of miners.

Parameters:

Name Type Description Default
api_factory Callable

Factory for creating XMRigAPI instances.

 XMRigAPI
db_url str

Database URL for storing miner data.

 'sqlite:///xmrig-api.db'
Source code in xmrig/manager.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, api_factory: Callable[..., XMRigAPI] = XMRigAPI, db_url: str = "sqlite:///xmrig-api.db"):
    """
    Initializes the manager with an empty collection of miners.

    Args:
        api_factory (Callable): Factory for creating XMRigAPI instances.
        db_url (str): Database URL for storing miner data.
    """
    self._miners = {}
    self._api_factory = api_factory
    self._db_url = db_url
    if self._db_url is not None:
        XMRigDatabase.init_db(self._db_url)

add_miner(miner_name, ip, port, access_token=None, tls_enabled=False)

Adds a new miner to the manager.

Parameters:

Name Type Description Default
miner_name str

A unique name for the miner.

 required
ip str

IP address or domain of the XMRig API.

 required
port int

Port of the XMRig API.

 required
access_token Optional[str]

Access token for authorization. Defaults to None.

 None
tls_enabled bool

TLS status of the miner/API. Defaults to False.

 False

Raises:

Type Description
XMRigManagerError

If an error occurs while adding the miner.
Source code in xmrig/manager.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def add_miner(self, miner_name: str, ip: str, port: int, access_token: Optional[str] = None, tls_enabled: bool = False) -> None:
    """
    Adds a new miner to the manager.

    Args:
        miner_name (str): A unique name for the miner.
        ip (str): IP address or domain of the XMRig API.
        port (int): Port of the XMRig API.
        access_token (Optional[str], optional): Access token for authorization. Defaults to None.
        tls_enabled (bool, optional): TLS status of the miner/API. Defaults to False.

    Raises:
        XMRigManagerError: If an error occurs while adding the miner.
    """
    try:
        if miner_name in self._miners:
            raise ValueError(f"Miner with name '{miner_name}' already exists.")
        # Use the injected factory to create the API instance
        self._miners[miner_name] = self._api_factory(miner_name, ip, port, access_token, tls_enabled, self._db_url)
        log.info(f"Miner called '{miner_name}' added to manager.")
    except Exception as e:
        raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred adding miner '{miner_name}':") from e

edit_miner(miner_name, new_details)

Edits the details of a miner. The following details can be edited:

  • miner_name (str): A unique name for the miner.
  • ip (str): IP address or domain of the XMRig API.
  • port (str): Port of the XMRig API.
  • access_token (Optional[str]): Access token for authorization.
  • tls_enabled (bool): TLS status of the miner/API.

The dictionary can be in any order and can contain any number of the above keys. For example:

full_details = { 'miner_name': 'new_name', 'ip': 'new_ip_or_domain_with_tld', 'port': '1234', 'access_token': 'new-token', 'tls_enabled': True }

partial_details = { 'miner_name': 'new_name', 'port': '1234' }

Parameters:

Name Type Description Default
miner_name str

The unique name of the miner.

 required
new_details dict

The new details for the miner.

 required

Raises:

Type Description
XMRigManagerError

If an error occurs while editing the miner.
Source code in xmrig/manager.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def edit_miner(self, miner_name: str, new_details: dict) -> None:
    """
    Edits the details of a miner. The following details can be edited:

    - miner_name (str): A unique name for the miner.
    - ip (str): IP address or domain of the XMRig API.
    - port (str): Port of the XMRig API.
    - access_token (Optional[str]): Access token for authorization.
    - tls_enabled (bool): TLS status of the miner/API.

    The dictionary can be in any order and can contain any number of the above keys. For example:

    full_details = {
        'miner_name': 'new_name',
        'ip': 'new_ip_or_domain_with_tld',
        'port': '1234',
        'access_token': 'new-token',
        'tls_enabled': True
    }

    partial_details = {
        'miner_name': 'new_name',
        'port': '1234'
    }

    Args:
        miner_name (str): The unique name of the miner.
        new_details (dict): The new details for the miner.

    Raises:
        XMRigManagerError: If an error occurs while editing the miner.
    """
    try:
        new_name = ""
        miner_api = self.get_miner(miner_name)
        for key, value in new_details.items():
            if key == "miner_name":
                if value in self._miners:
                    raise ValueError(f"Miner with name '{value}' already exists.")
                new_name = value
                miner_api._miner_name = new_name
                # Remove old entry and replace with new entry
                del self._miners[miner_name]
                self._miners[value] = miner_api
            elif key == "ip":
                miner_api._ip = value
            elif key == "port":
                miner_api._port = value
            elif key == "access_token":
                miner_api.set_auth_header(value)
            elif key == "tls_enabled":
                miner_api._tls_enabled = value
        # Get the miner API instance with the new name to edit further and then return
        miner_api = self.get_miner(new_name)
        # Check if keys "ip", "port" or "tls_enabled" are in the new_details dictionary to construct the new base URL
        if "ip" in new_details or "port" in new_details or "tls_enabled" in new_details:
            miner_api._base_url = f"http://{miner_api._ip}:{miner_api._port}"
            if miner_api._tls_enabled:
                self._base_url = f"https://{miner_api._ip}:{miner_api._port}"
        log.info(f"Miner called '{miner_name}' successfully edited." if new_name == "" else f"Miner called '{miner_name}' successfully edited to '{new_name}'.")
        return miner_api
    except Exception as e:
        raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred editing miner '{miner_name}':") from e

get_miner(miner_name)

Retrieves a specific miner's API instance.

Parameters:

Name Type Description Default
miner_name str

The unique name of the miner.

 required

Returns:

Name Type Description
XMRigAPI XMRigAPI

The API instance for the requested miner.

Raises:

Type Description
XMRigManagerError

If an error occurs while retrieving the miner.
Source code in xmrig/manager.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def get_miner(self, miner_name: str) -> XMRigAPI:
    """
    Retrieves a specific miner's API instance.

    Args:
        miner_name (str): The unique name of the miner.

    Returns:
        XMRigAPI: The API instance for the requested miner.

    Raises:
        XMRigManagerError: If an error occurs while retrieving the miner.
    """
    try:
        if miner_name not in self._miners:
            raise ValueError(f"Miner with name '{miner_name}' does not exist.")
        return self._miners[miner_name]
    except Exception as e:
        raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred retrieving miner '{miner_name}':") from e

list_miners()

Lists all managed miners.

Returns:

Type Description
List[str]

List[str]: A list of miner names.
Source code in xmrig/manager.py 
224
225
226
227
228
229
230
231
def list_miners(self) -> List[str]:
    """
    Lists all managed miners.

    Returns:
        List[str]: A list of miner names.
    """
    return list(self._miners.keys())

perform_action_on_all(action)

Performs the specified action on all miners.

Parameters:

Name Type Description Default
action str

The action to perform ('pause', 'resume', 'stop', 'start').

 required

Raises:

Type Description
XMRigManagerError

If an error occurs while performing the action on all miners.
Source code in xmrig/manager.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def perform_action_on_all(self, action: str) -> None:
    """
    Performs the specified action on all miners.

    Args:
        action (str): The action to perform ('pause', 'resume', 'stop', 'start').

    Raises:
        XMRigManagerError: If an error occurs while performing the action on all miners.
    """
    try:
        for miner_name, miner_api in self._miners.items():
            success = miner_api.perform_action(action)
            if success:
                log.info(f"Action '{action}' successfully performed on '{miner_name}'.")
            else:
                log.warning(f"Action '{action}' failed on '{miner_name}'.")
    except Exception as e:
        raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred performing action '{action}' on all miners:") from e

remove_miner(miner_name)

Removes a miner from the manager.

Parameters:

Name Type Description Default
miner_name str

The unique name of the miner to remove.

 required

Raises:

Type Description
XMRigManagerError

If an error occurs while removing the miner.
Source code in xmrig/manager.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def remove_miner(self, miner_name: str) -> None:
    """
    Removes a miner from the manager.

    Args:
        miner_name (str): The unique name of the miner to remove.

    Raises:
        XMRigManagerError: If an error occurs while removing the miner.
    """
    try:
        if miner_name not in self._miners:
            raise ValueError(f"Miner with name '{miner_name}' does not exist.")
        if self._db_url is not None:
            XMRigDatabase.delete_all_miner_data_from_db(miner_name, self._db_url)
        del self._miners[miner_name]
        log.info(f"Miner '{miner_name}' removed from manager.")
    except Exception as e:
        raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred removing miner '{miner_name}':") from e

update_miners(endpoint=None)

Updates all miners' cached data or calls a specific endpoint on all miners.

Parameters:

Name Type Description Default
endpoint Optional[str]

The endpoint to call on each miner. If None, updates all cached data. Defaults to None.

 None

Returns:

Name Type Description
bool bool

True if successful, or False if an error occurred.

Raises:

Type Description
XMRigManagerError

If an error occurs while updating the miners or calling the endpoint.
Source code in xmrig/manager.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def update_miners(self, endpoint: Optional[str] = None) -> bool:
    """
    Updates all miners' cached data or calls a specific endpoint on all miners.

    Args:
        endpoint (Optional[str], optional): The endpoint to call on each miner. If None, updates all cached data. Defaults to None.

    Returns:
        bool: True if successful, or False if an error occurred.

    Raises:
        XMRigManagerError: If an error occurs while updating the miners or calling the endpoint.
    """
    try:
        for miner_name, miner_api in self._miners.items():
            if endpoint:
                response = miner_api.call_endpoint(endpoint)
                if response:
                    log.info(f"{endpoint.capitalize()} endpoint successfully called on '{miner_name}'.")
                else:
                    log.warning(f"Failed to call '{endpoint}' endpoint on '{miner_name}'.")
            else:
                success = miner_api.get_all_responses()
                if success:
                    log.info(f"Miner called '{miner_name}' successfully updated.")
                else:
                    log.warning(f"Failed to update miner '{miner_name}'.")
        return True
    except Exception as e:
        raise XMRigManagerError(e, traceback.print_exc(), f"An error occurred updating miners or calling endpoint '{endpoint}' on all miners:") from e