kromatography.solve package

Subpackages

Submodules

kromatography.solve.api module

API functions and classes for the Reveal Chromatography SOLVE sub-package.

kromatography.solve.api.run_cadet_simulator(input_file, use_slurm=False)[source]

API function to run CADET simulator for the given input file.

This is useful for people less comfortable with OOP, as well as for submitting cadet runs to other processes, since we can’t pickle CADETSimulationExecutor instances.

Parameters:
  • input_file (file_path) – The HDF5 input file for the CADET simulator.
  • use_slurm (bool) – Override the preferences and for to use SLURM or not? Leave blank to follow preferences.
Returns:

Returns the dictionary of information about the CADET run that the executor’s execute method returns.

Return type:

dict

kromatography.solve.api.run_simulations(simulation_list, job_manager=None, wait=True, timeout=None)[source]

Function to submit a list of simulations.

Can block the process and wait for all simulations to finish running.

Parameters:
  • simulation_list (list) – List of simulation instances to run.
  • job_manager (JobManager [OPTIONAL]) – JobManager instance to use to manage the CADET runs. If not provided, a new manager will be created.
  • wait (bool) – Should the process wait for all simulation to finish running?
  • timeout (int or None) – Number of seconds for the wait to timeout. Ignored if wait=False.
Returns:

Simulation runner created to run the provided simulations.

Return type:

SimulationRunner

kromatography.solve.api.create_cadet_file_for_sim(sim)[source]

Create

Parameters:sim (Simulation) – Simulation for which to create the CADET file.
Returns:File path created for the simulation.
Return type:str

kromatography.solve.cadet_executor module

Low level classes and functions to launch CADET binary on HDF5 input file.

class kromatography.solve.cadet_executor.BaseCADETExecutor(**traits)[source]

Bases: kromatography.solve.work_executor.WorkExecutor

A WorkExecutor to manage the execution of a CADET run on an input file.

This executor can be given a custom cadet solver binary command. Otherwise, it will search for it. Call execute once created, to launch CADET in a different process, and wait for that process to finish.

Since this object and methods are expected to be called from sub-processes, all relevant data is collected, and exceptions and logging are avoided.

cadet_binary = Str

The full path or the name of the CADET binary command to use.

num_cpu_per_task = Int

Number of CPUs per simulation run

solver_prefs = Instance(SolverPreferenceGroup)

Current Solver preferences

execute(input_file)[source]

Execute the CADET run on the provided input file.

Parameters:input_file (str) – Path to the HDF5 CADET input file.
Returns:
Return type:Dict with description of the parameters of the run.
call_executable(input_file)[source]
class kromatography.solve.cadet_executor.CADETExecutor(**traits)[source]

Bases: kromatography.solve.cadet_executor.BaseCADETExecutor

A WorkExecutor to manage the execution of a CADET run on an input file.

This executor can be given a custom cadet solver binary command. Otherwise, it will search for it. Call execute once created, to launch CADET in a different process, and wait for that process to finish.

Since this object and methods are expected to be called from sub-processes, all relevant data is collected, and exceptions and logging are avoided.

call_executable(input_file)[source]
kromatography.solve.cadet_executor.check_cadet_output(stdout)[source]

Try and guess when CADET failed to run properly by detecting exception raised and other major issues preventing from computing simulation output.

kromatography.solve.cadet_executor.check_valid_command(command)[source]

Return True if command is an executable command.

kromatography.solve.sim_runner_manager module

class kromatography.solve.sim_runner_manager.SimulationRunnerManager[source]

Bases: traits.has_traits.HasStrictTraits

Class to manager multiple simulation runners, and connect a sim to its runner.

known_runners = Set

Set of runner uuids in the _runner_list

add_runner(runner, index_sims=False)[source]

Add a new runner to the manager.

Parameters:
  • runner (SimulationRunner) – Simulation runner to keep track of.
  • index_sims (bool [OPTIONAL, default=False]) – Whether all sims submitted to the runner should be added to the internal maps for quick retrieval.
index_all_sims(runner_list=None)[source]

Rebuild maps for all simulations submitted to a list of runners.

kromatography.solve.simulation_job_utils module

Utilities to run CADET on a simualtion object.

kromatography.solve.simulation_job_utils.create_cadet_file_for_sim(sim)[source]

Create

Parameters:sim (Simulation) – Simulation for which to create the CADET file.
Returns:File path created for the simulation.
Return type:str
kromatography.solve.simulation_job_utils.prepare_simulation_for_run(sim)[source]

Prepare a simulation for the CADET solver to be run.

kromatography.solve.simulation_job_utils.walk_dataelement_editable(dataelement_obj, bool_val, skip_traits=None)[source]

Walks dataelement_obj recursively flipping editable flags to bool_val.

Parameters:
  • dataelement_obj (DataElement) – DataElement instance on which to walk and flip editable attr.
  • bool_val (bool) – Value to switch editable to.
  • skip_traits (None or list of str) – list of trait names to avoid flipping editable attr

kromatography.solve.simulation_runner module

