82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
#
|
|
# 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.
|
|
#
|
|
from aiohttp import web
|
|
import time
|
|
from ament_index_python import get_package_share_directory
|
|
import yaml
|
|
|
|
def main():
|
|
facts = Facts()
|
|
routes = Routes(facts)
|
|
|
|
app = web.Application()
|
|
app.add_routes([
|
|
web.get('/', routes.root),
|
|
web.get('/ping', routes.ping),
|
|
web.get('/uptime', routes.uptime),
|
|
web.get('/build_info', routes.build_info)
|
|
])
|
|
web.run_app(app)
|
|
|
|
class Facts:
|
|
def __init__(self):
|
|
self._start_time = time.monotonic()
|
|
|
|
def get_uptime(self):
|
|
return time.monotonic() - self._start_time
|
|
|
|
def get_buildinfo(self):
|
|
# Find the share directory for 'build_info_getter'.
|
|
build_info_getter_directory = None
|
|
try:
|
|
build_info_getter_directory = get_package_share_directory('build_info_getter')
|
|
except Exception as e:
|
|
print("Can't find build info.\n{}".format(e))
|
|
return None
|
|
# Find and read the project_state.repos file in it.
|
|
project_state_file = build_info_getter_directory + "/project_state.repos"
|
|
project_state_content = None
|
|
try:
|
|
with open(project_state_file, 'r') as file_obj:
|
|
project_state_content = yaml.safe_load(file_obj)
|
|
except Exception as e:
|
|
# We either didn't load the file or couldn't read it
|
|
# as json.
|
|
print("Can't find build info.\n{}".format(e))
|
|
return project_state_content
|
|
|
|
class Routes:
|
|
def __init__(self, facts):
|
|
self._facts = facts
|
|
|
|
async def root(self, request):
|
|
text = 'hello!'
|
|
return web.Response(text=text)
|
|
|
|
async def ping(self, request):
|
|
text = 'pong'
|
|
return web.Response(text=text)
|
|
|
|
async def uptime(self, request):
|
|
resp = {"uptime": self._facts.get_uptime()}
|
|
return web.json_response(resp)
|
|
|
|
async def build_info(self, request):
|
|
project_state = self._facts.get_buildinfo()
|
|
if not project_state:
|
|
resp = {"status": False, "message": "project_state not found."}
|
|
return web.json_response(resp)
|
|
project_state["status"] = True
|
|
return web.json_response(project_state)
|
|
|
|
|
|
|