pythonimmediate.engine module

Abstract engine class.

class pythonimmediate.engine.ChildProcessEngine(engine_name: EngineName, args: Iterable[str] = (), env: Optional[dict[str, str]] = None, autorestart: bool = False, debug_log_communication: Optional[Union[str, Path]] = None, from_dump: bool = False)[source]

Bases: Engine

An object that represents a \(\TeX\) engine that runs as a subprocess of this process.

Can be used as a context manager to automatically close the subprocess when the context is exited. Alternatively close() can be used to manually terminate the subprocess.

For example, the following Python code, if run alone, will spawn a \(\TeX\) process and use it to write “Hello world” to a file named a.txt in the temporary directory:

>>> from pythonimmediate.engine import ChildProcessEngine, default_engine
>>> from pythonimmediate import execute
>>> from pathlib import Path
>>> with ChildProcessEngine("pdftex") as engine, default_engine.set_engine(engine):
...     # do something with the engine, for example:
...     execute(r'''
...     \immediate\openout9=a.txt
...     \immediate\write9{Hello world}
...     \immediate\closeout9
...     ''')
...     (Path(engine.directory)/"a.txt").read_text()
'Hello world\n'
>>> # now the engine is closed and the file is cleaned up.
>>> (Path(engine.directory)/"a.txt").is_file()
False

You can also use this to generate PDF file programmatically:

>>> with ChildProcessEngine("pdftex") as engine, default_engine.set_engine(engine):
...     execute(r'\documentclass{article}')
...     execute(r'\begin{document}')
...     execute(r'Hello world')
...     engine.terminate()
...     print(engine.read_output_file()[:9])
b'%PDF-1.5\n'

See DefaultEngine for how to use the resulting engine object.

We attempt to do correct error detection on the Python side by parsing the output/log file:

>>> with ChildProcessEngine("pdftex") as engine, default_engine.set_engine(engine):
...     execute(r'\error')
Traceback (most recent call last):
    ...
pythonimmediate.engine.TeXProcessError: Undefined control sequence.

However, it’s not always easy. For example the following code will reproduce the output on error identically:

\message{^^J! error.^^J%
l.1 \errmessage{error}^^J%
^^J%
? }
\readline -1 to \xx

As an alternative to terminate(), you can also just execute \end{document}, but be sure to catch TeXProcessExited if you do so. If you do this, there’s no need to call terminate().

>>> with ChildProcessEngine("pdftex") as engine, default_engine.set_engine(engine):
...     execute(r'\documentclass{article}')
...     execute(r'\begin{document}')
...     execute(r'Hello world')
...     try: execute(r'\end{document}')
...     except TeXProcessExited: pass
...     print(engine.read_output_file()[:9])
b'%PDF-1.5\n'
Parameters

  • args – List of additional arguments to be passed to the executable, such as --recorder etc.

  • env – See documentation of env argument in subprocess.Popen. Note that it’s usually recommended to append environment variables that should be set to os.environ instead of replacing it entirely.

  • from_dump –

    Slightly undocumented feature at the moment.

    Refer to https://tex.stackexchange.com/a/687427 for explanation. Decompressing the format file is optional, explained in that link.

    Example:

    >>> from pythonimmediate import BalancedTokenList
>>> from pathlib import Path
>>> import tempfile
>>> import gzip
>>> with ChildProcessEngine("pdftex", args=["--ini", "&pdflatex"]) as engine, default_engine.set_engine(engine):
...     try: execute(r"\documentclass{article} \usepackage{tikz} \pythonimmediatechildprocessdump")
...     except TeXProcessExited: pass
...     else: assert False
...     f=Path(tempfile.mktemp(suffix=".fmt"))
...     _size=f.write_bytes(gzip.decompress(engine.read_output_file("fmt")))
>>> with ChildProcessEngine("pdftex", from_dump=True, args=["&"+str(f.with_suffix(""))]) as engine, default_engine.set_engine(engine):
...     BalancedTokenList(r"\the\numexpr 12+34\relax").expand_o()
<BalancedTokenList: 4₁₂ 6₁₂>
>>> f.unlink()

  • autorestart –

    Mostly for testing purpose – whenever an error happen, the engine will be killed and automatically restarted.

    For example:

    >>> from pythonimmediate import count
>>> engine=ChildProcessEngine("pdftex")
>>> with default_engine.set_engine(engine): execute(r'\error')
Traceback (most recent call last):
    ...
pythonimmediate.engine.TeXProcessError: Undefined control sequence.
>>> with default_engine.set_engine(engine): count[0]
Traceback (most recent call last):
    ...
pythonimmediate.engine.TeXProcessError: error already happened

    >>> engine=ChildProcessEngine("pdftex", autorestart=True)
>>> count[0]=2
>>> with default_engine.set_engine(engine): execute(r'\error')
Traceback (most recent call last):
    ...
pythonimmediate.engine.TeXProcessError: Undefined control sequence.
>>> with default_engine.set_engine(engine): count[0]
1
>>> with default_engine.set_engine(engine): execute(r'\error')
Traceback (most recent call last):
    ...
pythonimmediate.engine.TeXProcessError: Undefined control sequence.

read_output_file(extension: str = 'pdf') → bytes[source]

Read the output file with the given extension.

This is only reliable when the process has already been terminated. Refer to terminate().

See ChildProcessEngine for an usage example. See MultiChildProcessEngine for another example.

terminate() → None[source]

