pyasic

Miner Network

A class to handle a network containing miners. Handles scanning and gets miners via MinerFactory.

Parameters:

Name Type Description Default
ip_addr Union[str, List[str], None]
An IP address, range of IP addresses, or a list of IPs
  • Takes a single IP address as an ipadddress.ipaddress() or a string
  • Takes a string formatted as: f"{ip_range_1_start}-{ip_range_1_end}, {ip_address_1}, {ip_range_2_start}-{ip_range_2_end}, {ip_address_2}..."
  • Also takes a list of strings or ipaddress.ipaddress formatted as: [{ip_address_1}, {ip_address_2}, {ip_address_3}, ...]
None
mask Union[str, int, None]

A subnet mask to use when constructing the network. Only used if ip_addr is a single IP. Defaults to /24 (255.255.255.0 or 0.0.0.255)

None
Source code in pyasic/network/__init__.py
 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
class MinerNetwork:
    """A class to handle a network containing miners. Handles scanning and gets miners via [`MinerFactory`][pyasic.miners.miner_factory.MinerFactory].

    Parameters:
        ip_addr: ### An IP address, range of IP addresses, or a list of IPs
            * Takes a single IP address as an `ipadddress.ipaddress()` or a string
            * Takes a string formatted as:
            ```f"{ip_range_1_start}-{ip_range_1_end}, {ip_address_1}, {ip_range_2_start}-{ip_range_2_end}, {ip_address_2}..."```
            * Also takes a list of strings or `ipaddress.ipaddress` formatted as:
            ```[{ip_address_1}, {ip_address_2}, {ip_address_3}, ...]```
        mask: A subnet mask to use when constructing the network.  Only used if `ip_addr` is a single IP.
            Defaults to /24 (255.255.255.0 or 0.0.0.255)
    """

    def __init__(
        self,
        ip_addr: Union[str, List[str], None] = None,
        mask: Union[str, int, None] = None,
    ) -> None:
        self.network = None
        self.ip_addr = ip_addr
        self.connected_miners = {}
        if isinstance(mask, str):
            if mask.startswith("/"):
                mask = mask.replace("/", "")
        self.mask = mask
        self.network = self.get_network()

    def __len__(self):
        return len([item for item in self.get_network().hosts()])

    def __repr__(self):
        return str(self.network)

    def hosts(self):
        for x in self.network.hosts():
            yield x

    def get_network(self) -> ipaddress.ip_network:
        """Get the network using the information passed to the MinerNetwork or from cache.

        Returns:
            The proper network to be able to scan.
        """
        # if we have a network cached already, use that
        if self.network:
            return self.network

            # if there is no IP address passed, default to 192.168.1.0
        if not self.ip_addr:
            self.ip_addr = "192.168.1.0"
        if "-" in self.ip_addr:
            self.network = MinerNetworkRange(self.ip_addr)
        elif isinstance(self.ip_addr, list):
            self.network = MinerNetworkRange(self.ip_addr)
        else:
            # if there is no subnet mask passed, default to /24
            if not self.mask:
                subnet_mask = "24"
            # if we do have a mask passed, use that
            else:
                subnet_mask = str(self.mask)

            # save the network and return it
            self.network = ipaddress.ip_network(
                f"{self.ip_addr}/{subnet_mask}", strict=False
            )

        logging.debug(f"{self} - (Get Network) - Found network")
        return self.network

    async def scan_network_for_miners(self) -> List[AnyMiner]:
        """Scan the network for miners, and return found miners as a list.

        Returns:
            A list of found miners.
        """
        # get the network
        local_network = self.get_network()
        logging.debug(f"{self} - (Scan Network For Miners) - Scanning")

        # clear cached miners
        miner_factory.clear_cached_miners()

        limit = asyncio.Semaphore(PyasicSettings().network_scan_threads)
        miners = await asyncio.gather(
            *[self.ping_and_get_miner(host, limit) for host in local_network.hosts()]
        )

        # remove all None from the miner list
        miners = list(filter(None, miners))
        logging.debug(
            f"{self} - (Scan Network For Miners) - Found {len(miners)} miners"
        )

        # return the miner objects
        return miners

    async def scan_network_generator(self) -> AsyncIterator[AnyMiner]:
        """
        Scan the network for miners using an async generator.

        Returns:
             An asynchronous generator containing found miners.
        """
        # get the current event loop
        loop = asyncio.get_event_loop()

        # get the network
        local_network = self.get_network()

        # create a list of scan tasks
        limit = asyncio.Semaphore(PyasicSettings().network_scan_threads)
        miners = asyncio.as_completed(
            [
                loop.create_task(self.ping_and_get_miner(host, limit))
                for host in local_network.hosts()
            ]
        )
        for miner in miners:
            try:
                yield await miner
            except TimeoutError:
                yield None

    @staticmethod
    async def ping_miner(
        ip: ipaddress.ip_address, semaphore: asyncio.Semaphore
    ) -> Union[None, ipaddress.ip_address]:
        async with semaphore:
            try:
                miner = await ping_miner(ip)
                if miner:
                    return miner
            except ConnectionRefusedError:
                tasks = [ping_miner(ip, port=port) for port in [4029, 8889]]
                for miner in asyncio.as_completed(tasks):
                    try:
                        miner = await miner
                        if miner:
                            return miner
                    except ConnectionRefusedError:
                        pass

    @staticmethod
    async def ping_and_get_miner(
        ip: ipaddress.ip_address, semaphore: asyncio.Semaphore
    ) -> Union[None, AnyMiner]:
        async with semaphore:
            try:
                miner = await ping_and_get_miner(ip)
                if miner:
                    return miner
            except ConnectionRefusedError:
                tasks = [ping_and_get_miner(ip, port=port) for port in [4029, 8889]]
                for miner in asyncio.as_completed(tasks):
                    try:
                        miner = await miner
                        if miner:
                            return miner
                    except ConnectionRefusedError:
                        pass

