task_manager

task_manager

Author: Corentin Ravoux

Description : Task manager developed to launch several Tomography jobs. Tested on Cobalt and Irene clusters.

Machine

Machine(ending_str, error_str, wait_check)

Bases: object

Base class for a compute backend that launches solver jobs.

A machine knows how to write a launcher script, submit it, and detect job completion/errors from the solver log files. Concrete subclasses are :class:Nersc, :class:Irene (SLURM/MSUB clusters) and :class:Bash (local execution).

Attributes:

Name	Type	Description
ending_str	str	Log line marking a successful job end.
error_str	list[str]	Tokens whose presence in a log flags an error.
wait_check	bool	Whether the manager should poll for completion.

Initialise the machine.

Parameters:

Name	Type	Description	Default
ending_str	str	Log line marking successful completion.	required
error_str	list[str]	First-token markers indicating an error.	required
wait_check	bool	Whether to wait/poll for job completion.	required

Source code in lelantos/task_manager.py

def __init__(self, ending_str, error_str, wait_check):
    """Initialise the machine.

    Args:
        ending_str (str): Log line marking successful completion.
        error_str (list[str]): First-token markers indicating an error.
        wait_check (bool): Whether to wait/poll for job completion.
    """
    self.ending_str = ending_str
    self.error_str = error_str
    self.wait_check = wait_check

is_finished

is_finished(f)

Return True if the log lines contain the completion marker.

Parameters:

Name	Type	Description	Default
f	list[str]	Lines of a solver output/log file.	required

Returns:

Name	Type	Description
bool		True if ending_str is found.

Source code in lelantos/task_manager.py

def is_finished(self, f):
    """Return True if the log lines contain the completion marker.

    Args:
        f (list[str]): Lines of a solver output/log file.

    Returns:
        bool: True if ``ending_str`` is found.
    """
    for i in range(len(f)):
        if f[i].strip() == self.ending_str:
            return True
    return False

gives_error

gives_error(f)

Return True if the log lines contain an error marker.

Parameters:

Name	Type	Description	Default
f	list[str]	Lines of a solver error/log file.	required

Returns:

Name	Type	Description
bool		True if the first token of any line matches error_str.

Source code in lelantos/task_manager.py

def gives_error(self, f):
    """Return True if the log lines contain an error marker.

    Args:
        f (list[str]): Lines of a solver error/log file.

    Returns:
        bool: True if the first token of any line matches ``error_str``.
    """
    for i in range(len(f)):
        for j in range(len(self.error_str)):
            if f[i].strip().split()[0] == self.error_str[j]:
                return True
    return False

load_cluster_optional_arguments

load_cluster_optional_arguments()

Read the (N, n, c) SLURM resource counts from kwargs.

Returns:

Type	Description
	tuple[int, int, int]: (N, n, c) = number of nodes, tasks and
	cpus-per-task, each defaulting to 1.

Source code in lelantos/task_manager.py

def load_cluster_optional_arguments(self):
    """Read the (N, n, c) SLURM resource counts from ``kwargs``.

    Returns:
        tuple[int, int, int]: ``(N, n, c)`` = number of nodes, tasks and
        cpus-per-task, each defaulting to 1.
    """
    N = utils.return_key(self.kwargs, "N", 1)
    n = utils.return_key(self.kwargs, "n", 1)
    c = utils.return_key(self.kwargs, "c", 1)
    return (N, n, c)

Nersc

Nersc(**kwargs)

Bases: Machine

SLURM (sbatch) launcher for the NERSC cluster.

Supports two launch modes selected by the mode kwarg: "separated" (one sbatch job per sub-map) or "unified" (a single job running all sub-maps with srun ... & + wait).

Initialise the NERSC machine.

Parameters:

Name	Type	Description	Default
**kwargs		Optional SLURM parameters (mode, queue, partition, time, N, n, c).	{}

Source code in lelantos/task_manager.py

def __init__(self, **kwargs):
    """Initialise the NERSC machine.

    Args:
        **kwargs: Optional SLURM parameters (``mode``, ``queue``,
            ``partition``, ``time``, ``N``, ``n``, ``c``).
    """
    ending_str = "Execution Sum Up"
    error_str = ["srun:"]
    wait_check = True
    super(Nersc, self).__init__(ending_str, error_str, wait_check)

    self.project_name = "desi"
    self.launcher_name = "Tomography_start.sl"
    self.out_file_name = "{}.out"
    self.error_file_name = "{}.err"
    self.mem_per_cpu = 1952
    self.launch_mode = utils.return_key(kwargs, "mode", "separated")
    self.kwargs = kwargs

create_launcher

create_launcher(pwd, dir_paths, software_command_lines, software_name)

Write the SLURM launcher(s), dispatching on the launch mode.

Parameters:

Name	Type	Description	Default
pwd	str	Working directory (used for the unified launcher).	required
dir_paths	list[str]	Per-sub-map run directories.	required
software_command_lines	list[str]	Solver command per sub-map.	required
software_name	str	Job name / solver identifier.	required

Source code in lelantos/task_manager.py

def create_launcher(self, pwd, dir_paths, software_command_lines, software_name):
    """Write the SLURM launcher(s), dispatching on the launch mode.

    Args:
        pwd (str): Working directory (used for the unified launcher).
        dir_paths (list[str]): Per-sub-map run directories.
        software_command_lines (list[str]): Solver command per sub-map.
        software_name (str): Job name / solver identifier.
    """
    if self.launch_mode == "separated":
        self.create_launcher_separated(
            dir_paths, software_command_lines, software_name
        )
    elif self.launch_mode == "unified":
        self.create_launcher_unified(
            pwd, dir_paths, software_command_lines, software_name
        )