Terminate the current process.

This must be used in place of close() in order to stop the process but still keep the generated files.

See ChildProcessEngine for an usage example.

Similar to file.close, subprocess.Popen.wait or subprocess.Popen.kill, this method does nothing if the process is already terminated.

>>> from pythonimmediate import execute
>>> with ChildProcessEngine("pdftex") as engine, default_engine.set_engine(engine):
...     execute(r'\typeout{123}')
...     engine.terminate()
...     engine.terminate()
...     engine.close()
>>> engine.terminate()
>>> engine.close()
>>> engine.close()
class pythonimmediate.engine.DefaultEngine[source]

Bases: object

A convenience class that can be used to avoid passing explicit engine argument to functions.

This is thread-safe, which means that each thread can have its own set default engine and set_engine() for one thread does not affect other threads.

Users should not instantiate this class directly. Instead, use default_engine.

Usage example:

default_engine.set_engine(engine)  # set only for this thread
execute("hello world")  # this is executed on engine=engine

See also

set_engine()

property engine: Optional[Engine]

Return the engine, or None if the engine is not set.

get_engine() → Engine[source]

Convenience helper function, return the engine.

All the other functions that use this one (those that make use of the engine) will raise RuntimeError if the engine is None.

set_engine(engine: Optional[Engine]) → _SetDefaultEngineContextManager[source]

Set the default engine to another engine.

Can also be used as a context manager to revert to the original engine. Example:

>>> from pythonimmediate import execute
>>> _new_engine=ChildProcessEngine("pdftex")
>>> with default_engine.set_engine(_new_engine) as engine:  # only for this thread
...     assert engine is _new_engine
...     assert default_engine.engine is _new_engine
...     execute(r"\count0=5")
>>> # now the original engine is restored
>>> _new_engine.close()

Note that the following form, while allowed, is discouraged because it may cause resource leak (the engine may keeps running even after the block exits depends on whether it’s garbage-collected):

>>> with default_engine.set_engine(ChildProcessEngine("pdftex")):
...     pass

It’s recommended to write the following instead, which will close e at the end of the block:

>>> with ChildProcessEngine("pdftex") as e, default_engine.set_engine(e):
...     pass
class pythonimmediate.engine.Engine[source]

Bases: ABC

add_on_close(f: Callable[[Engine], None]) → None[source]

Add a function that will be executed when the engine is closed.

This function takes the engine object itself as the only argument to avoid a circular reference issue which would prevent the engine from being garbage-collected.

>>> e=ChildProcessEngine("pdftex")
>>> e.add_on_close(lambda _: print(1))
>>> e.add_on_close(lambda _: print(2))
>>> e.close()
1
2

The same engine may be closed multiple times in case of autorestart, in that case the function will only be called once.

>>> e=ChildProcessEngine("pdftex", autorestart=True)
>>> a=[1, 2, 3]
>>> e.add_on_close(lambda e: a.pop())
>>> from pythonimmediate import execute
>>> with default_engine.set_engine(e): execute(r"\error")
Traceback (most recent call last):
    ...
pythonimmediate.engine.TeXProcessError: Undefined control sequence.
>>> a
[1, 2, 3]
>>> e.close()
>>> a
[1, 2]
close() → None[source]

Terminates the \(\TeX\) subprocess gracefully.

property config: GlobalConfiguration

Self-explanatory.

property is_unicode: bool

Self-explanatory.

property name: EngineName

Self-explanatory.

read() → bytes[source]

Internal function.

Read one line from the engine.

It must not be EOF otherwise there’s an error.

The returned line does not contain the newline character.

pythonimmediate.engine.EngineName

The EngineName type is a string that specifies the name of the engine.

alias of Literal[‘pdftex’, ‘xetex’, ‘luatex’]

class pythonimmediate.engine.EngineStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Represent the status of the engine. Helper functions are supposed to manage the status by itself.

  • running: the engine is running \(\TeX\) code and Python cannot write to it, only wait for Python commands.

  • waiting: the engine is waiting for Python to write a handler.

  • error: an error happened.

  • exited: the engine exited cleanly.

class pythonimmediate.engine.ParentProcessEngine(pseudo_config: GlobalConfiguration)[source]

Bases: Engine

Represent the engine if this process is started by the \(\TeX\)’s pythonimmediate library.

This should not be instantiated directly. Only pythonimmediate.textopy.main() should instantiate this.

exception pythonimmediate.engine.TeXProcessError[source]

Bases: RuntimeError

exception pythonimmediate.engine.TeXProcessExited[source]

Bases: Exception

An exception that will be raised if some operation makes the process exits.

It is, however, safe to just catch this in case of ChildProcessEngine. See there for an example.

>>> from pythonimmediate import execute, BalancedTokenList
>>> execute(r'\documentclass{article}\begin{document}hello world')
>>> execute(r'\end{document}')
Traceback (most recent call last):
    ...
pythonimmediate.engine.TeXProcessExited
>>> execute(r'\documentclass{article}\begin{document}hello world')
>>> BalancedTokenList(r'\end{document}').execute()
Traceback (most recent call last):
    ...
pythonimmediate.engine.TeXProcessExited
pythonimmediate.engine.default_engine = <pythonimmediate.engine.DefaultEngine object>

A constant that can be used to avoid passing explicit engine argument to functions.

See documentation of DefaultEngine for more details.

For Python running inside a \(\TeX\) process, useful attributes are name and is_unicode.