python_pinger.git

@@ -15,13 +15,13 @@ import signal

 printer_lock = threading.Lock()

-def log_exit(sig, frame):

+def log_exit(sig: int, frame) -> None:

     log(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} '

         f'PID={os.getpid()} signal={signal.Signals(sig).name} Exiting.')

     sys.exit(0)

-def pinger(host, delay):

+def pinger(host: str, delay: float) -> None:

     """Independent worker thread: repeatedly ping, check, print and sleep."""

     last_results = None

     while True:

@@ -34,12 +34,12 @@ def pinger(host, delay):

         time.sleep(delay)

-def ping(host):

+def ping(host: str) -> int:

     """Returns True if host (str) responds to a ping request."""

     return subprocess.call([*_ping_cmd, host], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) == 0

-def log(*args, **kwargs):

+def log(*args, **kwargs) -> None:

     """Opens the logfile, writes the log, and closes the logfile."""

     # I don't leave the log file open for writing because another process needs access too.

     with printer_lock:

@@ -15,14 +15,14 @@ import queue

 import signal

-def log_exit(sig, frame):

+def log_exit(sig: int, frame) -> None:

     _print_queue.put([f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} '

                       f'PID={os.getpid()} signal={signal.Signals(sig).name} Exiting.'])

     _print_queue.join()

     sys.exit(0)

-def pinger(host, delay):

+def pinger(host: str, delay: float) -> None:

     """Independent worker thread: repeatedly ping, check, print and sleep."""

     last_results = None

     while True:

@@ -35,12 +35,12 @@ def pinger(host, delay):

         time.sleep(delay)

-def ping(host):

+def ping(host: str) -> int:

     """Returns True if host (str) responds to a ping request."""

     return subprocess.call([*_ping_cmd, host], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) == 0

-def print_manager():

+def print_manager() -> None:

     """The only thread allowed to write output."""

     while True:

         job = _print_queue.get()

@@ -49,11 +49,11 @@ def print_manager():

         _print_queue.task_done()

-def log(*args, **kwargs):

+def log(*args, **kwargs) -> None:

     """Opens the logfile, writes the log, and closes the logfile."""

     # I don't leave the log file open for writing because another process needs access too.

     with open(_logname, 'a') as fp:

-        return print(*args, **kwargs, file=fp)

+        print(*args, **kwargs, file=fp)

 if __name__ == '__main__':

@@ -13,18 +13,19 @@ from argparse import ArgumentParser

 import threading

 import queue

 import signal

+from typing import Dict, Optional

 pinger_lock = threading.Lock()

-def log_exit(sig, frame):

+def log_exit(sig: int, frame) -> None:

     _print_queue.put([f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} '

                       f'PID={os.getpid()} signal={signal.Signals(sig).name} Exiting.'])

     _print_queue.join()

     sys.exit(0)

-def pinger():

+def pinger() -> None:

     """Each queue item is a new address to ping. Ping and compare and print.

     Issue: Each of these threads reads from and writes to the global _results dict.

     """

@@ -43,12 +44,12 @@ def pinger():

         _ping_queue.task_done()

-def ping(host):

+def ping(host: str) -> int:

     """Returns True if host (str) responds to a ping request."""

     return subprocess.call([*_ping_cmd, host], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) == 0

-def print_manager():

+def print_manager() -> None:

     """The only thread allowed to write output."""

     while True:

         job = _print_queue.get()

@@ -57,11 +58,11 @@ def print_manager():

         _print_queue.task_done()

-def log(*args, **kwargs):

+def log(*args, **kwargs) -> None:

     """Opens the logfile, writes the log, and closes the logfile."""

     # I don't leave the log file open for writing because another process needs access too.

     with open(_logname, 'a') as fp:

-        return print(*args, **kwargs, file=fp)

+        print(*args, **kwargs, file=fp)

 if __name__ == '__main__':

@@ -70,7 +71,7 @@ if __name__ == '__main__':

     parser.add_argument('addresses', nargs='+', help='Addresses to ping')

     parser_args = parser.parse_args()

-    _results = dict()

+    _results: Dict[str, Optional[int]] = dict()

     for addr in parser_args.addresses:

         _results[addr] = None

     delay = parser_args.delay

@@ -13,18 +13,19 @@ from argparse import ArgumentParser

 import threading

 import queue

 import signal

+from typing import Dict, Optional

 pinger_lock = threading.Lock()

-def log_exit(sig, frame):

+def log_exit(sig: int, frame) -> None:

     _print_queue.put([f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} '

                       f'PID={os.getpid()} signal={signal.Signals(sig).name} Exiting.'])

     _print_queue.join()

     sys.exit(0)

-def pinger(host):

+def pinger(host: str) -> None:

     """Executes one ping, prints if there was a change, exits thread."""

     now = time.localtime()

     success = ping(host)

@@ -38,12 +39,12 @@ def pinger(host):

         _print_queue.put([f'{time.strftime("%Y-%m-%d %H:%M:%S", now)} {host} {status}'])

-def ping(host):

+def ping(host: str) -> int:

     """Returns True if host (str) responds to a ping request."""

     return subprocess.call([*_ping_cmd, host], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) == 0

-def print_manager():

+def print_manager() -> None:

     """ The only thread allowed to write output. """

     while True:

         job = _print_queue.get()

@@ -52,11 +53,11 @@ def print_manager():

         _print_queue.task_done()

-def log(*args, **kwargs):

+def log(*args, **kwargs) -> None:

     """Opens the logfile, writes the log, and closes the logfile."""

     # I don't leave the log file open for writing because another process needs access too.

     with open(_logname, 'a') as fp:

-        return print(*args, **kwargs, file=fp)

+        print(*args, **kwargs, file=fp)

 if __name__ == '__main__':

@@ -65,7 +66,7 @@ if __name__ == '__main__':

     parser.add_argument('addresses', nargs='+', help='Addresses to ping')

     parser_args = parser.parse_args()

-    _results = dict()

+    _results: Dict[str, Optional[int]] = dict()

     for addr in parser_args.addresses:

         _results[addr] = None

     delay = parser_args.delay