create_launcher_separated

create_launcher_separated(dir_paths, software_command_lines, software_name)

Write one SLURM batch script per sub-map directory.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	required
software_command_lines	list[str]	Solver command per sub-map.	required
software_name	str	Job name / solver identifier.	required

Source code in lelantos/task_manager.py

def create_launcher_separated(
    self, dir_paths, software_command_lines, software_name
):
    """Write one SLURM batch script per sub-map directory.

    Args:
        dir_paths (list[str]): Per-sub-map run directories.
        software_command_lines (list[str]): Solver command per sub-map.
        software_name (str): Job name / solver identifier.
    """
    queue = utils.return_key(self.kwargs, "queue", "regular")
    partition = utils.return_key(self.kwargs, "partition", "knl")
    (N, n, c) = self.load_cluster_optional_arguments()
    time = utils.return_key(self.kwargs, "time", "06:00:00")
    for i in range(len(dir_paths)):
        f = open(os.path.join(dir_paths[i], self.launcher_name), "w")
        f.write("#!/bin/bash -l\n")
        f.write(f"#SBATCH -N {N}" + "\n")
        f.write(f"#SBATCH -n {n}" + "\n")
        f.write(f"#SBATCH -c {c}" + "\n")
        f.write(f"#SBATCH -C {partition}" + "\n")
        f.write(f"#SBATCH -q {queue}" + "\n")
        f.write(f"#SBATCH -J {software_name}" + "\n")
        f.write(f"#SBATCH -t {time}" + "\n")
        f.write("#SBATCH -L project \n")
        f.write(f"#SBATCH -A {self.project_name}" + "\n")
        f.write(
            f"#SBATCH -o {os.path.join(dir_paths[i],self.out_file_name.format(software_name))}"
            + "\n"
        )
        f.write(
            f"#SBATCH -e {os.path.join(dir_paths[i],self.error_file_name.format(software_name))}"
            + "\n"
        )
        f.write("\n")
        f.write(f"export OMP_NUM_THREADS={c}" + "\n")
        f.write("\n")
        f.write(f"""srun {software_command_lines[i]}""" + " \n")

create_launcher_unified

create_launcher_unified(pwd, dir_paths, software_command_lines, software_name)

Write a single SLURM script running all sub-maps concurrently.

Parameters:

Name	Type	Description	Default
pwd	str	Directory where the unified launcher is written.	required
dir_paths	list[str]	Per-sub-map run directories.	required
software_command_lines	list[str]	Solver command per sub-map.	required
software_name	str	Job name / solver identifier.	required

Source code in lelantos/task_manager.py

def create_launcher_unified(
    self, pwd, dir_paths, software_command_lines, software_name
):
    """Write a single SLURM script running all sub-maps concurrently.

    Args:
        pwd (str): Directory where the unified launcher is written.
        dir_paths (list[str]): Per-sub-map run directories.
        software_command_lines (list[str]): Solver command per sub-map.
        software_name (str): Job name / solver identifier.
    """
    queue = utils.return_key(self.kwargs, "queue", "regular")
    partition = utils.return_key(self.kwargs, "partition", "knl")
    (N, n, c) = self.load_cluster_optional_arguments()
    time = utils.return_key(self.kwargs, "time", "06:00:00")
    f = open(os.path.join(pwd, self.launcher_name), "w")
    f.write("#!/bin/bash -l\n")
    f.write(f"#SBATCH -N {N}" + "\n")
    f.write(f"#SBATCH -C {partition}" + "\n")
    f.write(f"#SBATCH -q {queue}" + "\n")
    f.write(f"#SBATCH -J {software_name}" + "\n")
    f.write(f"#SBATCH -t {time}" + "\n")
    f.write("#SBATCH -L project \n")
    f.write(f"#SBATCH -A {self.project_name}" + "\n")
    f.write(f"#SBATCH -o {os.path.join(pwd,self.launcher_name)}.out" + "\n")
    f.write(f"#SBATCH -e {os.path.join(pwd,self.launcher_name)}.err" + "\n")
    f.write("\n")
    for i in range(len(dir_paths)):
        f.write(
            f"""srun -n {n} -c {c} --mem {int(c*self.mem_per_cpu)}"""
            + f""" -o {os.path.join(dir_paths[i],self.out_file_name.format(software_name))}"""
            + f""" -e {os.path.join(dir_paths[i],self.error_file_name.format(software_name))}"""
            + f""" {software_command_lines[i]}"""
            + " & \n \n"
        )
    f.write("wait")

launch_unified

launch_unified(pwd)

Submit the single unified SLURM launcher via sbatch.

Parameters:

Name	Type	Description	Default
pwd	str	Directory containing the unified launcher.	required

Source code in lelantos/task_manager.py

def launch_unified(self, pwd):
    """Submit the single unified SLURM launcher via ``sbatch``.

    Args:
        pwd (str): Directory containing the unified launcher.
    """
    call(["sbatch", os.path.join(pwd, self.launcher_name)])

launch_separated

launch_separated(dir_paths)

Submit one SLURM job per sub-map directory via sbatch.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	required

Source code in lelantos/task_manager.py

