ros2
diff --git a/‎ros2doctor/package.xml‎
Lines changed: 1 addition & 1 deletion b/‎ros2doctor/package.xml‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎ros2doctor/ros2doctor/api/__init__.py‎
Lines changed: 101 additions & 13 deletions b/‎ros2doctor/ros2doctor/api/__init__.py‎
Lines changed: 101 additions & 13 deletions
diff --git a/‎ros2doctor/ros2doctor/api/format.py‎
Lines changed: 67 additions & 5 deletions b/‎ros2doctor/ros2doctor/api/format.py‎
Lines changed: 67 additions & 5 deletions
diff --git a/‎ros2doctor/ros2doctor/api/network.py‎
Lines changed: 54 additions & 21 deletions b/‎ros2doctor/ros2doctor/api/network.py‎
Lines changed: 54 additions & 21 deletions
@@ -8,7 +8,7 @@
   <license>Apache License 2.0</license>
 
   <depend>ros2cli</depend>
-  
+
   <exec_depend>python3-rosdistro-modules</exec_depend>
 
   <test_depend>ament_copyright</test_depend>
 
@@ -12,22 +12,110 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import List
+from typing import Set
+from typing import Tuple
+
 from pkg_resources import iter_entry_points
+from pkg_resources import UnknownExtra
+
+from ros2doctor.api.format import doctor_warn
+
+
+class DoctorCheck:
+    """Abstract base class of ros2doctor check."""
+
+    def category(self) -> str:
+        """:return: string linking checks and reports."""
+        raise NotImplementedError
+
+    def check(self) -> bool:
+        """:return: boolean indicating result of checks."""
+        raise NotImplementedError
+
+
+class DoctorReport:
+    """Abstract base class of ros2doctor report."""
+
+    def category(self) -> str:
+        """:return: string linking checks and reports."""
+        raise NotImplementedError
 
+    def report(self) -> 'Report':  # using str as wrapper for custom class Report
+        """:return: Report object storing report content."""
+        raise NotImplementedError
 
-def run_checks():
-    """Run all checks when `ros2 doctor/wtf` is called."""
-    all_results = []
-    failed_names = []
-    for check in iter_entry_points('ros2doctor.checks'):
-        result = check.load()()  # load() returns method
-        all_results.append(result)
+
+class Report:
+    """Stores report name and content."""
+
+    __slots__ = ['name', 'items']
+
+    def __init__(self, name: str):
+        """Initialize with report name."""
+        self.name = name
+        self.items = []
+
+    def add_to_report(self, item_name: str, item_info: str) -> None:
+        """Add report content to items list (list of string tuples)."""
+        self.items.append((item_name, item_info))
+
+
+def run_checks() -> Tuple[Set[str], int, int]:
+    """
+    Run all checks and return check results.
+
+    :return: 3-tuple (categories of failed checks, number of failed checks,
+             total number of checks)
+    """
+    failed_cats = set()  # remove repeating elements
+    fail = 0
+    total = 0
+    for check_entry_pt in iter_entry_points('ros2doctor.checks'):
+        try:
+            check_class = check_entry_pt.load()
+        except (ImportError, UnknownExtra):
+            doctor_warn('Check entry point %s fails to load.' % check_entry_pt.name)
+        try:
+            check_instance = check_class()
+        except Exception:
+            doctor_warn('Unable to instantiate check object from %s.' % check_entry_pt.name)
+        try:
+            check_category = check_instance.category()
+            result = check_instance.check()
+        except Exception:
+            doctor_warn('Fail to call %s class functions.' % check_entry_pt.name)
         if result is False:
-            failed_names.append(check.name)
-    return all_results, failed_names
+            fail += 1
+            failed_cats.add(check_category)
+        total += 1
+    return failed_cats, fail, total
+
 
+def generate_reports(*, categories=None) -> List[Report]:
+    """
+    Print all reports or reports of failed checks to terminal.
 
-def generate_report():
-    """Print report to terminal when `-r/--report` is attached."""
-    for report in iter_entry_points('ros2doctor.report'):
-        report.load()()  # load() returns method
+    :return: list of Report objects
+    """
+    reports = []
+    for report_entry_pt in iter_entry_points('ros2doctor.report'):
+        try:
+            report_class = report_entry_pt.load()
+        except (ImportError, UnknownExtra):
+            doctor_warn('Report entry point %s fails to load.' % report_entry_pt.name)
+        try:
+            report_instance = report_class()
+        except Exception:
+            doctor_warn('Unable to instantiate report object from %s.' % report_entry_pt.name)
+        try:
+            report_category = report_instance.category()
+            report = report_instance.report()
+        except Exception:
+            doctor_warn('Fail to call %s class functions.' % report_entry_pt.name)
+        if categories:
+            if report_category in categories:
+                reports.append(report)
+        else:
+            reports.append(report)
+    return reports
@@ -12,9 +12,71 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import sys
+from typing import List
+from typing import Tuple
+import warnings
 
