feat: create tests for all currect functionality and add get available tags router

ref: N25B-334
This commit is contained in:
Björn Otgaar
2025-12-04 16:33:27 +01:00
parent fe4a060188
commit 531526f7bc
4 changed files with 769 additions and 8 deletions

View File

@@ -58,6 +58,45 @@ async def ping(request: Request):
pass
@router.get("/get_available_gesture_tags")
async def get_available_gesture_tags(request: Request):
"""
Endpoint to retrieve the available gesture tags for the robot.
:param request: The FastAPI request object.
:return: A list of available gesture tags.
"""
sub_socket = Context.instance().socket(zmq.SUB)
sub_socket.connect(settings.zmq_settings.internal_sub_address)
sub_socket.setsockopt(zmq.SUBSCRIBE, b"get_gestures")
pub_socket: Socket = request.app.state.endpoints_pub_socket
topic = b"send_gestures"
# TODO: Implement a way to get a certain ammount from the UI, rather than everything.
amount = None
timeout = 5 # seconds
await pub_socket.send_multipart([topic, amount.to_bytes(4, "big") if amount else b""])
try:
_, body = await asyncio.wait_for(sub_socket.recv_multipart(), timeout=timeout)
except TimeoutError:
body = b"tags: []"
logger.debug("got timeout error fetching gestures")
# Handle empty response and JSON decode errors
available_tags = []
if body:
try:
available_tags = json.loads(body).get("tags", [])
except json.JSONDecodeError as e:
logger.error(f"Failed to parse gesture tags JSON: {e}, body: {body}")
# Return empty list on JSON error
available_tags = []
return {"available_gesture_tags": available_tags}
@router.get("/ping_stream")
async def ping_stream(request: Request):
"""