def launch_separated(self, dir_paths):
    """Submit one SLURM job per sub-map directory via ``sbatch``.

    Args:
        dir_paths (list[str]): Per-sub-map run directories.
    """
    for i in range(len(dir_paths)):
        call(["sbatch", os.path.join(dir_paths[i], self.launcher_name)])

launch

launch(dir_paths=None, pwd=None, software_command_lines=None, software_name=None)

Submit the NERSC job(s), dispatching on the launch mode.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	None
pwd	str	Working directory (unified mode).	None
software_command_lines	list[str]	Unused here.	None
software_name	str	Unused here.	None

Source code in lelantos/task_manager.py

def launch(
    self, dir_paths=None, pwd=None, software_command_lines=None, software_name=None
):
    """Submit the NERSC job(s), dispatching on the launch mode.

    Args:
        dir_paths (list[str], optional): Per-sub-map run directories.
        pwd (str, optional): Working directory (unified mode).
        software_command_lines (list[str], optional): Unused here.
        software_name (str, optional): Unused here.
    """
    if self.launch_mode == "separated":
        self.launch_separated(dir_paths)
    elif self.launch_mode == "unified":
        self.launch_unified(pwd)

Irene

Irene(**kwargs)

Bases: Machine

MSUB (ccc_msub) launcher for the TGCC Irene cluster.

Initialise the Irene machine.

Parameters:

Name	Type	Description	Default
**kwargs		Optional scheduler parameters (partition, time, N, n, c).	{}

Source code in lelantos/task_manager.py

def __init__(self, **kwargs):
    """Initialise the Irene machine.

    Args:
        **kwargs: Optional scheduler parameters (``partition``, ``time``,
            ``N``, ``n``, ``c``).
    """
    ending_str = ""
    error_str = ["srun:"]
    wait_check = True
    super(Irene, self).__init__(ending_str, error_str, wait_check)

    self.project_name = "gen12028"
    self.launcher_name = "Tomography_start.sl"
    self.out_file_name = "{}.out"
    self.error_file_name = "{}.err"
    self.kwargs = kwargs

create_launcher

create_launcher(pwd, dir_paths, software_command_lines, software_name)

Write one MSUB batch script per sub-map directory.

Parameters:

Name	Type	Description	Default
pwd	str	Unused (kept for interface parity with other machines).	required
dir_paths	list[str]	Per-sub-map run directories.	required
software_command_lines	list[str]	Solver command per sub-map.	required
software_name	str	Job name / solver identifier.	required

Source code in lelantos/task_manager.py

def create_launcher(self, pwd, dir_paths, software_command_lines, software_name):
    """Write one MSUB batch script per sub-map directory.

    Args:
        pwd (str): Unused (kept for interface parity with other machines).
        dir_paths (list[str]): Per-sub-map run directories.
        software_command_lines (list[str]): Solver command per sub-map.
        software_name (str): Job name / solver identifier.
    """
    (N, n, c) = self.load_cluster_optional_arguments()
    partition = utils.return_key(self.kwargs, "partition", "rome")
    time = utils.return_key(self.kwargs, "time", "60000")
    for i in range(len(dir_paths)):
        f = open(os.path.join(dir_paths[i], self.launcher_name), "w")
        f.write("#!/bin/bash -l\n")
        f.write("\n")
        f.write(f"#MSUB -r {software_name}" + "\n")
        f.write(f"#MSUB -T {time}" + "\n")
        f.write(f"#MSUB -q {partition}" + "\n")
        f.write(
            f"#MSUB -o {os.path.join(dir_paths[i],self.out_file_name.format(software_name))}"
            + "\n"
        )
        f.write(
            f"#MSUB -e {os.path.join(dir_paths[i],self.error_file_name.format(software_name))}"
            + "\n"
        )
        f.write("#MSUB -m scratch,work  \n")
        f.write(f"#MSUB -N {N}" + "\n")
        f.write(f"#MSUB -n {n}" + "\n")
        f.write(f"#MSUB -c {c}" + "\n")
        f.write(f"#MSUB -A {self.project_name}" + "\n")
        f.write("\n")
        f.write("export OMP_NUM_THREADS=2\n")
        f.write("\n")
        f.write(f"ccc_mprun {software_command_lines[i]}" + "\n")
        f.close()

launch

launch(dir_paths=None, pwd=None, software_command_lines=None, software_name=None)

Submit one MSUB job per sub-map directory via ccc_msub.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	None
pwd	str	Unused.	None
software_command_lines	list[str]	Unused.	None
software_name	str	Unused.	None

Source code in lelantos/task_manager.py

def launch(
    self, dir_paths=None, pwd=None, software_command_lines=None, software_name=None
):
    """Submit one MSUB job per sub-map directory via ``ccc_msub``.

    Args:
        dir_paths (list[str], optional): Per-sub-map run directories.
        pwd (str, optional): Unused.
        software_command_lines (list[str], optional): Unused.
        software_name (str, optional): Unused.
    """
    for i in range(len(dir_paths)):
        call(["ccc_msub", os.path.join(dir_paths[i], self.launcher_name)])

Bash

Bash(ending_str='', error_str=[], **kwargs)

Bases: Machine

Local (no scheduler) launcher running the solver via subprocess.

Runs jobs serially, or in parallel with a :mod:multiprocessing pool when the n kwarg is greater than 1.

Initialise the Bash machine.

Parameters:

Name	Type	Description	Default
ending_str	str	Completion marker (usually taken from the solver).	''
error_str	list[str]	Error markers.	[]
**kwargs		Optional parameters; n sets the number of processes.	{}

