tomato.daemon: Functions and modules comprising the tomato daemon

Code author: Peter Kraus

tomato.daemon.setup_logging(daemon: Daemon)

Helper function to set up logging (folder, filename, verbosity, format) based on the passed daemon state.

tomato.daemon.tomato_daemon()

The function called when tomato-daemon is executed.

Manages the state of the tomato daemon, spawning manager threads for jobs (job), drivers (driver), and pipelines (pip). Parses the configuration in the settings file and devices file.

Submodules

tomato.daemon.cmd: command parsing for tomato daemon

Code author: Peter Kraus

All functions in this module expect a dict containing the command specification and a Daemon object as arguments. The Daemon object may be altered by the command.

All functions in this module return a Reply.

tomato.daemon.cmd.status(msg: dict, daemon: Daemon) → Reply

Return daemon status, containing the current tomato configuration.

tomato.daemon.cmd.stop(msg: dict, daemon: Daemon) → Reply

Stop the tomato daemon.

tomato.daemon.cmd.reload(msg: dict, daemon: Daemon) → Reply

Set-up or reload the tomato daemon using its configuration files.

Note

When reloading settings, tomato checks whether any running jobs use resources (i.e. drivers, pipelines, components) that would be removed if the new configuration were to be applied. If this is the case, the configuration will not be updated.

tomato.daemon.crates: functions for creating RO-crates

Code author: Peter Kraus

tomato.daemon.crates.RepositoryObject(crate: ROCrate, identifier: str | None = None, properties: dict | None = None) → ContextEntity

tomato.daemon.crates.Profile(crate: ROCrate, identifier: str = 'https://github.com/MADICES/MADICES-2025/discussions/25', properties: dict | None = None) → ContextEntity

tomato.daemon.crates.to_rocrate(datapath: str, userid: str, sampleid: str, make_child: bool = True) → ROCrate | None

tomato.daemon.db: shared functions using sqlite3

Code author: Peter Kraus

tomato.daemon.db.connect_db(dbpath: str | Path)

tomato.daemon.db.setup_db(dbpath: str | Path) → None

tomato.daemon.driver: the driver manager of tomato daemon

Code author: Peter Kraus

tomato.daemon.driver.tomato_driver_bootstrap(req: Socket, logger: Logger, interface: ModelInterface | ModelInterface, driver: str)

Function that attempts to register all configured components for this driver.

This helper function is executed when the register command is set to the driver process. The daemon is first polled for up-to-date configuration, and then each of the returned components is registered, if necessary, using cmp_register() of the driver interface.

In case the registration fails, a limited number of retries (as specified by the MAX_REGISTER_RETRIES constant) can be attempted on subsequent runs of this function.

tomato.daemon.driver.perform_idle_measurements(interface: ModelInterface | ModelInterface, t_last: float | None) → float | None

Function running idle measurements on the driver.

This function periodically runs the cmp_measure() on each component on the driver. The interval is determined from driver configuration using the "idle_measurement_interval" setting, driver defaults using the interface.idle_measurement_interval object, or tomato default (IDLE_MEASUREMENT_INTERVAL).

Note

How idle measurements are handled is up to the individual driver. By default, the cmp_measure() function will not submit new measurements when a task or a measurement is already running.

tomato.daemon.driver.stop_tomato_driver(port: int) → Reply

The default mechanism for stopping tomato drivers.

This function is used by the tomato driver manager to gracefully stop the driver, if an existing driver port is known.

tomato.daemon.driver.kill_tomato_driver(pid: int)

The backup mechanism for stoping tomato drivers.

This function is useful if the driver port is unknown or not responsive.

Wrapper around psutil.terminate(). Here we kill the (grand)children of the process with the name of tomato-job, i.e. the individual task functions. This allows the tomato-job process to exit gracefully once the task functions join.

Note

On Windows, the tomato-job.exe process has two children: a python.exe process which is the actual process running the job, and conhost.exe process, which we want to avoid killing.

tomato.daemon.driver.tomato_driver() → None

The function called when tomato-driver is executed.

This function is responsible for managing all activities involving devices of a single driver type.

First, the list of devices (and their channel/address) for the specified driver is fetched from the tomato-daemon. Then, a new instance of the specified driver is spawned, populating its device map using the above list. The state of the driver is stored .

Afterwards, the main loop handles all requests related to each of the devices managed by this driver process, including job commands. Finally, if the driver is instructed to stop, it attempts to perform a teardown before exiting.

tomato.daemon.driver.manager(timeout: int = 1000)

The driver manager thread of tomato-daemon.

This manager ensures individual driver processes are (re-)spawned and instructed to quit as necessary. The drivers are periodically checked using the HEARTBEAT constant as the interval. All changes are stored in the drivers table.

tomato.daemon.drvdb: the sqlite database for drivers in tomato

Code author: Peter Kraus

tomato.daemon.drvdb.insert_drv(drv: DrvState, dbpath: str) → DrvState | None

tomato.daemon.drvdb.update_drv(name: str, params: dict, dbpath: str) → DrvState | None

tomato.daemon.drvdb.get_drv(name: str, dbpath: str) → DrvState | None

tomato.daemon.drvdb.get_drvs_where(where: str, dbpath: str) → list[DrvState]

tomato.daemon.drvdb.del_drv(name: str, dbpath: str) → DrvState | None

tomato.daemon.io: functions for storing job data

Code author: Peter Kraus

tomato.daemon.io.merge_netcdfs(job: Job, snapshot=False) → str

