SubprocessTao

pytao.SubprocessTao

SubprocessTao(*args, env=None, cwd=None, **kwargs)

Bases: Tao

Subprocess helper for Tao.

This special version of the Tao class executes a Python subprocess which interacts with Tao through ctypes.

This can be used exactly as the normal Tao object with the primary added benefit that Fortran crashes will not affect the main Python process.

For full parameter information, see the Tao class documentation.

Usage

When creating many SubprocessTao objects, ensure to close the subprocess when done with it. This can be done manually:

>>> tao.close_subprocess()

Or automatically by way of a context manager:

>>> with SubprocessTao(init_file="$ACC_ROOT_DIR/bmad-doc/tao_examples/cbeta_cell/tao.init", plot=True) as tao:
...     tao.plot("floor")

To add a new environment variable in addition to the parent process environment:

>>> import os
>>> with SubprocessTao(init_file="...", env={**os.environ, "NEW_VAR": "NEW_VALUE"}) as tao:
...     print(tao.version())

Parameters:

Name	Type	Description	Default
env	dict[str, str] or None	Environment variables to use for the subprocess. If None, defaults to os.environ.	None
cwd	Path or str	The current working directory to use when spawning the subprocess. Any filenames used in initializing Tao will be relative to this. Supports environment variables and home directory (~) references. By default, this is the current working directory of your Python process.	None

Attributes:

Name	Type	Description
subprocess_env	dict[str, str]	Environment variables to use for the subprocess. It is recommended to use a new SubprocessTao instance in order to update these environment variables. However, while this dictionary may be updated in place, it will only be applied after the next subprocess starts and initializes. That is, tao.close_subprocess() and tao.init().

Source code in pytao/subproc.py

def __init__(
    self,
    *args,
    env: dict[str, str] | None = None,
    cwd: str | pathlib.Path | None = None,
    **kwargs,
):
    self._subproc_pipe_ = None
    self.subprocess_env = dict(env if env is not None else os.environ)
    self.subprocess_cwd = normalize_path(cwd) if cwd is not None else pathlib.Path(".")

    try:
        super().__init__(*args, **kwargs)
    except Exception:
        try:
            self.close_subprocess()
        except Exception as ex:
            logger.debug("Failed to close subprocess: %s", ex)
        raise

Attributes

pytao.SubprocessTao.subprocess_alive property

subprocess_alive

Subprocess is still running.

Methods:

pytao.SubprocessTao.close_subprocess

close_subprocess(*, force=False)

Close the Tao subprocess.

Source code in pytao/subproc.py

def close_subprocess(self, *, force: bool = False) -> None:
    """Close the Tao subprocess."""
    if self._subproc_pipe_ is not None:
        if force:
            self._subproc_pipe_.close_forcefully()
        else:
            self._subproc_pipe_.close()
    self._subproc_pipe_ = None

pytao.SubprocessTao.cmd

cmd(cmd, raises=True)

Runs a command, and returns the output.

Source code in pytao/subproc.py

@override
def cmd(self, cmd: str, raises: bool = True) -> list[str]:
    """Runs a command, and returns the output."""
    logger.debug(f"Tao> {cmd}")
    return self._send_command_through_pipe("cmd", cmd, raises=raises)

pytao.SubprocessTao.cmd_integer

cmd_integer(cmd, raises=True)

Runs a command, and returns an integer array.

Source code in pytao/subproc.py

@override
def cmd_integer(self, cmd: str, raises: bool = True) -> np.ndarray | None:
    """Runs a command, and returns an integer array."""
    return cast(
        Optional[np.ndarray],
        self._send_command_through_pipe("cmd_integer", cmd, raises=raises),
    )

pytao.SubprocessTao.cmd_real

cmd_real(cmd, raises=True)

Runs a command, and returns a floating point array.

Source code in pytao/subproc.py

@override
def cmd_real(self, cmd: str, raises: bool = True) -> np.ndarray | None:
    """Runs a command, and returns a floating point array."""
    return cast(
        Optional[np.ndarray],
        self._send_command_through_pipe("cmd_real", cmd, raises=raises),
    )

pytao.SubprocessTao.get_active_beam_track_element

get_active_beam_track_element()

Get the active element index being tracked via shared memory.

Source code in pytao/subproc.py

def get_active_beam_track_element(self) -> int:
    """Get the active element index being tracked via shared memory."""
    assert self._subproc_pipe_ is not None
    return self._subproc_pipe_.read_beam_track_element()

pytao.SubprocessTao.subprocess_call

subprocess_call(func, **kwargs)

Run a custom function in the subprocess.

The function must be readily importable by Python and not a dynamically created function or lambda. The first argument passed will be the tao object, and the remainder of the arguments are user-specified by keyword only.

Source code in pytao/subproc.py

def subprocess_call(self, func: Callable, **kwargs):
    """
    Run a custom function in the subprocess.

    The function must be readily importable by Python and not a dynamically
    created function or `lambda`.  The first argument passed will be the
    `tao` object, and the remainder of the arguments are user-specified by
    keyword only.
    """
    if not self.subprocess_alive:
        raise TaoDisconnectedError(
            "Tao subprocess is no longer running. Make a new `SubprocessTao` "
            "object or reinitialize with `.init()`."
        )
    assert self._subproc_pipe_ is not None
    return self._subproc_pipe_.send_receive_custom(func, kwargs)

pytao.SubprocessTao.timeout

timeout(when)

Context manager to set a timeout for a block of SubprocessTao calls.

Note that there is no possibility for a graceful timeout. In the event of a timeout, the subprocess will be terminated.

Parameters:

Name	Type	Description	Default
when	float	The timeout duration in seconds.	required

Yields:

Type	Description
None	Yields control back to the calling context.

Raises:

Type	Description
TimeoutError	If the block of code does not execute within when seconds. The Tao subprocess is forcefully terminated at this point.

Source code in pytao/subproc.py

@contextlib.contextmanager
def timeout(self, when: float):
    """
    Context manager to set a timeout for a block of SubprocessTao calls.

    Note that there is no possibility for a graceful timeout. In the event
    of a timeout, the subprocess will be terminated.

    Parameters
    ----------
    when : float
        The timeout duration in seconds.

    Yields
    ------
    None
        Yields control back to the calling context.

    Raises
    ------
    TimeoutError
        If the block of code does not execute within `when` seconds.
        The Tao subprocess is forcefully terminated at this point.
    """
    with subprocess_timeout_context([self], timeout=when):
        yield