Source code in lelantos/task_manager.py

def __init__(self, ending_str="", error_str=[], **kwargs):
    """Initialise the Bash machine.

    Args:
        ending_str (str): Completion marker (usually taken from the solver).
        error_str (list[str]): Error markers.
        **kwargs: Optional parameters; ``n`` sets the number of processes.
    """
    wait_check = False
    super(Bash, self).__init__(ending_str, error_str, wait_check)

    self.out_file_name = "{}.out"
    self.error_file_name = "{}.err"
    self.kwargs = kwargs

create_launcher

create_launcher(pwd, dir_path, command_line, software_name)

No-op: local execution needs no launcher script.

Returns:

Name	Type	Description
tuple		An empty tuple.

Source code in lelantos/task_manager.py

def create_launcher(self, pwd, dir_path, command_line, software_name):
    """No-op: local execution needs no launcher script.

    Returns:
        tuple: An empty tuple.
    """
    return ()

launch

launch(dir_paths=None, pwd=None, software_command_lines=None, software_name=None)

Run the solver locally, serially or in parallel.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	None
pwd	str	Unused.	None
software_command_lines	list[str]	Solver command per sub-map.	None
software_name	str	Solver identifier (log file names).	None

Source code in lelantos/task_manager.py

def launch(
    self, dir_paths=None, pwd=None, software_command_lines=None, software_name=None
):
    """Run the solver locally, serially or in parallel.

    Args:
        dir_paths (list[str], optional): Per-sub-map run directories.
        pwd (str, optional): Unused.
        software_command_lines (list[str], optional): Solver command per
            sub-map.
        software_name (str, optional): Solver identifier (log file names).
    """
    number_process = utils.return_key(self.kwargs, "n", 1)
    if number_process == 1:
        self.launch_serial(
            dir_paths=dir_paths,
            pwd=pwd,
            software_command_lines=software_command_lines,
            software_name=software_name,
        )
    else:
        self.launch_parallel(
            number_process,
            dir_paths=dir_paths,
            pwd=pwd,
            software_command_lines=software_command_lines,
            software_name=software_name,
        )

launch_parallel

launch_parallel(number_process, dir_paths=None, pwd=None, software_command_lines=None, software_name=None)

Run the sub-map solver jobs concurrently in a process pool.

Parameters:

Name	Type	Description	Default
number_process	int	Pool size.	required
dir_paths	list[str]	Per-sub-map run directories.	None
pwd	str	Unused.	None
software_command_lines	list[str]	Solver command per sub-map.	None
software_name	str	Solver identifier (log file names).	None

Source code in lelantos/task_manager.py

def launch_parallel(
    self,
    number_process,
    dir_paths=None,
    pwd=None,
    software_command_lines=None,
    software_name=None,
):
    """Run the sub-map solver jobs concurrently in a process pool.

    Args:
        number_process (int): Pool size.
        dir_paths (list[str], optional): Per-sub-map run directories.
        pwd (str, optional): Unused.
        software_command_lines (list[str], optional): Solver command per
            sub-map.
        software_name (str, optional): Solver identifier (log file names).
    """
    list_launch = [
        [dir_paths[i], software_command_lines[i]] for i in range(len(dir_paths))
    ]
    func = partial(self.launch_single, software_name)
    with mp.Pool(number_process) as pool:
        pool.map(func, list_launch)

launch_single

launch_single(software_name, param_launch)

Run one solver command, redirecting stdout/stderr to log files.

Parameters:

Name	Type	Description	Default
software_name	str	Solver identifier (used for log file names).	required
param_launch	list	[run_directory, command_line] pair.	required

Source code in lelantos/task_manager.py

def launch_single(self, software_name, param_launch):
    """Run one solver command, redirecting stdout/stderr to log files.

    Args:
        software_name (str): Solver identifier (used for log file names).
        param_launch (list): ``[run_directory, command_line]`` pair.
    """
    dir_paths, software_command_lines = param_launch[0], param_launch[1]
    out_name = os.path.join(dir_paths, self.out_file_name.format(software_name))
    err_name = os.path.join(dir_paths, self.error_file_name.format(software_name))
    print(f"launch of {software_command_lines}")
    call(
        software_command_lines.split(),
        stdout=open(out_name, "w"),
        stderr=open(err_name, "w"),
    )
    print(f"end of {software_command_lines}")

launch_serial

launch_serial(dir_paths=None, pwd=None, software_command_lines=None, software_name=None)

Run the sub-map solver jobs one after another.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	None
pwd	str	Unused.	None
software_command_lines	list[str]	Solver command per sub-map.	None
software_name	str	Solver identifier (log file names).	None

Source code in lelantos/task_manager.py

def launch_serial(
    self, dir_paths=None, pwd=None, software_command_lines=None, software_name=None
):
    """Run the sub-map solver jobs one after another.

    Args:
        dir_paths (list[str], optional): Per-sub-map run directories.
        pwd (str, optional): Unused.
        software_command_lines (list[str], optional): Solver command per
            sub-map.
        software_name (str, optional): Solver identifier (log file names).
    """
    for i in range(len(dir_paths)):
        out_name = os.path.join(
            dir_paths[i], self.out_file_name.format(software_name)
        )
        err_name = os.path.join(
            dir_paths[i], self.error_file_name.format(software_name)
        )
        print(f"launch of {software_command_lines[i]}")
        call(
            software_command_lines[i].split(),
            stdout=open(out_name, "w"),
            stderr=open(err_name, "w"),
        )