get_network()

Get the network using the information passed to the MinerNetwork or from cache.

Returns:

Type Description
ip_network

The proper network to be able to scan.

Source code in pyasic/network/__init__.py
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
def get_network(self) -> ipaddress.ip_network:
    """Get the network using the information passed to the MinerNetwork or from cache.

    Returns:
        The proper network to be able to scan.
    """
    # if we have a network cached already, use that
    if self.network:
        return self.network

        # if there is no IP address passed, default to 192.168.1.0
    if not self.ip_addr:
        self.ip_addr = "192.168.1.0"
    if "-" in self.ip_addr:
        self.network = MinerNetworkRange(self.ip_addr)
    elif isinstance(self.ip_addr, list):
        self.network = MinerNetworkRange(self.ip_addr)
    else:
        # if there is no subnet mask passed, default to /24
        if not self.mask:
            subnet_mask = "24"
        # if we do have a mask passed, use that
        else:
            subnet_mask = str(self.mask)

        # save the network and return it
        self.network = ipaddress.ip_network(
            f"{self.ip_addr}/{subnet_mask}", strict=False
        )

    logging.debug(f"{self} - (Get Network) - Found network")
    return self.network

scan_network_for_miners() async

Scan the network for miners, and return found miners as a list.

Returns:

Type Description
List[AnyMiner]

A list of found miners.

Source code in pyasic/network/__init__.py
 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
async def scan_network_for_miners(self) -> List[AnyMiner]:
    """Scan the network for miners, and return found miners as a list.

    Returns:
        A list of found miners.
    """
    # get the network
    local_network = self.get_network()
    logging.debug(f"{self} - (Scan Network For Miners) - Scanning")

    # clear cached miners
    miner_factory.clear_cached_miners()

    limit = asyncio.Semaphore(PyasicSettings().network_scan_threads)
    miners = await asyncio.gather(
        *[self.ping_and_get_miner(host, limit) for host in local_network.hosts()]
    )

    # remove all None from the miner list
    miners = list(filter(None, miners))
    logging.debug(
        f"{self} - (Scan Network For Miners) - Found {len(miners)} miners"
    )

    # return the miner objects
    return miners

scan_network_generator() async

Scan the network for miners using an async generator.

Returns:

Type Description
AsyncIterator[AnyMiner]

An asynchronous generator containing found miners.

Source code in pyasic/network/__init__.py
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
async def scan_network_generator(self) -> AsyncIterator[AnyMiner]:
    """
    Scan the network for miners using an async generator.

    Returns:
         An asynchronous generator containing found miners.
    """
    # get the current event loop
    loop = asyncio.get_event_loop()

    # get the network
    local_network = self.get_network()

    # create a list of scan tasks
    limit = asyncio.Semaphore(PyasicSettings().network_scan_threads)
    miners = asyncio.as_completed(
        [
            loop.create_task(self.ping_and_get_miner(host, limit))
            for host in local_network.hosts()
        ]
    )
    for miner in miners:
        try:
            yield await miner
        except TimeoutError:
            yield None