[ros2doctor] Improve doctor_warn()

ros2doctor/ros2doctor/api/__init__.py

@@ -71,12 +71,10 @@ def __init__(self):
         self.error = 0
         self.warning = 0
 
-    def add_error(self, msg) -> None:
-        doctor_warn(msg)
+    def add_error(self):
         self.error += 1
 
-    def add_warning(self, msg) -> None:
-        doctor_warn(msg)
+    def add_warning(self):
         self.warning += 1
 
 
@@ -87,28 +85,28 @@ def run_checks(*, include_warnings=False) -> Tuple[Set[str], int, int]:
     :return: 3-tuple (categories of failed checks, number of failed checks,
              total number of checks)
     """
-    failed_cats = set()  # remove repeating elements
+    fail_categories = 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)
+            doctor_warn(f'Check entry point {check_entry_pt.name} fails to load.')
         try:
             check_instance = check_class()
         except Exception:
-            doctor_warn('Unable to instantiate check object from %s.' % check_entry_pt.name)
+            doctor_warn(f'Unable to instantiate check object from {check_entry_pt.name}.')
         try:
             check_category = check_instance.category()
             result = check_instance.check()
             if result.error or (include_warnings and result.warning):
                 fail += 1
-                failed_cats.add(check_category)
+                fail_categories.add(check_category)
             total += 1
         except Exception:
-            doctor_warn('Fail to call %s class functions.' % check_entry_pt.name)
-    return failed_cats, fail, total
+            doctor_warn(f'Fail to call {check_entry_pt.name} class functions.')
+    return fail_categories, fail, total
 
 
 def generate_reports(*, categories=None) -> List[Report]:
@@ -122,11 +120,11 @@ def generate_reports(*, categories=None) -> List[Report]:
         try:
             report_class = report_entry_pt.load()
         except (ImportError, UnknownExtra):
-            doctor_warn('Report entry point %s fails to load.' % report_entry_pt.name)
+            doctor_warn(f'Report entry point {report_entry_pt.name} fails to load.')
         try:
             report_instance = report_class()
         except Exception:
-            doctor_warn('Unable to instantiate report object from %s.' % report_entry_pt.name)
+            doctor_warn(f'Unable to instantiate report object from {report_entry_pt.name}.')
         try:
             report_category = report_instance.category()
             report = report_instance.report()
@@ -136,5 +134,5 @@ def generate_reports(*, categories=None) -> List[Report]:
             else:
                 reports.append(report)
         except Exception:
-            doctor_warn('Fail to call %s class functions.' % report_entry_pt.name)
+            doctor_warn(f'Fail to call {report_entry_pt.name} class functions.')
     return reports

ros2doctor/ros2doctor/api/format.py

@@ -23,8 +23,9 @@ def format_print(report):
 
     :param report: Report object with name and items list
     """
-    print('\n ', report.name)
-    padding_num = compute_padding(report.items)
+    print('\n  ', report.name)
+    if report.items:
+        padding_num = compute_padding(report.items)
     for item_name, item_content in report.items:
         print('{:{padding}}: {}'.format(item_name, item_content, padding=padding_num))
 
@@ -68,9 +69,19 @@ def __exit__(self, t, v, trb):
 
 def doctor_warn(msg: str) -> None:
     """
-    Use CustomWarningFormat to print customized warning message.
+    Print customized warning message with package and line info.
 
     :param msg: warning message to be printed
     """
     with CustomWarningFormat():
-        warnings.warn(msg)
+        warnings.warn(msg, stacklevel=2)
+
+
+def doctor_error(msg: str) -> None:
+    """
+    Print customized error message with package and line info.
+
+    :param msg: error message to be printed
+    """
+    with CustomWarningFormat():
+        warnings.warn(f'ERROR: {msg}', stacklevel=2)

ros2doctor/ros2doctor/api/network.py

@@ -19,13 +19,15 @@
 from ros2doctor.api import DoctorReport
 from ros2doctor.api import Report
 from ros2doctor.api import Result
+from ros2doctor.api.format import doctor_error
 from ros2doctor.api.format import doctor_warn
 
 try:
     import ifcfg
 except ImportError:  # check import error for windows and osx
-    doctor_warn('Failed to import ifcfg. '
-                'Use `python -m pip install ifcfg` to install needed package.')
+    doctor_warn(
+        'Unable to import ifcfg. '
+        'Use `python3 -m pip install ifcfg` to install needed package.')
 
 
 def _is_unix_like_platform() -> bool:
@@ -62,22 +64,31 @@ def check(self):
         try:
             ifcfg_ifaces = ifcfg.interfaces()
         except NameError:
-            result.add_error('ERROR: ifcfg is not imported. Unable to run network check.')
+            doctor_error(
+                '`ifcfg` module is not imported. '
+                'Unable to run network check.')
+            result.add_error()
             return result
 
         has_loopback, has_non_loopback, has_multicast = _check_network_config_helper(ifcfg_ifaces)
         if not _is_unix_like_platform():
             if not has_loopback and not has_non_loopback:
                 # no flags found, otherwise one of them should be True.