TomographySoftware

TomographySoftware(exec_file)

Bases: object

Base class for a tomographic solver wrapped by the manager.

Attributes:

Name	Type	Description
exec_file	str	Path to the solver executable.

Initialise the solver wrapper.

Parameters:

Name	Type	Description	Default
exec_file	str	Path to the solver executable.	required

Source code in lelantos/task_manager.py

def __init__(self, exec_file):
    """Initialise the solver wrapper.

    Args:
        exec_file (str): Path to the solver executable.
    """
    self.exec_file = exec_file

Other

Other()

Bases: TomographySoftware

Placeholder solver backend (other.exe) for a custom algorithm.

Locate the other.exe executable in the package exec folder.

Source code in lelantos/task_manager.py

def __init__(self):
    """Locate the ``other.exe`` executable in the package ``exec`` folder."""
    source_path = os.path.dirname(os.path.realpath(__file__))
    name_exec = "other.exe"
    exec_file = os.path.join(source_path, "exec", name_exec)

    super(Other, self).__init__(exec_file)
    self.command_line = self.exec_file + " {}"

create_input

create_input(dir_path, launcher_params, name)

No-op input writer for the placeholder solver.

Returns:

Name	Type	Description
tuple		An empty tuple.

Source code in lelantos/task_manager.py

def create_input(self, dir_path, launcher_params, name):
    """No-op input writer for the placeholder solver.

    Returns:
        tuple: An empty tuple.
    """
    return ()

Dachshund

Dachshund()

Bases: TomographySoftware

Wrapper for the Dachshund Wiener-filter tomography solver.

Locate the dachshund.exe executable in the package exec folder.

Source code in lelantos/task_manager.py

def __init__(self):
    """Locate the ``dachshund.exe`` executable in the package ``exec`` folder."""
    source_path = os.path.dirname(os.path.realpath(__file__))
    name_exec = "dachshund.exe"
    exec_file = os.path.join(source_path, "exec", name_exec)

    super(Dachshund, self).__init__(exec_file)
    self.command_line = self.exec_file + " {}"
    self.ending_str = "Total time"

create_input

create_input(dir_paths, launcher_params, launcher_names)

Write the Dachshund .cfg input file for each sub-map.

Each file declares the box size, pixel count, map grid, signal covariance (sigma_f, l_perp, l_par), PCG solver settings and the input/output binary paths.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	required
launcher_params	list[dict]	Per-sub-map geometry/solver params (keys lx, ly, lz, npix, nx, ny, nz, sigmaf, lperp, lpar, namepixel, namemap).	required
launcher_names	list[str]	Config file name for each sub-map.	required

Source code in lelantos/task_manager.py

def create_input(self, dir_paths, launcher_params, launcher_names):
    """Write the Dachshund ``.cfg`` input file for each sub-map.

    Each file declares the box size, pixel count, map grid, signal
    covariance (sigma_f, l_perp, l_par), PCG solver settings and the
    input/output binary paths.

    Args:
        dir_paths (list[str]): Per-sub-map run directories.
        launcher_params (list[dict]): Per-sub-map geometry/solver params
            (keys ``lx, ly, lz, npix, nx, ny, nz, sigmaf, lperp, lpar,
            namepixel, namemap``).
        launcher_names (list[str]): Config file name for each sub-map.
    """
    for i in range(len(dir_paths)):
        name = os.path.join(dir_paths[i], launcher_names[i])
        (lx, ly, lz, npix, nx, ny, nz, sigmaf, lperp, lpar, namepixel, namemap) = (
            launcher_params[i]["lx"],
            launcher_params[i]["ly"],
            launcher_params[i]["lz"],
            launcher_params[i]["npix"],
            launcher_params[i]["nx"],
            launcher_params[i]["ny"],
            launcher_params[i]["nz"],
            launcher_params[i]["sigmaf"],
            launcher_params[i]["lperp"],
            launcher_params[i]["lpar"],
            launcher_params[i]["namepixel"],
            launcher_params[i]["namemap"],
        )
        f = open(name, "w")
        f.write("#lx, ly, lz: the domain size in each direction.\n")
        f.write("#num_pixels: the *total* number of pixels.\n")
        f.write(
            "#map_nx, map_ny, map_nz: the number of map points. The map points are arbitrary but for now these n's are used to setup a uniform grid across the domain given above.\n"
        )
        f.write("#corr_var_s: the signal cov prefactor sigma_f^2\n")
        f.write("#corr_l_perp: the signal cov perp scale.\n")
        f.write("#corr_l_para: the signal cov para scale.\n")
        f.write(
            "#pcg_max_iter: the PCG max number of iterations. 100 should be good.\n"
        )
        f.write(
            "#pcg_tol: the PCG stopping tolerance. I found 1.0e-3 is good enough. Set it very small if you want the most accurate map.\n"
        )
        f.write("lx = {}\n".format(lx))
        f.write("ly = {}\n".format(ly))
        f.write("lz = {}\n".format(lz))
        f.write("\n")
        f.write("# From output of GEN_DACH_INPUT.PRO\n")
        f.write("num_pixels = {}\n".format(npix))
        f.write("\n")
        f.write("map_nx = {}\n".format(nx))
        f.write("map_ny = {}\n".format(ny))
        f.write("map_nz = {}\n".format(nz))
        f.write("\n")
        f.write("corr_var_s = {}\n".format(sigmaf))
        f.write("corr_l_perp = {}\n".format(lperp))
        f.write("corr_l_para = {}\n".format(lpar))
        f.write("\n")
        f.write("pcg_max_iter = 500\n")
        f.write("pcg_tol = 1.0e-3\n")
        f.write("#pcg_step_r = 1\n")
        f.write("\n")
        f.write("option_map_covar = 0\n")
        f.write("option_noise_covar = 0\n")
        f.write(
            "pixel_data_path = {}\n".format(os.path.join(dir_paths[i], namepixel))
        )
        f.write("map_path = {}\n".format(os.path.join(dir_paths[i], namemap)))
        f.close()

