Sage wrapper around pexpect’s spawn class and#

the ptyprocess’s PtyProcess class.

AUTHOR:

  • Jeroen Demeyer (2015-02-01): initial version, see Issue #17686.

  • Jeroen Demeyer (2015-12-04): add support for pexpect 4 + ptyprocess, see Issue #10295.

class sage.interfaces.sagespawn.SagePtyProcess(pid, fd)[source]#

Bases: PtyProcess

close(force=None)[source]#

Quit the child process: send the quit string, close the pseudo-tty and kill the process.

This function returns immediately, it doesn’t wait for the child process to die.

EXAMPLES:

sage: from sage.interfaces.sagespawn import SageSpawn
sage: s = SageSpawn("sleep 1000")
sage: s.close()
sage: while s.isalive():  # long time (5 seconds)
....:     sleep(float(0.1))

>>> from sage.all import *
>>> from sage.interfaces.sagespawn import SageSpawn
>>> s = SageSpawn("sleep 1000")
>>> s.close()
>>> while s.isalive():  # long time (5 seconds)
...     sleep(float(RealNumber('0.1')))
terminate_async(interval=5.0)[source]#

Terminate the child process group asynchronously.

This function returns immediately, while the child is slowly being killed in the background.

INPUT:

  • interval – (default: 5) how much seconds to wait between sending two signals.

EXAMPLES:

Run an infinite loop in the shell:

sage: from sage.interfaces.sagespawn import SageSpawn
sage: s = SageSpawn("sh", ["-c", "while true; do sleep 1; done"])

>>> from sage.all import *
>>> from sage.interfaces.sagespawn import SageSpawn
>>> s = SageSpawn("sh", ["-c", "while true; do sleep 1; done"])

Check that the process eventually dies after calling terminate_async:

sage: s.ptyproc.terminate_async(interval=float(0.2))
sage: while True:
....:     try:
....:         os.kill(s.pid, 0)
....:     except OSError:
....:         sleep(float(0.1))
....:     else:
....:         break  # process got killed

>>> from sage.all import *
>>> s.ptyproc.terminate_async(interval=float(RealNumber('0.2')))
>>> while True:
...     try:
...         os.kill(s.pid, Integer(0))
...     except OSError:
...         sleep(float(RealNumber('0.1')))
...     else:
...         break  # process got killed
class sage.interfaces.sagespawn.SageSpawn(*args, **kwds)[source]#

Bases: spawn

Spawn a subprocess in a pseudo-tty.

  • *args, **kwds: see pexpect.spawn.

  • name – human-readable name for this process, used for display purposes only.

  • quit_string – (default: None) if not None, send this string to the child process before killing it.

EXAMPLES:

sage: from sage.interfaces.sagespawn import SageSpawn
sage: SageSpawn("sleep 1", name="Sleeping Beauty")
Sleeping Beauty with PID ... running ...

>>> from sage.all import *
>>> from sage.interfaces.sagespawn import SageSpawn
>>> SageSpawn("sleep 1", name="Sleeping Beauty")
Sleeping Beauty with PID ... running ...
expect_peek(*args, **kwds)[source]#

Like expect() but restore the read buffer such that it looks like nothing was actually read. The next reading will continue at the current position.

EXAMPLES:

sage: from sage.interfaces.sagespawn import SageSpawn
sage: E = SageSpawn("sh", ["-c", "echo hello world"])
sage: _ = E.expect_peek("w")
sage: E.read().decode('ascii')
'hello world\r\n'

>>> from sage.all import *
>>> from sage.interfaces.sagespawn import SageSpawn
>>> E = SageSpawn("sh", ["-c", "echo hello world"])
>>> _ = E.expect_peek("w")
>>> E.read().decode('ascii')
'hello world\r\n'
expect_upto(*args, **kwds)[source]#

Like expect() but restore the read buffer starting from the matched string. The next reading will continue starting with the matched string.

EXAMPLES:

sage: from sage.interfaces.sagespawn import SageSpawn
sage: E = SageSpawn("sh", ["-c", "echo hello world"])
sage: _ = E.expect_upto("w")
sage: E.read().decode('ascii')
'world\r\n'

>>> from sage.all import *
>>> from sage.interfaces.sagespawn import SageSpawn
>>> E = SageSpawn("sh", ["-c", "echo hello world"])
>>> _ = E.expect_upto("w")
>>> E.read().decode('ascii')
'world\r\n'