Refactor. Add in diagnostics.

am_i_up/Api.py

@@ -28,8 +28,8 @@ class Api:
             web.get('/api/build_info', self.build_info),
             web.get('/api/env', self.env),
             web.get('/api/status', self.status),
-            web.get('/api/costmap_image', self.costmap_image),
             web.get("/api/position", self.position),
+            web.get("/api/diagnostics", self.diagnostics),
             web.static("/assets", ui_static_directory),
             # we're not actually using key anywhere, but doing this allows react router
             # to work correctly.
@@ -86,13 +86,6 @@ class Api:
         resp = {"message": status}
         return web.json_response(resp)
 
-    async def costmap_image(self, request):
-        (image_format, image_data) = self._facts.get_costmap_image()
-        if (not image_format) or (not image_data):
-            raise web.HTTPUnsupportedMediaType()
-        resp = web.Response(body=image_data, content_type='image/{}'.format(image_format))
-        return resp
-
     async def position(self, request):
         resp = {}
         (lat, lon) = self._facts.get_lat_long()
@@ -104,3 +97,13 @@ class Api:
         resp["longitude"] = lon
         return web.json_response(resp)
 
+    async def diagnostics(self, request):
+        resp = {}
+        diagnostics = self._facts.get_diagnostics_json()
+        if not diagnostics:
+            resp["status"] = False
+            return web.json_response(resp)
+        resp["status"] = True
+        resp["message"] = diagnostics
+        return web.json_response(resp)
+

am_i_up/Facts.py

@@ -8,39 +8,62 @@
 # This Source Code Form is "Incompatible With Secondary Licenses", as
 # defined by the Mozilla Public License, v. 2.0.
 #
+import rclpy
+from ament_index_python import get_package_share_directory
+from std_msgs.msg import String
+from sensor_msgs.msg import CompressedImage
+from sensor_msgs.msg import NavSatFix
+from diagnostic_msgs.msg import DiagnosticArray
+from diagnostic_msgs.msg import DiagnosticStatus
+import asyncio
 import time
 import ipaddress
 import subprocess
 import os
 import yaml
-from ament_index_python import get_package_share_directory
+import j7s_diagnostics_py
-
+import json
-from am_i_up.Ros import Ros
 
 class Facts:
     def __init__(self):
-        self._ros = Ros()
+
+        rclpy.init()
+        self._node = rclpy.create_node('am_i_up')
+
+        self._status_sub = self._node.create_subscription(String, "status", self._status_callback, 1)
+        self._navsat_sub = self._node.create_subscription(NavSatFix, "navsatfix", self._navsat_callback, 1)
+        self._diagnostics_sub = self._node.create_subscription(DiagnosticArray, "/diagnostics_agg", self._diag_callback, 1)
+
+        self._listen_port = self._node.declare_parameter("listen_port", "8888").value
+        self._ui_pkg = self._node.declare_parameter("ui_pkg", "am_i_up_ui").value
+
+        self._status_string = None
+        self._lat_long = None
+        self._diag_msg = None
 
         self._start_time = time.monotonic()
-        self._status_string = None
 
     async def run(self):
-        await self._ros.run()
+        while rclpy.ok():
+            rclpy.spin_once(self._node, timeout_sec=0)
+            await asyncio.sleep(1e-4)
 
     def get_lat_long(self):
-        return self._ros.get_lat_long()
+        return self._lat_long
-
-    def get_costmap_image(self):
-        return self._ros.get_costmap_image()
 
     def get_status(self):
-        return self._ros.get_status_string()
+        return self._status_string
 
     def get_listen_port(self):
-        return self._ros.get_listen_port()
+        return self._listen_port
 
     def get_ui_pkg(self):
-        return self._ros.get_ui_pkg()
+        return self._ui_pkg
+
+    def get_diagnostics_json(self):
+        if not self._diag_msg:
+            return None
+        return diag_msg_to_json(self._diag_msg)
 
     def get_uptime(self):
         return time.monotonic() - self._start_time
@@ -91,6 +114,56 @@ class Facts:
             print("Can't find build info.\n{}".format(e))
         return project_state_content
 