TomographyManager

TomographyManager(pwd, software, machine, name_pixel, launch_file, symlink_folder=None, **kwargs)

Bases: object

Orchestrate tomographic solver runs on a chosen machine.

Loads the launch description (a pickle listing sub-map names and their solver parameters), sets up temporary run directories, writes the solver inputs and machine launchers, submits the jobs, waits for completion, checks for errors and copies the resulting maps back.

Attributes:

Name	Type	Description
available_software	tuple[str]	Supported solver names.
available_machine	tuple[str]	Supported machine names.

Initialise the manager and its machine/software backends.

Parameters:

Name	Type	Description	Default
pwd	str	Run/output directory for the tomography.	required
software	str	Solver name ("dachshund" or "other").	required
machine	str	Machine name ("irene", "nersc" or "bash").	required
name_pixel	str	Path to the input pixel binary.	required
launch_file	str	Pickle listing sub-map names and solver params.	required
symlink_folder	str	If set, run inside a symlink to this folder (e.g. cluster scratch).	None
**kwargs		Machine-specific options (e.g. n for bash processes).	{}

Source code in lelantos/task_manager.py

def __init__(
    self,
    pwd,
    software,
    machine,
    name_pixel,
    launch_file,
    symlink_folder=None,
    **kwargs,
):
    """Initialise the manager and its machine/software backends.

    Args:
        pwd (str): Run/output directory for the tomography.
        software (str): Solver name (``"dachshund"`` or ``"other"``).
        machine (str): Machine name (``"irene"``, ``"nersc"`` or ``"bash"``).
        name_pixel (str): Path to the input pixel binary.
        launch_file (str): Pickle listing sub-map names and solver params.
        symlink_folder (str, optional): If set, run inside a symlink to this
            folder (e.g. cluster scratch).
        **kwargs: Machine-specific options (e.g. ``n`` for bash processes).
    """
    self.pwd = pwd
    if symlink_folder is not None:
        self.link_tomography_folder(symlink_folder)

    self.name_pixel = name_pixel
    self.launch_file = launch_file

    self.log = utils.create_report_log(name=os.path.join(self.pwd, "Python_Report"))
    self.software = self.init_sofware(software)
    self.machine = self.init_machine(machine, **kwargs)
    if (self.machine.ending_str == "") & (self.software.ending_str is not None):
        self.machine.ending_str = self.software.ending_str

init_sofware

init_sofware(software)

Instantiate the solver backend selected by name.

Parameters:

Name	Type	Description	Default
software	str	Solver name (case-insensitive).	required

Returns:

Name	Type	Description
TomographySoftware		The solver wrapper, or a KeyError instance
		if the name is unknown.

Source code in lelantos/task_manager.py

def init_sofware(self, software):
    """Instantiate the solver backend selected by name.

    Args:
        software (str): Solver name (case-insensitive).

    Returns:
        TomographySoftware: The solver wrapper, or a ``KeyError`` instance
        if the name is unknown.
    """
    if software.lower() == "dachshund":
        return Dachshund()
    elif software.lower() == "other":
        return Other()
    else:
        return KeyError(
            f"The software {software} is not available, please choose in {TomographyManager.available_software}"
        )

init_machine

init_machine(machine, **kwargs)

Instantiate the machine backend selected by name.

Parameters:

Name	Type	Description	Default
machine	str	Machine name (case-insensitive).	required
**kwargs		Machine-specific options.	{}

Returns:

Name	Type	Description
Machine		The machine backend, or a KeyError instance if the
		name is unknown.

Source code in lelantos/task_manager.py

def init_machine(self, machine, **kwargs):
    """Instantiate the machine backend selected by name.

    Args:
        machine (str): Machine name (case-insensitive).
        **kwargs: Machine-specific options.

    Returns:
        Machine: The machine backend, or a ``KeyError`` instance if the
        name is unknown.
    """
    if machine.lower() == "irene":
        return Irene(**kwargs)
    if machine.lower() == "nersc":
        return Nersc(**kwargs)
    elif machine.lower() == "bash":
        return Bash(**kwargs)
    else:
        return KeyError(
            f"The machine {machine} is not available, please choose in {TomographyManager.available_machine}"
        )

create_python_dir staticmethod

create_python_dir(pwd)

Create the Tmp working sub-directory if absent.

Parameters:

Name	Type	Description	Default
pwd	str	Parent directory.	required

Source code in lelantos/task_manager.py

@staticmethod
def create_python_dir(pwd):
    """Create the ``Tmp`` working sub-directory if absent.

    Args:
        pwd (str): Parent directory.
    """
    if os.path.isdir(os.path.join(pwd, "Tmp")) == False:
        os.mkdir(os.path.join(pwd, "Tmp"))

create_dir staticmethod

create_dir(pwd, dirnames)

Create one Tmp/<name> run directory per sub-map.

Parameters:

