outage_detector/main.py

@@ -6,91 +6,61 @@ import requests
 
 
 class LogEntry:
-    '''Class describing a connection attempt.'''
+    """Class describing a connection attempt"""
 
     def __init__(self: 'LogEntry', log_entry: str) -> None:
         self.date = datetime.strptime(log_entry[1:20], '%Y-%m-%d %H:%M:%S')
         self.host = log_entry.split('[')[2].split(']')[0]
         self.failed = 'FAIL' in log_entry
 
-    def __repr__(self: 'LogEntry') -> str:
-        return f"[{self.date}][{self.host}][{'OK' if not self.failed else 'FAIL'}]"
-
 
 def printOutages(filepath: str, time: int) -> None:
-    '''
-        Print outages from a log file
+    """Print outages from a log file"""
 
-        `filepath` Path to a file with connection logs.
-
-        `time` Minimum duration for a connection loss to be considered an outage.
-    '''
-
-    # Get a sorted list of log entries.
     with open(filepath, 'r') as file:
-        log = sorted(
-            map(lambda x: LogEntry(x), file.readlines()),
-            key=lambda x: x.date
-        )
+        log = list(map(lambda x: LogEntry(x), file.readlines()))
 
-    # Cluster log entries by host.
+    # cluster log entries by host
     log_by_host = {}
-
     for entry in log:
         if entry.host not in log_by_host:
             log_by_host[entry.host] = [entry]
         else:
             log_by_host[entry.host].append(entry)
 
-    # Cluster fails to outages.
-    # We consider the time of subsequent failed connection attempts to the same host an outage.
+    # cluster fails to outages
+    # we consider the time of subsequent failed connection attempts to the same host an outage
     outages: list = []
     last_failed = False
     for entries in log_by_host.values():
         for entry in entries:
-
-            # Continue an outage.
             if entry.failed and last_failed:
                 outages[-1].append(entry)
-
-            # Start a new outage.
-            elif entry.failed and not last_failed:
+            if entry.failed and not last_failed:
                 outages.append([entry])
                 last_failed = True
-
-            # End an outage.
-            elif not entry.failed and last_failed:
+            if not entry.failed and last_failed:
                 last_failed = False
 
-    # Print the outages.
-    for outage in outages:
+    # print the outages by date of first fail
+    for outage in sorted(outages, key=lambda x: x[0].date):
 
-        # Get duration of outage in hours and minutes.
+        # get duration of outage in hours and minutes
         outage_duration = round(
-            (outage[-1].date - outage[0].date).seconds / 60
-        )
-
-        # Skip printing if outage is shorter than the minimum duration.
+            (outage[-1].date - outage[0].date).seconds / 60)
+        # continue if outage is shorter than the minimum duration
         if outage_duration < time:
             continue
-
-        # Get hours and minutes for printing.
         hours = outage_duration // 60
         minutes = outage_duration - (hours * 60)
         host = outage[0].host
 
-        # Outputs outages in the form of "Outage at: 2023-19-01 06:29:01 lasting for 2 Hours and 39 Minutes".
+        # Outputs outages in the form of "Outage at: 2023-19-01 06:29:01 lasting for 2 Hours and 39 Minutes"
         print(f"[{outage[0].date}][{host}] lasting for {'{} Hours and '.format(hours) if hours >= 1 else ''}{minutes} Minutes")
 
 
 def isOnline(host: str, timeout: int) -> bool:
-    '''
-        Check if connection to a host can be established.
-
-        `host` IP or hostname of a server to query.
-
-        `timeout` Set the timeout in seconds to use for the connection test.
-    '''
+    """Check if connection to a host can be established"""
 
     try:
         requests.head(f"https://{host}", timeout=timeout)
@@ -100,15 +70,7 @@ def isOnline(host: str, timeout: int) -> bool:
 
 
 def log(host: str, timeout: int, filepath: None | str = None) -> None:
-    '''
-        Log the connection status of a host.
-
-        `host` IP or hostname of a server to query.
-
-        `timeout` Set the timeout in seconds to use for the connection test.
-
-        `filepath` Path to a the file where results should be appended to. Creates the file if it does not exist. If not specified, results are printed to stdout.
-    '''
+    """Log the connection status of a host"""
 
     logline = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}][{host}][{'OK' if isOnline(host, timeout) else 'FAIL'}]"
 
@@ -120,11 +82,6 @@ def log(host: str, timeout: int, filepath: None | str = None) -> None:
 
 
 def parseArgs(args: list[str]) -> argparse.Namespace:
-    '''
-        Parse command line arguments.
-
-        `args` List of user-supplied command line arguments.
-    '''
 
     parser = argparse.ArgumentParser(
         prog='outage_detector', description='Log outages and print statistics.'
@@ -132,7 +89,7 @@ def parseArgs(args: list[str]) -> argparse.Namespace:
 
     subparsers = parser.add_subparsers(dest='command')
 
-    # Arguments for the log command.
+    # Arguments for the log command
     parser_log = subparsers.add_parser(
         'log', help='Log the connection status.'
     )
@@ -149,7 +106,7 @@ def parseArgs(args: list[str]) -> argparse.Namespace:
         help='Set the timeout in seconds to use for the connection test. (default: %(default)s)'
     )
 
-    # Arguments for the outages command.
+    # Arguments for the outages command
     parser_outages = subparsers.add_parser(
         'outages', help='Print/log outages'
     )
@@ -165,7 +122,7 @@ def parseArgs(args: list[str]) -> argparse.Namespace:
     return parser.parse_args(args)
 
 
-def main() -> None:
+def main():
 
     args = parseArgs(sys.argv[1:])