-                print('No flags found. \
-                    Run `ipconfig` on cmd to check network interfaces.')
+                doctor_error(
+                    'No flags found. '
+                    'Run `ipconfig` on Windows or '
+                    'install `ifconfig` on Unix to check network interfaces.')
+                result.add_error()
                 return result
         if not has_loopback:
-            result.add_error('ERROR: No loopback IP address is found.')
+            doctor_error('No loopback IP address is found.')
+            result.add_error()
         if not has_non_loopback:
-            result.add_warning('Only loopback IP address is found.')
+            doctor_warn('Only loopback IP address is found.')
+            result.add_warning()
         if not has_multicast:
-            result.add_warning('No multicast IP address is found.')
+            doctor_warn('No multicast IP address is found.')
+            result.add_warning()
         return result
 
 
@@ -93,7 +104,7 @@ def report(self):
         try:
             ifcfg_ifaces = ifcfg.interfaces()
         except NameError:
-            doctor_warn('ifcfg is not imported. Unable to generate network report.')
+            doctor_error('ifcfg is not imported. Unable to generate network report.')
             return Report('')
 
         network_report = Report('NETWORK CONFIGURATION')

ros2doctor/ros2doctor/api/package.py

@@ -13,8 +13,7 @@
 # limitations under the License.
 
 import os
-from typing import List
-from urllib.error import URLError
+import textwrap
 
 from ament_index_python import get_packages_with_prefixes
 from catkin_pkg.package import parse_package
@@ -24,6 +23,8 @@
 from ros2doctor.api import DoctorReport
 from ros2doctor.api import Report
 from ros2doctor.api import Result
+from ros2doctor.api.format import doctor_error
+from ros2doctor.api.format import doctor_warn
 
 import rosdistro
 
@@ -35,11 +36,26 @@ def get_distro_package_versions() -> dict:
     :return: dictionary of rosdistro package name and version
     """
     distro_name = os.environ.get('ROS_DISTRO')
+    if not distro_name:
+        doctor_error('ROS_DISTRO is not set.')
+        return
     distro_name = distro_name.lower()
     url = rosdistro.get_index_url()
+    if not url:
+        doctor_error(
+            'Unable to access ROSDISTRO_INDEX_URL or DEFAULT_INDEX_URL. '
+            'Check network setting to make sure machine is connected to internet.')
+        return
     i = rosdistro.get_index(url)
-    distro_data = rosdistro.get_distribution(i, distro_name).get_data()
-    repos_info = distro_data.get('repositories')
+    distro_info = rosdistro.get_distribution(i, distro_name)
+    if not distro_info:
+        doctor_warn(f'Distribution name {distro_name} is not found')
+        return
+    try:
+        repos_info = distro_info.get_data().get('repositories')
+    except AttributeError:
+        doctor_warn('No repository information found.')
+        return
     distro_package_vers = {}
     for _, info in repos_info.items():
         try:
@@ -69,7 +85,7 @@ def get_local_package_versions() -> dict:
     return local_packages
 
 
-def compare_versions(local_packages: dict, distro_packages: dict) -> List:
+def compare_versions(result: Result, local_packages: dict, distro_packages: dict):
     """
     Return warning messages for PackageCheck, and info for PackageReport.
 