Name	Type	Description	Default
pwd	str	Parent directory.	required
dirnames	list[str]	Sub-map directory names.	required

Source code in lelantos/task_manager.py

@staticmethod
def create_dir(pwd, dirnames):
    """Create one ``Tmp/<name>`` run directory per sub-map.

    Args:
        pwd (str): Parent directory.
        dirnames (list[str]): Sub-map directory names.
    """
    for i in range(len(dirnames)):
        dir_path = os.path.join(pwd, "Tmp", dirnames[i])
        if os.path.isdir(dir_path) == False:
            os.mkdir(dir_path)

create_file staticmethod

create_file(name)

Create a placeholder log file containing "wait" if absent.

Parameters:

Name	Type	Description	Default
name	str	File path.	required

Source code in lelantos/task_manager.py

@staticmethod
def create_file(name):
    """Create a placeholder log file containing ``"wait"`` if absent.

    Args:
        name (str): File path.
    """
    if os.path.isfile(name) == False:
        f = open(name, "w")
        f.write("wait")
        f.close()

is_finished

is_finished(file_lines)

Delegate completion detection to the machine backend.

Parameters:

Name	Type	Description	Default
file_lines	list[str]	Lines of a solver output file.	required

Returns:

Name	Type	Description
bool		True if the job completed.

Source code in lelantos/task_manager.py

def is_finished(self, file_lines):
    """Delegate completion detection to the machine backend.

    Args:
        file_lines (list[str]): Lines of a solver output file.

    Returns:
        bool: True if the job completed.
    """
    return self.machine.is_finished(file_lines)

gives_error

gives_error(file_lines)

Delegate error detection to the machine backend.

Parameters:

Name	Type	Description	Default
file_lines	list[str]	Lines of a solver error file.	required

Returns:

Name	Type	Description
bool		True if the log signals an error.

Source code in lelantos/task_manager.py

def gives_error(self, file_lines):
    """Delegate error detection to the machine backend.

    Args:
        file_lines (list[str]): Lines of a solver error file.

    Returns:
        bool: True if the log signals an error.
    """
    return self.machine.gives_error(file_lines)

wait_until_finished

wait_until_finished(pwd, listname)

Poll the sub-map output files until every job has completed.

Parameters:

Name	Type	Description	Default
pwd	str	Parent run directory.	required
listname	list[str]	Sub-map directory names to monitor.	required

Source code in lelantos/task_manager.py

def wait_until_finished(self, pwd, listname):
    """Poll the sub-map output files until every job has completed.

    Args:
        pwd (str): Parent run directory.
        listname (list[str]): Sub-map directory names to monitor.
    """
    list_finished = [False for i in range(len(listname))]
    out_file_name = self.machine.out_file_name.format(self.software.name)
    while self.all_finished(list_finished) == False:
        for i in range(len(listname)):
            out_file = os.path.join(pwd, "Tmp", listname[i], out_file_name)
            TomographyManager.create_file(out_file)
            file = open(out_file, "r")
            file_lines = file.readlines()
            file.close()
            list_finished[i] = self.is_finished(file_lines)
        time.sleep(10)
    self.log.add("All calculation are finished")

check_errors

check_errors(pwd, listname)

Scan each sub-map error file and log whether it reported an error.

Parameters:

Name	Type	Description	Default
pwd	str	Parent run directory.	required
listname	list[str]	Sub-map directory names to check.	required

Source code in lelantos/task_manager.py

def check_errors(self, pwd, listname):
    """Scan each sub-map error file and log whether it reported an error.

    Args:
        pwd (str): Parent run directory.
        listname (list[str]): Sub-map directory names to check.
    """
    list_error = [False for i in range(len(listname))]
    error_file_name = self.machine.error_file_name.format(self.software.name)
    self.log.add("List of errors :")
    self.log.add("")
    for i in range(len(listname)):
        error_file = os.path.join(pwd, "Tmp", listname[i], error_file_name)
        TomographyManager.create_file(error_file)
        file = open(error_file, "r")
        file_lines = file.readlines()
        file.close()
        list_error[i] = self.gives_error(file_lines)
        self.log.add(f"The file {listname[i]} gave an error : {list_error[i]}")

all_finished

all_finished(listFinished)

Return True only if every entry of the status list is True.

Parameters:

Name	Type	Description	Default
listFinished	list[bool]	Per-sub-map completion flags.	required

Returns:

Name	Type	Description
bool		True if all jobs are finished.

Source code in lelantos/task_manager.py

def all_finished(self, listFinished):
    """Return True only if every entry of the status list is True.

    Args:
        listFinished (list[bool]): Per-sub-map completion flags.

    Returns:
        bool: True if all jobs are finished.
    """
    allFinished = True
    for i in range(len(listFinished)):
        if listFinished[i] == False:
            allFinished = False
    return allFinished

copy_files

copy_files(dir_paths, listname)

Copy each sub-map's pixel input into its run directory.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	required
listname	list[str]	Sub-map names (pixel file suffixes).	required

Source code in lelantos/task_manager.py

def copy_files(self, dir_paths, listname):
    """Copy each sub-map's pixel input into its run directory.

    Args:
        dir_paths (list[str]): Per-sub-map run directories.
        listname (list[str]): Sub-map names (pixel file suffixes).
    """
    for i in range(len(listname)):
        call(
            [
                "cp",
                os.path.join(self.pwd, f"{self.name_pixel}_{listname[i]}"),
                dir_paths[i],
            ]
        )

