blob: 948b114a6962bc8a7123efdfa3f95f9229c37b2f [file] [edit]
# Copyright (c) 2026 Intel Corporation
# SPDX-License-Identifier: Apache-2.0
"""Base class shared by NativeSimLibrary and QemuSimLibrary."""
import re
import threading
import time
class SimLibraryBase:
"""Common Robot Framework keyword base for Zephyr simulator libraries.
Provides shared output-buffering state and the ``wait_for_output`` /
``wait_for_literal_output`` keywords used by both the native_sim and
QEMU back-ends.
"""
ROBOT_LIBRARY_SCOPE = 'SUITE'
def __init__(self):
self._buffer = ''
self._lock = threading.Lock()
self._reader = None
@staticmethod
def _strip_ansi(data):
"""Decode *data* bytes and strip ANSI escape sequences."""
return re.sub(
r'\x1b\[[0-9;]*[mABCDEFGHJKSTfhilmnsu]',
'',
data.decode('utf-8', errors='replace'),
)
def wait_for_output(self, pattern, timeout=15):
"""Wait until *pattern* (regex) appears in the simulator's output."""
end_time = time.time() + float(timeout)
regex = re.compile(pattern, re.MULTILINE)
while time.time() < end_time:
with self._lock:
m = regex.search(self._buffer)
if m:
return m.group(0)
time.sleep(0.05)
with self._lock:
captured = self._buffer
raise AssertionError(
f"Pattern '{pattern}' not found within {timeout}s.\nCaptured output:\n{captured}"
)
def wait_for_literal_output(self, text, timeout=15):
"""Wait until the literal *text* string appears in the simulator's output."""
end_time = time.time() + float(timeout)
while time.time() < end_time:
with self._lock:
if text in self._buffer:
return text
time.sleep(0.05)
with self._lock:
captured = self._buffer
raise AssertionError(
f"Text '{text}' not found within {timeout}s.\nCaptured output:\n{captured}"
)