Ping multiple hosts to check for liveness

multi_ping.py

import asyncio, concurrent, os
            
def ping(host, count=2, timeout=2):
    response = os.system(f'ping -c {count} -W {timeout} {host} > /dev/null')
    status = 'success' if response == 0 else 'fail'
    return {host: status}

async def main(hosts, loop, **kw):
    args = [2, 2]
    if 'timeout' in kw:
        args[1] = kw['timeout']
    if 'count' in kw:
        args[0] = kw['count']

    with concurrent.futures.ThreadPoolExecutor(max_workers=min(32, os.cpu_count() + 4)) as executor:
        futures = [
            loop.run_in_executor(executor, ping, host, *args) for host in hosts
        ]
        results = await asyncio.gather(*futures)
    response = {}
    for d in results:
        for k,v in d.items():
            response[k] = v
    return response

def multi_ping(hosts: list, **kw):
    """
    hosts: ['google.com', 'abc.com', 'netflix.com', 'python.org']
    kw:
        timeout=<int> # ping times out after n seconds
        count=2  # number of pings 
    returns:
        {'host1': 'success', 'host2': 'fail'}
    """
    loop = asyncio.new_event_loop()
    return loop.run_until_complete(main(hosts, loop, **kw))
"""
Usage:
Python 3.7.5 (default, Nov  7 2019, 10:50:52) 
[GCC 8.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from multi_ping import multi_ping
>>> multi_ping(['google.com', 'abc.com', 'netflix.com', 'python.org'], timeout=5)
{'google.com': 'success', 'abc.com': 'success', 'netflix.com': 'fail', 'python.org': 'success'}

"""