Runner capable of preparing a list of simulations for CADET runs and executing these runs asynchronously using an SimpleAsyncJobManager.

This is designed to work well in multi-processing environments since it submits thinly wrapped pure CADET jobs on CADET input files to avoid pickling large python objects (simulations).

kromatography.solve.simulation_runner.run_simulations(simulation_list, job_manager=None, wait=True, timeout=None)[source]

Function to submit a list of simulations.

Can block the process and wait for all simulations to finish running.

Parameters:
  • simulation_list (list) – List of simulation instances to run.
  • job_manager (JobManager [OPTIONAL]) – JobManager instance to use to manage the CADET runs. If not provided, a new manager will be created.
  • wait (bool) – Should the process wait for all simulation to finish running?
  • timeout (int or None) – Number of seconds for the wait to timeout. Ignored if wait=False.
Returns:

Simulation runner created to run the provided simulations.

Return type:

SimulationRunner

class kromatography.solve.simulation_runner.SimulationRunner(**traits)[source]

Bases: traits.has_traits.HasStrictTraits

Submit a list of simulation for run to a job manager, and update them once CADET has run.

If no job manager is provided, a multi-processing one with be created, with cpu_count-1 max workers.

job_manager = Instance(JobManager)

JobManager responsible for submitting CADET jobs asynchronously

simulation_list = List

list of simulations to run

job_id = Str

ID of the job that contains running all simulations

work_item_ids = List(Str)

List of work items submitted, one for each simulation

max_workers = Int

Number of workers a job manager should be created with if not provided.

uuid = Str

A unique ID for the object

has_run = Property

Have all self.simulations run?

start()[source]
submit_jobs()[source]

Submit all simulations for CADET run.

wait(timeout=None)[source]

Wait until all simulations have been updated.

update_simulation_from_results(work_id)[source]

Trigger simulation update when a job manager work item has finished

kromatography.solve.slurm_cadet_executor module

Low level classes and functions to launch CADET binary on HDF5 input file.

class kromatography.solve.slurm_cadet_executor.SlurmCADETExecutor(**traits)[source]

Bases: kromatography.solve.cadet_executor.BaseCADETExecutor

WorkExecutor to manage executing CADET using a SLURM scheduler.

slurm_binary = Str

SLURM executable to use to submit jobs

slurm_squeue_binary = Str('squeue')

SLURM executable to use to inquire about jobs

slurm_script_fname = Str

Name of the slurm script to (over)write to submit jobs

slurm_scratch_dir = Str

Folder to create SLURM files

call_executable(input_file)[source]
class kromatography.solve.slurm_cadet_executor.SlurmBatchJobParams[source]

Bases: traits.has_traits.HasStrictTraits

Batch SLURM job execution parameters.

partition = Str

Name of the SLURM partition to submit the job to

nodes = Int(1)

Number of cluster nodes to allocate for the job

ntasks = Int(1)

Max number of tasks

cpus_per_task = Int

Number of CPUs to allocate for the task

time = Str('00:30:00')

Max amount of time for a single job

job_name = Str

Name of the CADET run job

mem = Int(1)

Max amount of memory the task can use, in GB

output_pattern = Str(SLURM_LOG_PATTERN)

Pattern of the output file SLURM will write STDOUT to

solver_prefs = Instance(SolverPreferenceGroup)

Solver related preferences

kromatography.solve.slurm_cadet_executor.extract_slurm_run_id(launch_output)[source]

Parse the run id out of the sbatch output.

kromatography.solve.slurm_cadet_executor.wait_on_slurm_job(job_id, squeue_executable='squeue')[source]

Wait until a SLURM job has finished.

This is done by checking the output of squeue for the specified job. As long as the job is running, it is in the queue. Once it has finished, the output from the squeue command is only the headers of the table.

kromatography.solve.slurm_cadet_executor.generate_slurm_script(slurm_file, input_file, solver_executable='cadet-cs', scratch_dir='.', job_params=None)[source]

Generate slurm script to execute the cadet executable on a CADET file.

Parameters:
  • slurm_file (str) – Path to the SLURM script to generate.
  • input_file (str) – Path to the solver input file.
  • solver_executable (str) – Path to the executable to launch the solver.
  • scratch_dir (str) – Path to the scratch directory to store SLURM script files and output files.
  • job_params (SlurmBatchJobParams) – SLURM job parameters.
kromatography.solve.slurm_cadet_executor.check_slurm_installed(executable=None)[source]

Check that slurm is installed and that batch jobs can be submitted.

kromatography.solve.work_executor module

exception kromatography.solve.work_executor.ExecutionFailedError[source]

Bases: exceptions.Exception

Thrown if a WorkExecutor encountered an error during execution.

exception kromatography.solve.work_executor.InvalidExecutorError[source]

Bases: exceptions.RuntimeError

Thrown if a WorkExecutor could not be initialized.

class kromatography.solve.work_executor.WorkExecutor[source]

Bases: traits.has_traits.ABCHasStrictTraits

Interface for a work item executor.

execute(work_id, *args, **kwargs)[source]

Execute the work item corresponding to work_id.

This method can be called multiple times with different arguments.