run_iocsh

Package for running IOC and checking output.

Submodules

Attributes

Exceptions

IocshAlreadyRunningError

Exception raised when IOC is started a second time.

IocshFileNotFoundError

Exception raised when a file referenced in the startup script is not found.

IocshMissingSharedLibraryError

Exception raised when shared library is missing.

IocshModuleNotFoundError

Exception raised when the required module is not found.

IocshOutputError

Base exception for errors detected in IOC output.

IocshPatternMatchError

Exception raised when a fail_on pattern matches the IOC output.

IocshProcessError

Exception raised when the iocsh script exits with a non null return code.

IocshStartupError

Exception raised when IOC exits before the expected readiness pattern appears.

IocshStateError

Exception raised for programming errors (wrong state transitions).

IocshTimeoutError

Exception raised when a timeout occurred trying to send exit to the IOC.

RunIocshError

Base class for exceptions in this module.

Classes

IOC

Class to wrap IOC process.

Functions

run_iocsh(→ IOC)

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

wait_for(→ None)

Poll predicate until it returns True or timeout elapses.

Package Contents

exception run_iocsh.IocshAlreadyRunningError[source]

Bases: IocshStateError

Exception raised when IOC is started a second time.

exception run_iocsh.IocshFileNotFoundError[source]

Bases: IocshOutputError, FileNotFoundError

Exception raised when a file referenced in the startup script is not found.

exception run_iocsh.IocshMissingSharedLibraryError[source]

Bases: IocshOutputError

Exception raised when shared library is missing.

exception run_iocsh.IocshModuleNotFoundError[source]

Bases: IocshOutputError

Exception raised when the required module is not found.

exception run_iocsh.IocshOutputError[source]

Bases: RunIocshError

Base exception for errors detected in IOC output.

exception run_iocsh.IocshPatternMatchError[source]

Bases: IocshOutputError

Exception raised when a fail_on pattern matches the IOC output.

exception run_iocsh.IocshProcessError[source]

Bases: IocshOutputError

Exception raised when the iocsh script exits with a non null return code.

exception run_iocsh.IocshStartupError[source]

Bases: RunIocshError

Exception raised when IOC exits before the expected readiness pattern appears.

exception run_iocsh.IocshStateError[source]

Bases: RunIocshError

Exception raised for programming errors (wrong state transitions).

exception run_iocsh.IocshTimeoutError[source]

Bases: RunIocshError

Exception raised when a timeout occurred trying to send exit to the IOC.

exception run_iocsh.RunIocshError[source]

Bases: Exception

Base class for exceptions in this module.

run_iocsh.DEFAULT_FAIL_ON: tuple[str, Ellipsis][source]
class run_iocsh.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)

Bases: enum.Enum

Lifecycle state of the IOC subprocess.

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

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

property stdout: str

Return accumulated stdout as a single newline-joined string.

property stderr: str

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.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.

run_iocsh.wait_for(predicate: collections.abc.Callable[[], bool], timeout: float = DEFAULT_WAIT_FOR_TIMEOUT, poll_interval: float = DEFAULT_POLL_INTERVAL) None[source]

Poll predicate until it returns True or timeout elapses.

Exceptions from predicate are caught and treated as a false result (polling continues until timeout).

Args:

predicate: Callable invoked repeatedly; success when it returns True. timeout: Maximum seconds to wait before giving up. poll_interval: Seconds to sleep between polls.

Raises:

TimeoutError: If the predicate never returns True within timeout.