gui.jobs.InProcessJob

gui.jobs.InProcessJob(function)

A function that runs in a worker thread, one run at a time that counts.

Every start bumps a generation counter. A result of an older generation is dropped, so a slow run that the user has already replaced never overwrites the newer answer.

Examples

from pathlib import Path
from tempfile import TemporaryDirectory

from spotregen import load_spec, make_sample_repo, run_checks
from spotregen.gui.jobs import InProcessJob

with TemporaryDirectory() as tmp:
    spec = load_spec(make_sample_repo(Path(tmp) / "demo"))
    job = InProcessJob(run_checks)
    job.start(spec)
    print(job.wait().value)

[]

Attributes

Name	Description
running	Whether the latest run has not finished.

Methods

Name	Description
poll	The result of the latest run if it is ready, else None. Never blocks.
start	Run the function with these arguments in a new daemon thread.
wait	Block until the latest run has finished; meant for scripts and tests.

poll

gui.jobs.InProcessJob.poll()

The result of the latest run if it is ready, else None. Never blocks.

Returns

Name	Type	Description
	JobResult | None	JobResult | None: The newest finished result; results of older runs are discarded.

start

gui.jobs.InProcessJob.start(*args, **kwargs)

Run the function with these arguments in a new daemon thread.

Parameters

Name	Type	Description	Default
*args	Any	Positional arguments of the function.	()
**kwargs	Any	Keyword arguments of the function.	{}

Returns

Name	Type	Description
int	int	The generation of this run.

wait

gui.jobs.InProcessJob.wait(timeout=30.0)

Block until the latest run has finished; meant for scripts and tests.

Parameters

Name	Type	Description	Default
timeout	float	Seconds to wait.	30.0

Returns

Name	Type	Description
JobResult	JobResult	The result of the latest run.

Raises

Name	Type	Description
	TimeoutError	If the run does not finish in time.