@@ -78,7 +94,6 @@ def compare_versions(local_packages: dict, distro_packages: dict) -> List:
     :param: boolean value determines which output to populate, msgs or report
     :return: list of warning messages
     """
-    warning_msgs = []
     missing_req = ''
     missing_local = ''
     for name, local_ver_str in local_packages.items():
@@ -91,14 +106,31 @@ def compare_versions(local_packages: dict, distro_packages: dict) -> List:
         local_ver = version.parse(local_ver_str).base_version
         required_ver = version.parse(required_ver_str).base_version
         if local_ver < required_ver:
-            warning_msgs.append(f'{name} has been updated to a new version.'
-                                f' local: {local_ver} <'
-                                f' required: {required_ver}')
+            doctor_warn(
+                f'{name} has been updated to a new version.'
+                f' local: {local_ver} <'
+                f' required: {required_ver}')
+            result.add_warning()
     if missing_req:
-        warning_msgs.append('Cannot find required versions of packages:' + missing_req)
+        if len(missing_req) > 100:
+            doctor_warn(
+                'Cannot find required versions of packages: ' +
+                textwrap.shorten(missing_req, width=100) +
+                ' Use `ros2 doctor --report` to see full list.')
+        else:
+            doctor_warn(
+                'Cannot find required versions of packages: ' +
+                missing_req)
     if missing_local:
-        warning_msgs.append('Cannot find local versions of packages:' + missing_local)
-    return warning_msgs
+        if len(missing_local) > 100:
+            doctor_warn(
+                'Cannot find local versions of packages: ' +
+                textwrap.shorten(missing_local, width=100) +
+                ' Use `ros2 doctor --report` to see full list.')
+        else:
+            doctor_warn(
+                'Cannot find local versions of packages: ' +
+                missing_local)
 
 
 class PackageCheck(DoctorCheck):
@@ -110,21 +142,18 @@ def category(self):
     def check(self):
         """Check packages within the directory where command is called."""
         result = Result()
-        try:
-            distro_package_vers = get_distro_package_versions()
-            if not distro_package_vers:
-                result.add_error('ERROR: distro packages info is not found.')
-        except (AttributeError, RuntimeError, URLError):
-            result.add_error('ERROR: Unable to fetch package information from rosdistro.')
+        distro_package_vers = get_distro_package_versions()
+        if not distro_package_vers:
+            doctor_error('distro packages info is not found.')
+            result.add_error()
         local_package_vers = get_local_package_versions()
         if not local_package_vers:
-            result.add_error('ERROR: local package info is not found.')
+            doctor_error('local package info is not found.')
+            result.add_error()
         if result.error != 0:
             return result
 
-        warning_msgs = compare_versions(local_package_vers, distro_package_vers)
-        for msg in warning_msgs:
-            result.add_warning(msg)
+        compare_versions(result, local_package_vers, distro_package_vers)
         return result
 
 
@@ -137,11 +166,8 @@ def category(self):
     def report(self):
         """Report packages within the directory where command is called."""
         report = Report('PACKAGE VERSIONS')
-        try:
-            distro_package_vers = get_distro_package_versions()
-        except (AttributeError, RuntimeError, URLError):
-            return report
         local_package_vers = get_local_package_versions()
+        distro_package_vers = get_distro_package_versions()
         if not local_package_vers or not distro_package_vers:
             return report
         for name, local_ver_str in local_package_vers.items():

ros2doctor/ros2doctor/api/platform.py

@@ -20,6 +20,7 @@
 from ros2doctor.api import DoctorReport
 from ros2doctor.api import Report
 from ros2doctor.api import Result
+from ros2doctor.api.format import doctor_error
 from ros2doctor.api.format import doctor_warn
 
 import rosdistro
@@ -33,20 +34,24 @@ def _check_platform_helper() -> Tuple[str, dict, dict]:
     """
     distro_name = os.environ.get('ROS_DISTRO')
     if not distro_name:
-        doctor_warn('ROS_DISTRO is not set.')
+        doctor_error('ROS_DISTRO is not set.')
         return
-    else:
-        distro_name = distro_name.lower()
+    distro_name = distro_name.lower()
     u = rosdistro.get_index_url()
     if not u:
-        doctor_warn('Unable to access ROSDISTRO_INDEX_URL or DEFAULT_INDEX_URL.')
+        doctor_error(
+            'Unable to access ROSDISTRO_INDEX_URL or DEFAULT_INDEX_URL. '
+            'Check network setting to make sure machine is connected to internet.')
         return
     i = rosdistro.get_index(u)
     distro_info = i.distributions.get(distro_name)
     if not distro_info:
-        doctor_warn("Distribution name '%s' is not found" % distro_name)
+        doctor_warn(f'Distribution name {distro_name} is not found')
         return
-    distro_data = rosdistro.get_distribution(i, distro_name).get_data()
+    try:
+        distro_data = rosdistro.get_distribution(i, distro_name).get_data()
+    except AttributeError:
+        distro_data = ''
     return distro_name, distro_info, distro_data
 
 
@@ -61,19 +66,24 @@ def check(self):
         result = Result()
         distros = _check_platform_helper()
         if not distros:
-            result.add_error('ERROR: Missing rosdistro info. Unable to check platform.')
+            doctor_error('Missing rosdistro info. Unable to check platform.')
+            result.add_error()
             return result
         distro_name, distro_info, _ = distros
 
         # check distro status
         if distro_info.get('distribution_status') == 'prerelease':
-            result.add_warning('Distribution %s is not fully supported or tested. '
-                               'To get more consistent features, download a stable version at '
-                               'https://index.ros.org/doc/ros2/Installation/' % distro_name)
+            doctor_warn(
+                f'Distribution {distro_name} is not fully supported or tested. '
+                'To get more consistent features, download a stable version at '
+                'https://index.ros.org/doc/ros2/Installation/')
+            result.add_warning()
         elif distro_info.get('distribution_status') == 'end-of-life':
-            result.add_warning('Distribution %s is no longer supported or deprecated. '
-                               'To get the latest features, download the new versions at '
-                               'https://index.ros.org/doc/ros2/Installation/' % distro_name)
+            doctor_warn(
+                f'Distribution {distro_name} is no longer supported or deprecated. '
+                'To get the latest features, download the new versions at '
+                'https://index.ros.org/doc/ros2/Installation/')
+            result.add_warning()
         return result