scan_network.py 文件源码

def ping_scan(to_scan: dict, threads=None, threads_per_cpu=4):
    """Scans the specified networks using `ping`.

    The `to_scan` dictionary must be in the format:

        {<interface_name>: <iterable-of-cidr-strings>, ...}

    If the `threads` argument is supplied, the specified number of threads
    will be used for concurrent scanning. If threads=1 is specified, scanning
    will use a single process (and be very slow).
    """
    jobs = yield_ping_parameters(to_scan)
    if threads is None:
        threads = cpu_count() * threads_per_cpu
    if threads == 1:
        yield from (run_ping(job) for job in jobs)
    else:
        with ThreadPool(processes=threads) as pool:
            yield from pool.imap(run_ping, jobs)