Merges all of the individual pickled Dataset files from each component found in job.jobpath into a single DataTree, which is then stored in the NetCDF file. The role of each component is used as the group label.

tomato.daemon.io.data_to_pickle(ds: Dataset, path: Path, role: str)

Dumps the data provided as Dataset using pickle. Concatenates the new data with any existing data stored in the existing .pkl file.

tomato.daemon.job: the job manager of tomato daemon

Code author: Peter Kraus

tomato.daemon.job.method_validate(method: Sequence[Task], pip: Pipeline, daemon: Daemon) → bool

Function for validating Task parameters against components.

This function finds a component on the pipeline that matches the role of each Task, and passes each Task to the task_validate() function of that component.

Note

This function should be used to theck the method only after a matching pipeline has been identified, e.g. using find_matching_pipelines().

tomato.daemon.job.find_matching_pipelines(daemon: Daemon, method: Sequence[Task]) → list[str]

Function for finding the names of pipelines that match the provided method.

The matching is performed using the required roles of the method as well as using the required capabilities of the method. The role-matching can be checked statically against the daemon.devicefile; the latter is checked dynamically by polling each component for its capabilities using cmp_capabilities().

tomato.daemon.job.kill_tomato_job(process: Process)

Wrapper around psutil.terminate().

Here we kill the (grand)children of the process with the name of tomato-job, i.e. the individual task functions. This allows the tomato-job process to exit gracefully once the task functions join.

Note

On Windows, the tomato-job.exe process has two children: a python.exe process, which is the actual process running the job, and conhost.exe process, which we want to avoid killing.

tomato.daemon.job.manage_running(daemon: Daemon)

Function that manages jobs within the tomato job manager.

The function only affects jobs marked as running, i.e. with a set pid. Jobs scheduled for killing (i.e. status == "rd") are terminated. Jobs that are supposed to be running but have crashed are given appropriate status ("ce").

Note

Successful job completions are not processed here, but within the job process.

tomato.daemon.job.check_queued(daemon: Daemon) → dict[int, list[str]]

Function to check whether the queued jobs can be submitted onto any configured pipeline.

Returns a dict containing the jobids as keys and lists of matched Pipelines as values.

tomato.daemon.job.action_queued(daemon: Daemon, matched: dict[int, list[str]])

Function that assigns jobs if the pipeline is ready and contains the requested sample.

Warning

No validation except checking the sample name and readiness is performed. A matching and validated pipeline has to be previously identified, using e.g. the check_queued() function.

Note

The tomato-job process is launched from this function.

tomato.daemon.job.manager(timeout: int = 500)

The job manager thread of tomato-daemon.

This manager ensures the job queue is iterated over and jobs are submitted to pipelines.

Note

Note that we poll the tomato-daemon for configuration only once per iteration of the main loop.

tomato.daemon.job.tomato_job() → None

The function called when tomato-job is executed.

This function is responsible for managing all activities of a single job, including updating the queue table with the job pid, spawning of sub-processes to run tasks on each component of the pipeline, merging data at the end of the job, and updating the state of the pipeline once the job is successfully finished.

tomato.daemon.job.job_thread(role: str, tasks: list[Task], component: Component, device: Device, dport: int, dsettings: dict, jobpath: Path, logpath: Path)

A subthread of tomato-job, responsible for tasks on one component of a pipeline.

For each Task, this thread starts the task at an appropriate moment, then monitors the component status and polls periodically for data, and moves on to the next task as instructed in the task list.

Note

The data from all tasks for that component is stored using the tomato.daemon.io.data_to_pickle() function.

tomato.daemon.job.job_main_loop(port: int, job: Job, pipname: str, logpath: Path) → int | None

The main loop function of tomato-job, split for better readability.

tomato.daemon.jobdb: the sqlite database for jobs in tomato

Code author: Peter Kraus

tomato.daemon.jobdb.insert_job(job: Job, dbpath: str) → Job

tomato.daemon.jobdb.update_job_id(id: int, params: dict, dbpath: str) → Job

tomato.daemon.jobdb.get_job_id(id: int, dbpath: str) → Job

tomato.daemon.jobdb.get_jobs_where(where: str, dbpath: str) → list[Job]

tomato.daemon.lpp: a lazy pirate pattern implementation

Code author: Peter Kraus

tomato.daemon.lpp.comm(req: Socket, data: Any, endpoint: str, retries: int = 3, timeout: int = 1000, sender: str | None = None) → tuple[Reply, Socket]

Communication function based on the lazy pirate pattern.

This is a more robust version of the classic request–reply pattern, as the reply has a timeout (by default set to the REQ_TIMEOUT constant). The request–reply pattern is retried several times (by default REQ_RETRIES times), closing the connection between each retry.

tomato.daemon.pip: the pipeline manager of tomato daemon

Code author: Peter Kraus

tomato.daemon.pip.manager(timeout: int = 500)

The pipeline manager thread of tomato-daemon.

This manager ensures the job queue is iterated over and pipelines are managed/reset. Note that we poll the tomato-daemon for status only once per iteration of the main loop.

tomato.daemon.pipdb: the sqlite database for pipelines in tomato

Code author: Peter Kraus

tomato.daemon.pipdb.insert_pip(pip: PipState, dbpath: str) → PipState | None

tomato.daemon.pipdb.update_pip(name: str, params: dict, dbpath: str) → PipState | None

tomato.daemon.pipdb.get_pip(name: str, dbpath: str) → PipState | None

tomato.daemon.pipdb.get_pips_where(where: str, dbpath: str) → list[PipState]