create_machine_launcher

create_machine_launcher(dir_paths, launcher_params)

Write the solver inputs and the machine launcher scripts.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	required
launcher_params	list[dict]	Per-sub-map solver parameters.	required

Source code in lelantos/task_manager.py

def create_machine_launcher(self, dir_paths, launcher_params):
    """Write the solver inputs and the machine launcher scripts.

    Args:
        dir_paths (list[str]): Per-sub-map run directories.
        launcher_params (list[dict]): Per-sub-map solver parameters.
    """
    launcher_names = [
        launcher_params[i]["nameinput"] for i in range(len(launcher_params))
    ]
    command_lines = [
        self.software.command_line.format(
            os.path.join(dir_paths[i], launcher_names[i])
        )
        for i in range(len(dir_paths))
    ]
    self.software.create_input(dir_paths, launcher_params, launcher_names)
    self.machine.create_launcher(
        self.pwd, dir_paths, command_lines, self.software.name
    )

launch

launch(dir_paths, launcher_params)

Submit the solver jobs on the configured machine.

Parameters:

Name	Type	Description	Default
dir_paths	list[str]	Per-sub-map run directories.	required
launcher_params	list[dict]	Per-sub-map solver parameters.	required

Source code in lelantos/task_manager.py

def launch(self, dir_paths, launcher_params):
    """Submit the solver jobs on the configured machine.

    Args:
        dir_paths (list[str]): Per-sub-map run directories.
        launcher_params (list[dict]): Per-sub-map solver parameters.
    """
    launcher_names = [
        launcher_params[i]["nameinput"] for i in range(len(launcher_params))
    ]
    command_lines = [
        self.software.command_line.format(
            os.path.join(dir_paths[i], launcher_names[i])
        )
        for i in range(len(dir_paths))
    ]
    self.machine.launch(
        dir_paths=dir_paths,
        pwd=self.pwd,
        software_command_lines=command_lines,
        software_name=self.software.name,
    )

treat_launch

treat_launch(pwd, listname)

Optionally wait for completion, then check for errors.

Parameters:

Name	Type	Description	Default
pwd	str	Parent run directory.	required
listname	list[str]	Sub-map directory names.	required

Source code in lelantos/task_manager.py

def treat_launch(self, pwd, listname):
    """Optionally wait for completion, then check for errors.

    Args:
        pwd (str): Parent run directory.
        listname (list[str]): Sub-map directory names.
    """
    wait_check = self.machine.wait_check
    if wait_check:
        self.wait_until_finished(pwd, listname)
    self.check_errors(pwd, listname)

launch_all

launch_all()

Run the full launch sequence for all sub-maps.

Loads the launch pickle, creates the temporary run directories, copies inputs, writes launchers, submits the jobs, logs them and finally waits for completion / checks for errors.

Source code in lelantos/task_manager.py

def launch_all(self):
    """Run the full launch sequence for all sub-maps.

    Loads the launch pickle, creates the temporary run directories, copies
    inputs, writes launchers, submits the jobs, logs them and finally waits
    for completion / checks for errors.
    """
    launching_file = pickle.load(open(self.launch_file, "rb"))
    listname, launcher_params = launching_file[0], launching_file[1]
    TomographyManager.create_python_dir(self.pwd)
    TomographyManager.create_dir(self.pwd, listname)
    dir_paths = [
        os.path.join(self.pwd, "Tmp", listname[i]) for i in range(len(listname))
    ]
    self.copy_files(dir_paths, listname)
    self.create_machine_launcher(dir_paths, launcher_params)
    self.launch(dir_paths, launcher_params)
    for i in range(len(listname)):
        self.log.add("Launch of the input " + str(launcher_params[i]["nameinput"]))
    time.sleep(5)
    self.treat_launch(self.pwd, listname)

copy

copy()

Copy each sub-map's output map from Tmp back to pwd.

Source code in lelantos/task_manager.py

def copy(self):
    """Copy each sub-map's output map from ``Tmp`` back to ``pwd``."""
    launching_file = pickle.load(open(self.launch_file, "rb"))
    listname, launcher_params = launching_file[0], launching_file[1]
    for i in range(len(listname)):
        call(
            [
                "cp",
                self.pwd
                + "/Tmp/"
                + listname[i]
                + "/"
                + launcher_params[i]["namemap"],
                self.pwd,
            ]
        )

remove_tmp

remove_tmp()

Delete the temporary Tmp run directory tree.

Source code in lelantos/task_manager.py

def remove_tmp(self):
    """Delete the temporary ``Tmp`` run directory tree."""
    shutil.rmtree(os.path.join(self.pwd, "Tmp"), ignore_errors=True)

link_tomography_folder

link_tomography_folder(symlink_folder)

Replace pwd with a symlink to an external folder.

Parameters:

Name	Type	Description	Default
symlink_folder	str	Target directory (created if missing); the run directory pwd becomes a symlink to it.	required

Source code in lelantos/task_manager.py

def link_tomography_folder(self, symlink_folder):
    """Replace ``pwd`` with a symlink to an external folder.

    Args:
        symlink_folder (str): Target directory (created if missing); the
            run directory ``pwd`` becomes a symlink to it.
    """
    os.makedirs(symlink_folder, exist_ok=True)
    if os.path.isdir(self.pwd):
        try:
            os.remove(self.pwd)
        except:
            shutil.rmtree(self.pwd)
    os.symlink(symlink_folder, self.pwd, target_is_directory=True)