+    def _status_callback(self, msg):
+        self._status_string = msg.data
+
+    def _diag_callback(self, msg):
+        self._diag_msg = msg
+
+    def _navsat_callback(self, msg):
+        self._lat_long = (msg.latitude, msg.longitude)
+
+def diag_level_to_rust(level):
+    if level == DiagnosticStatus.OK:
+        return j7s_diagnostics_py.DiagnosticLevel.OK
+    if level == DiagnosticStatus.WARN:
+        return j7s_diagnostics_py.DiagnosticLevel.WARN
+    if level == DiagnosticStatus.ERROR:
+        return j7s_diagnostics_py.DiagnosticLevel.ERROR
+    if level == DiagnosticStatus.STALE:
+        return j7s_diagnostics_py.DiagnosticLevel.STALE
+    return j7s_diagnostics_py.DiagnosticLevel.UNSET
+
+def key_values_to_dict(values):
+    to_return = {}
+    for value in values:
+        to_return[value.key] = value.value
+    return to_return
+
+def diag_status_to_rust(status):
+    rust = j7s_diagnostics_py.DiagnosticStatus()
+    rust.level = diag_level_to_rust(status.level)
+    rust.name = status.name
+    rust.message = status.message
+    rust.hardware_id = status.hardware_id
+    rust.values = key_values_to_dict(status.values)
+
+    return rust
+
+def diag_msg_to_json(diagnostic_array):
+    # Make a DiagnosticTree.
+    tree = j7s_diagnostics_py.DiagnosticsTree()
+    # Iterate through the statuses in message.
+    for status in diagnostic_array.status:
+        # Convert the statues to the type used by the tree.
+        rust_status = diag_status_to_rust(status)
+        # Insert them into the tree.
+        tree.insert(rust_status)
+    # Call the to json function.
+    tree.reconcile_levels()
+    json_string = tree.to_json()
+    return json.loads(json_string)
+
 
 def is_valid_ip(address):
     try:

am_i_up/Ros.py

@@ -1,64 +0,0 @@
-#
-# Copyright 2025 James Pace
-#
-# This Source Code Form is subject to the terms of the Mozilla Public
-# License, v. 2.0. If a copy of the MPL was not distributed with this
-# file, You can obtain one at https://mozilla.org/MPL/2.0/.
-#
-# This Source Code Form is "Incompatible With Secondary Licenses", as
-# defined by the Mozilla Public License, v. 2.0.
-#
-import rclpy
-from std_msgs.msg import String
-from sensor_msgs.msg import CompressedImage
-from sensor_msgs.msg import NavSatFix
-import asyncio
-
-class Ros:
-    def __init__(self):
-        rclpy.init()
-        self._node = rclpy.create_node('am_i_up')
-
-        self._status_sub = self._node.create_subscription(String, "status", self._status_callback, 1)
-        self._image_sub = self._node.create_subscription(CompressedImage, "image/compressed", self._image_callback, 1)
-        self._navsat_sub = self._node.create_subscription(NavSatFix, "navsatfix", self._navsat_callback, 1)
-
-        self._listen_port = self._node.declare_parameter("listen_port", "8888").value
-        self._ui_pkg = self._node.declare_parameter("ui_pkg", "am_i_up_ui").value
-
-        self._status_string = None
-        self._lat_long = None
-        self._costmap_image = None
-
-    def get_status_string(self):
-        return self._status_string
-
-    def get_lat_long(self):
-        return self._lat_long
-
-    def get_costmap_image(self):
-        return self._costmap_image
-
-    def get_listen_port(self):
-        return self._listen_port
-
-    def get_ui_pkg(self):
-        return self._ui_pkg
-
-    async def run(self):
-        while rclpy.ok():
-            rclpy.spin_once(self._node, timeout_sec=0)
-            await asyncio.sleep(1e-4)
-
-    def _status_callback(self, msg):
-        self._status_string = msg.data
-
-    def _navsat_callback(self, msg):
-        self._lat_long = (msg.latitude, msg.longitude)
-
-    def _image_callback(self, msg):
-        if msg.format != "jpg" and msg.format != "png":
-            # I don't know what to do with this image....
-            return
-        self._costmap_image = (msg.format, bytes(msg.data))
-