-def print_term(k, v):
-    """Print term only if it exists."""
-    if v:
-        # TODO(clairewang): 20 padding needs to be dynamically set
-        print('{:20}: {}'.format(k, v))
+
+def format_print(report):
+    """
+    Format print report content.
+
+    :param report: Report object with name and items list
+    """
+    # temp fix for missing ifcfg
+    if report is None:
+        sys.stderr.write('No report found. Skip print...\n')
+        return
+
+    print('\n ', report.name)
+    padding_num = compute_padding(report.items)
+    for item_name, item_content in report.items:
+        print('{:{padding}}: {}'.format(item_name, item_content, padding=padding_num))
+
+
+def compute_padding(report_items: List[Tuple[str, str]]) -> int:
+    """
+    Compute padding based on report content.
+
+    :param report_items: list of item name and item content tuple
+    :return: padding number
+    """
+    padding = 8
+    check_items = list(zip(*report_items))[0]  # get first elements of tuples
+    max_len = len(max(check_items, key=len))  # find the longest string length
+    if max_len >= padding:
+        padding = max_len + 4  # padding number is longest string length + 4 spaces
+    return padding
+
+
+def custom_warning_format(msg, cat, filename, linenum, file=None, line=None):
+    return '%s: %s: %s: %s\n' % (filename, linenum, cat.__name__, msg)
+
+
+class CustomWarningFormat:
+    """Support custom warning format without modifying default format."""
+
+    def __enter__(self):
+        self._default_format = warnings.formatwarning
+        warnings.formatwarning = custom_warning_format
+
+    def __exit__(self, t, v, trb):
+        """
+        Define exit action for context manager.
+
+        :param t: type
+        :param v: value
+        :param trb: traceback
+        """
+        warnings.formatwarning = self._default_format
+
+
+def doctor_warn(msg: str) -> None:
+    """
+    Use CustomWarningFormat to print customized warning message.
+
+    :param msg: warning message to be printed
+    """
+    with CustomWarningFormat():
+        warnings.warn(msg)
@@ -14,20 +14,35 @@
 
 import os
 import sys
+from typing import Tuple
+import warnings
 
-import ifcfg
+from ros2doctor.api import DoctorCheck
+from ros2doctor.api import DoctorReport
+from ros2doctor.api import Report
+from ros2doctor.api.format import doctor_warn
 
-from ros2doctor.api.format import print_term
+try:
+    import ifcfg
+except ImportError:
+    warnings.warn('Failed to import ifcfg. '
+                  'Use `python -m pip install ifcfg` to install needed package.',
+                  ImportWarning)  # ImportWarning is suppressed by default
 
 
-def _is_unix_like_platform():
+def _is_unix_like_platform() -> bool:
     """Return True if conforms to UNIX/POSIX-style APIs."""
     return os.name == 'posix'
 
 
-def check_network_config_helper():
+def _check_network_config_helper() -> Tuple[bool, bool, bool]:
     """Check if loopback and multicast IP addresses are found."""
     has_loopback, has_non_loopback, has_multicast = False, False, False
+    # temp fix for ifcfg package, maunually pass network check
+    if 'ifcfg' not in sys.modules:
+        doctor_warn('ifcfg not imported. Skip network check...')
+        return True, True, True
+
     for name, iface in ifcfg.interfaces().items():
         flags = iface.get('flags')
         if 'LOOPBACK' in flags:
@@ -39,22 +54,40 @@ def check_network_config_helper():
     return has_loopback, has_non_loopback, has_multicast
 
 
-def check_network_config():
-    """Conduct network checks and output error/warning messages."""
-    has_loopback, has_non_loopback, has_multicast = check_network_config_helper()
-    if not has_loopback:
-        sys.stderr.write('ERROR: No loopback IP address is found.\n')
-    if not has_non_loopback:
-        sys.stderr.write('WARNING: Only loopback IP address is found.\n')
-    if not has_multicast:
-        sys.stderr.write('WARNING: No multicast IP address is found.\n')
-    return has_loopback and has_non_loopback and has_multicast
+class NetworkCheck(DoctorCheck):
+    """Check network interface configuration for loopback and multicast."""
 
+    def category(self):
+        return 'network'
 
-def print_network():
-    """Print all system and ROS network information."""
-    print('NETWORK CONFIGURATION')
-    for name, iface in ifcfg.interfaces().items():
-        for k, v in iface.items():
-            print_term(k, v)
-    print('\n')
+    def check(self):
+        """Check network configuration."""
+        has_loopback, has_non_loopback, has_multicast = _check_network_config_helper()
+        if not has_loopback:
+            doctor_warn('ERROR: No loopback IP address is found.')
+        if not has_non_loopback:
+            doctor_warn('Only loopback IP address is found.')
+        if not has_multicast:
+            doctor_warn('No multicast IP address is found.')
+        return has_loopback and has_non_loopback and has_multicast
+
+
+class NetworkReport(DoctorReport):
+    """Report network configuration."""
+
+    def category(self):
+        return 'network'
+
+    def report(self):
+        """Print system and ROS network information."""
+        network_report = Report('NETWORK CONFIGURATION')
+        # temp fix for ifcfg package, return none for report
+        if 'ifcfg' not in sys.modules:
+            doctor_warn('ERROR: ifcfg package not imported. Skipping network report...')
+            return None
+
+        for name, iface in ifcfg.interfaces().items():
+            for k, v in iface.items():
+                if v:
+                    network_report.add_to_report(k, v)
+        return network_report