run_iocsh.ioc

Class for running an IOC and capturing output.

Attributes

Classes

IOC

Class to wrap IOC process.

Functions

run_iocsh(→ IOC)

Start IOC, wait for iocInit, sleep delay seconds, exit, check output.

Module Contents

run_iocsh.ioc.log[source]
run_iocsh.ioc.RE_MODULE_NOT_AVAILABLE[source]
run_iocsh.ioc.RE_CANT_OPEN[source]
run_iocsh.ioc.RE_DOES_NOT_EXIST[source]
run_iocsh.ioc.RE_MISSING_SHARED_LIB[source]
run_iocsh.ioc.RE_BUILTIN_FAIL_ON = '^ERROR'[source]
run_iocsh.ioc.DEFAULT_FAIL_ON: tuple[str, Ellipsis][source]
run_iocsh.ioc.DEFAULT_EXECUTABLE = 'iocsh'[source]
run_iocsh.ioc.DEFAULT_EXIT_TIMEOUT: float | None = None[source]
run_iocsh.ioc.DEFAULT_WAIT_FOR_TIMEOUT = 5.0[source]
run_iocsh.ioc.DEFAULT_POLL_INTERVAL = 0.1[source]
run_iocsh.ioc.DEFAULT_DELAY = 5.0[source]
run_iocsh.ioc.DEFAULT_THREAD_TIMEOUT = 5.0[source]
run_iocsh.ioc.TAIL_CHARS = 500[source]
class run_iocsh.ioc.IOC(*args: str, executable: str = DEFAULT_EXECUTABLE, timeout: float | None = DEFAULT_EXIT_TIMEOUT, fail_on: collections.abc.Sequence[str] = DEFAULT_FAIL_ON)[source]

Class to wrap IOC process.

Not thread-safe: all public methods should be called from a single thread. Internal reader threads are managed by the class itself.

class State(*args, **kwds)[source]

Bases: enum.Enum

Lifecycle state of the IOC subprocess.

INITIALIZED[source]
STARTED[source]
EXITED[source]
proc = None[source]
args = ()[source]
executable = 'iocsh'[source]
timeout = None[source]
state[source]
__enter__() Self[source]
__exit__(exc_type: object, exc_value: object, exc_traceback: object) None[source]
property pid: int | None[source]

Return the subprocess PID, or None if not yet started.

property stdout: str[source]

Return accumulated stdout as a single newline-joined string.

property stderr: str[source]

Return accumulated stderr as a single newline-joined string.

is_running() bool[source]

Return True if the subprocess is still running.

This only reflects subprocess state — it does NOT indicate that iocInit has completed, that records are available, or that CA/PVA is ready to serve clients. Use wait_for_output() for IOC readiness checks.

start() None[source]

Start the IOC subprocess.

Raises:

IocshAlreadyRunningError: If the IOC is already running. IocshStateError: If the IOC has already exited.

exit() None[source]

Send the exit command to the running IOC.

wait_for_output(pattern: str = 'iocRun: All initialization complete', timeout: float = DEFAULT_WAIT_FOR_TIMEOUT, poll_interval: float = DEFAULT_POLL_INTERVAL) None[source]

Block until pattern appears in stdout or stderr.

Returns immediately if the pattern is already present in buffered output.

Args:

pattern: Regex pattern to search for in combined stdout+stderr. timeout: Maximum seconds to wait before raising. poll_interval: Seconds to sleep between polls.

Raises:

IocshStateError: If called before the process has started. IocshStartupError: If the IOC exits before the pattern appears. IocshTimeoutError: If timeout expires while the IOC is still running.

check_output(*, fail_on: collections.abc.Sequence[str] = DEFAULT_FAIL_ON) None[source]

Inspect accumulated output and raise on detected errors.

By default applies the DEFAULT_FAIL_ON patterns (^ERROR) plus the hardcoded checks (module not found, can’t open, missing shared library, file does not exist, non-zero exit code).

Args:
fail_on: Regex patterns to match against combined stdout+stderr.

Replaces DEFAULT_FAIL_ON entirely — pass (*DEFAULT_FAIL_ON, "MY:") to extend rather than replace. Pass () to disable pattern checks altogether.

Raises:

IocshStateError: If called before the process has exited. IocshPatternMatchError: If any pattern matches the output. IocshModuleNotFoundError: If a module failed to load. IocshFileNotFoundError: If a file could not be opened or does not exist. IocshMissingSharedLibraryError: If a required shared library is missing. IocshProcessError: If the process exited with a non-zero code.

run_iocsh.ioc.run_iocsh(*args: str, delay: float = DEFAULT_DELAY, timeout: float | None = DEFAULT_EXIT_TIMEOUT, executable: str = DEFAULT_EXECUTABLE, fail_on: collections.abc.Sequence[str] = DEFAULT_FAIL_ON) IOC[source]

Start IOC, wait for iocInit, sleep delay seconds, exit, check output.

Returns:

The exited IOC instance. Access .stdout and .stderr for output inspection after the call returns.