am_i_up/client.py

@@ -8,6 +8,7 @@
 # This Source Code Form is "Incompatible With Secondary Licenses", as
 # defined by the Mozilla Public License, v. 2.0.
 #
+import argparse
 import aiohttp
 import asyncio
 import rclpy
@@ -15,25 +16,31 @@ import json
 
 def main(args=None):
 
-    rclpy.init(args=args)
+    parser = argparse.ArgumentParser()
-    node = rclpy.create_node('am_i_up_client')
+    parser.add_argument("--action")
+    parser.add_argument("--options", nargs="*")
+    parser.add_argument("--host", default="http://localhost:8888")
+    args = parser.parse_args()
 
-    action = node.declare_parameter('action', value="").value
-    options = node.declare_parameter('options', value=[""]).value
 
-    if "" == action:
+    client = Client(args.host)
-        raise RunTimeError("Need to provide an action to take.")
 
-    if action == 'ping':
+    if args.action == 'ping':
-        asyncio.run(call_ping(options))
+        asyncio.run(client.call_ping(args.options))
-    if action == 'uptime':
+    if args.action == 'uptime':
-        asyncio.run(call_uptime())
+        asyncio.run(client.call_uptime())
-    if action == 'build_info':
+    if args.action == 'build_info':
-        asyncio.run(call_build_info())
+        asyncio.run(client.call_build_info())
-    if action == 'env':
+    if args.action == 'env':
-        asyncio.run(call_env())
+        asyncio.run(client.call_env())
+    if args.action == 'diagnostics':
+        asyncio.run(client.call_diagnostics())
 
-async def call_ping(options):
+class Client():
+    def __init__(self, host):
+        self._host = host
+
+    async def call_ping(self, options):
         if len(options) != 1 or options[0] == "":
             raise RunTimeError("Ping option is an address as a string.")
 
@@ -42,26 +49,33 @@ async def call_ping(options):
         print("Calling ping with request: {}", json.dumps(request))
 
         async with aiohttp.ClientSession() as session:
-        async with session.get('http://localhost:8888/api/ping', json=request) as resp:
+            async with session.get(f'{self._host}/api/ping', json=request) as resp:
                 print(await resp.text())
 
-async def call_uptime():
+    async def call_uptime(self):
         print("Calling uptime.")
 
         async with aiohttp.ClientSession() as session:
-        async with session.get('http://localhost:8888/api/uptime') as resp:
+            async with session.get(f'{self._host}/api/uptime') as resp:
                 print(await resp.text())
 
-async def call_build_info():
+    async def call_build_info(self):
         print("Calling build_info.")
 
         async with aiohttp.ClientSession() as session:
-        async with session.get('http://localhost:8888/api/build_info') as resp:
+            async with session.get(f'{self._host}/api/build_info') as resp:
                 print(await resp.text())
 
-async def call_env():
+    async def call_env(self):
         print("Calling env.")
 
         async with aiohttp.ClientSession() as session:
-        async with session.get('http://localhost:8888/api/env') as resp:
+            async with session.get(f'{self._host}/api/env') as resp:
+                print(await resp.text())
+
+    async def call_diagnostics(self):
+        print("Calling diagnostics.")
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(f'{self._host}/api/diagnostics') as resp:
                 print(await resp.text())

am_i_up/server.py

@@ -13,9 +13,11 @@ from am_i_up.Facts import Facts
 
 import asyncio
 
-def main():
+async def main():
     facts = Facts()
     api = Api(facts)
 
     future = asyncio.gather(api.run(), facts.run())
-    asyncio.get_event_loop().run_until_complete(future)
+    await future
+
+asyncio.run(main())

package.xml

@@ -12,6 +12,7 @@
   <depend>python3-aiohttp</depend>
   <!-- TODO: Run time dep on ping -->
 
+  <exec_depend>j7s_diagnostics_py</exec_depend>
   <exec_depend>rclpy</exec_depend>
 
   <export>