Skip to content

Python API

All public packages, functions and classes are available in this module.

Functions:

Data classes:

  • [ImasHandle][duqtools.api.ImasHandle]
  • Job
  • Run
  • Runs

Plotting:

Job(path, *, cfg)

Parameters:

  • path (Path) –

    Directory for simulation or model run.

  • cfg (Optional[Config]) –

    Duqtools config, defaults to global config if unspecified.

Source code in duqtools/models/_job.py 
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(self, path: Path, *, cfg: Config):
    """This class handles job status and submission.

    Parameters
    ----------
    path : Path
        Directory for simulation or model run.
    cfg : Optional[Config], optional
        Duqtools config, defaults to global config if unspecified.
    """
    self.path = Path(path).resolve()
    self.cfg = cfg

has_status: bool property

Return true if a status file exists.

has_submit_script: bool property

Return true if directory has submit script.

in_file: Path property

Return path to the input file for the job.

is_completed: bool property

Return true if the job has been completed succesfully.

is_done: bool property

Return true if the job is done (completed or failed).

is_failed: bool property

Return true if the job has failed.

is_running: bool property

Return true if the job is running.

is_submitted: bool property

Return true if the job has been submitted.

lockfile: Path property

Return the path of the lockfile.

out_file: Path property

Return path to the output file for the job.

status_file: Path property

Return the path of the status file.

status_symbol property

One letter status symbol.

submit_script: Path property

Return the path of the submit script.

start()

Submit job and return generate that raises StopIteration when done.

Source code in duqtools/models/_job.py 
165
166
167
168
169
170
171
172
def start(self):
    """Submit job and return generate that raises StopIteration when
    done."""
    click.echo(f'Submitting {self}\033[K')
    self.submit()

    while self.status() in (JobStatus.RUNNING, JobStatus.NOSTATUS):
        yield

status()

Return the status of the job.

Source code in duqtools/models/_job.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def status(self) -> str:
    """Return the status of the job."""
    if not self.has_status:
        return JobStatus.NOSTATUS

    sf = self.status_file
    with open(sf) as f:
        content = f.read()
        if self.cfg.system.msg_completed in content:
            return JobStatus.COMPLETED
        elif self.cfg.system.msg_failed in content:
            return JobStatus.FAILED
        elif self.cfg.system.msg_running in content:
            return JobStatus.RUNNING

    if self.is_submitted:
        return JobStatus.SUBMITTED

    return JobStatus.UNKNOWN

status_symbol_help() staticmethod

Return help string for status codes.

Source code in duqtools/models/_job.py 
67
68
69
70
@staticmethod
def status_symbol_help():
    """Return help string for status codes."""
    return JobStatus.symbol_help()

submit()

Submit job.

Source code in duqtools/models/_job.py 
157
158
159
160
161
162
163
def submit(self):
    """Submit job."""
    from duqtools.systems import get_system
    debug(f'Put lockfile in place for {self.lockfile}')
    self.lockfile.touch()

    get_system(self.cfg).submit_job(self)

wait_until_done(time_step=1.0)

Submit task and wait until done.

Parameters:

  • time_step (float, default: 1.0 ) –

    Time in seconds step between status updates.

Source code in duqtools/models/_job.py 
174
175
176
177
178
179
180
181
182
183
def wait_until_done(self, time_step: float = 1.0):
    """Submit task and wait until done.

    Parameters
    ----------
    time_step : float, optional
        Time in seconds step between status updates.
    """
    while self.status() in (JobStatus.RUNNING, JobStatus.NOSTATUS):
        time.sleep(time_step)

Run

Bases: BaseModel

Dataclass describing a run.

Runs

Bases: RootModel

Dataclass describing a collection of runs.

alt_errorband_chart(source, *, x, y, z='time')

Generate an altair errorband plot from a dataframe.

Parameters:

  • source (DataFrame) –

    Input dataframe

  • x (str) –

    X-value to plot, corresponds to a column in the source data

  • y (str) –

    Y-value to plot, corresponds to a column in the source data

  • z (str, default: 'time' ) –

    Slider variable (time), corresponds to a column in the source data

Returns:

  • Chart –

    Return an altair chart.

Source code in duqtools/_plot_utils.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def alt_errorband_chart(source: Union[pd.DataFrame, xr.Dataset],
                        *,
                        x: str,
                        y: str,
                        z: str = 'time') -> alt.Chart:
    """Generate an altair errorband plot from a dataframe.

    Parameters
    ----------
    source : pd.DataFrame
        Input dataframe
    x : str
        X-value to plot, corresponds to a column in the source data
    y : str
        Y-value to plot, corresponds to a column in the source data
    z : str
        Slider variable (time), corresponds to a column in the source data

    Returns
    -------
    alt.Chart
        Return an altair chart.
    """
    source = _standardize_data(source, z=z)

    max_y = source[y].max()
    max_slider = source['slider'].max()

    line = alt.Chart(source).mark_line().encode(
        x=f'{x}:Q',
        y=alt.Y(
            f'mean({y}):Q',
            scale=alt.Scale(domain=(0, max_y)),
            axis=alt.Axis(format='.4~g'),
        ),
        color=alt.Color('tstep:N'),
    )

    # altair-viz.github.io/user_guide/generated/core/altair.ErrorBandDef
    band = alt.Chart(source).mark_errorband(extent='stdev',
                                            interpolate='linear').encode(
                                                x=f'{x}:Q',
                                                y=f'{y}:Q',
                                                color=alt.Color('tstep:N'),
                                            )

    ref = alt.Chart(source).mark_line(strokeDash=[5, 5]).encode(
        x=f'{x}:Q',
        y=f'mean({y}):Q',
        color=alt.Color('tstep:N'),
    )

    if max_slider != 0:
        slider = alt.binding_range(min=0, max=max_slider, step=1)
        select_step = alt.selection_point(name=z,
                                          fields=['slider'],
                                          bind=slider,
                                          value=0)

        line = line.add_params(select_step).transform_filter(
            select_step).interactive()
        band = band.add_params(select_step).transform_filter(
            select_step).interactive()

        slider = alt.binding_range(name=f'Reference {z} index',
                                   min=0,
                                   max=max_slider,
                                   step=1)

        select_step = alt.selection_point(name='reference',
                                          fields=['slider'],
                                          bind=slider,
                                          value=0)

        ref = ref.add_params(select_step).transform_filter(
            select_step).interactive()

    return line + band + ref

alt_line_chart(source, *, x, y, z='time', std=False)

Generate an altair line chart from a dataframe.

Parameters:

  • source (DataFrame) –

    Input dataframe

  • x (str) –

    X-value to plot, corresponds to a column in the source data

  • y (str) –

    Y-value to plot, corresponds to a column in the source data

  • z (str, default: 'time' ) –

    Slider variable (time), corresponds to a column in the source data

  • std (bool, default: False ) –

    Plot the error bound from {x}_error_upper in the plot as well

Returns:

  • Chart –

    Return an altair chart.

Source code in duqtools/_plot_utils.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def alt_line_chart(source: Union[pd.DataFrame, xr.Dataset],
                   *,
                   x: str,
                   y: str,
                   z: str = 'time',
                   std: bool = False) -> alt.Chart:
    """Generate an altair line chart from a dataframe.

    Parameters
    ----------
    source : pd.DataFrame
        Input dataframe
    x : str
        X-value to plot, corresponds to a column in the source data
    y : str
        Y-value to plot, corresponds to a column in the source data
    z : str
        Slider variable (time), corresponds to a column in the source data

    std : bool
        Plot the error bound from {x}_error_upper in the plot as well

    Returns
    -------
    alt.Chart
        Return an altair chart.
    """
    source = _standardize_data(source, z=z)
    max_y = source[y].max()

    if std:
        source[y + '_upper'] = source[y] + source[y + '_error_upper']
        source[y + '_lower'] = source[y] - source[y + '_error_upper']
        max_y = source[y + '_upper'].max()

    max_slider = source['slider'].max()

    if std:
        band = alt.Chart(source).mark_area(opacity=0.3).encode(
            x=f'{x}:Q',
            y=alt.Y(f'{y}_upper:Q', title=y),
            y2=alt.Y2(f'{y}_lower:Q', title=y),
            color=alt.Color('run:N'),
        )

    line = alt.Chart(source).mark_line().encode(
        x=f'{x}:Q',
        y=alt.Y(
            f'{y}:Q',
            scale=alt.Scale(domain=(0, max_y)),
            axis=alt.Axis(format='.4~g'),
        ),
        color=alt.Color('run:N'),
        tooltip='run',
    )

    ref = alt.Chart(source).mark_line(strokeDash=[5, 5]).encode(
        x=f'{x}:Q', y=f'{y}:Q', color=alt.Color('run:N'), tooltip='run')

    if max_slider != 0:
        slider = alt.binding_range(name=f'{z} index',
                                   min=0,
                                   max=max_slider,
                                   step=1)
        select_step = alt.selection_point(name=z,
                                          fields=['slider'],
                                          bind=slider,
                                          value=0)
        line = line.add_params(select_step).transform_filter(
            select_step).interactive()
        if std:
            band = band.transform_filter(select_step).interactive()

        first_run = source.iloc[0].run
        slider = alt.binding_range(name=f'Reference {z} index',
                                   min=0,
                                   max=max_slider,
                                   step=1)
        select_step = alt.selection_point(name='reference',
                                          fields=['slider'],
                                          bind=slider,
                                          value=0)

        ref = ref.add_params(select_step).transform_filter(
            select_step).transform_filter(
                alt.datum.run == first_run).interactive()

    if std:
        return line + ref + band
    else:
        return line + ref

create(config, **kwargs)

Wrapper around create for python api.

Source code in duqtools/create.py 
302
303
304
305
306
307
308
309
310
311
312
313
def create_api(config: dict, **kwargs) -> dict[str, tuple[Job, Run]]:
    """Wrapper around create for python api."""
    cfg = Config.from_dict(config)
    runs = create(cfg=cfg, **kwargs)

    if len(runs) == 0:
        raise CreateError('No runs were created, check logs for errors.')

    return {
        str(run.shortname): (Job(run.dirname, cfg=cfg), run)
        for run in runs
    }

duqmap(function, *, runs=None, **kwargs)

Duqmap is a mapping function which can be used to map a user defined function function over either the runs created by duqtools, or the runs specified by the user in runs.

An important gotcha is that when Paths are used to define the runs, duqtools does not know how to associate the corresponding ImasHandles, as that information is not available. So when using it in this way, it is not possible to provide a function which takes an ImasHandle as input.

Parameters:

  • function (Callable[[Run | ImasHandle], Any]) –

    function which is called for each run, specified either by runs, or implicitly by any available runs.yaml

  • runs (Optional[List[Run | Path]], default: None ) –

    A list of runs over which to operate the function

  • kwargs –

    optional arguments that need to be passed to each function that you provide

Returns:

  • List[Any]: –

    A list of anything that your function returns

Source code in duqtools/duqmap.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def duqmap(function: Callable[[Run | ImasHandle], Any],
           *,
           runs: Optional[List[Run | Path]] = None,
           **kwargs) -> List[Any]:
    """Duqmap is a mapping function which can be used to map a user defined
    function `function` over either the runs created by duqtools, or the runs
    specified by the user in `runs`.

    An important gotcha is that when `Paths` are used to define the runs, duqtools
    does not know how to associate the corresponding ImasHandles, as that information
    is not available.  So when using it in this way, it is not possible to provide a
    function which takes an `ImasHandle` as input.

    Parameters
    ----------
    function : Callable[[Run | ImasHandle], Any]
        function which is called for each run, specified either by `runs`, or implicitly
        by any available `runs.yaml`
    runs : Optional[List[Run | Path]]
        A list of runs over which to operate the function
    kwargs :
        optional arguments that need to be passed to each `function` that you provide

    Returns
    -------
    List[Any]:
        A list of anything that your function returns
    """
    try:
        # Gets the type of the first argument to the function, if it exists
        argument = next(iter(signature(function).parameters.items()))[1]
    except Exception:
        raise NotImplementedError(
            f'Dont know how to map: {function}, which has no arguments')

    argument_type = argument.annotation

    if argument_type in [Run, 'Run']:
        map_fun: Callable[[Any], Any] = duqmap_run
    elif argument_type in [ImasHandle, 'ImasHandle']:
        map_fun = duqmap_imas
    else:
        raise NotImplementedError('Dont know how to map function signature:'
                                  f' {function.__name__}{signature(function)}')

    return map_fun(function, runs=runs, **kwargs)  # type: ignore

get_status(config, **kwargs)

Wrapper around status for python api.

Source code in duqtools/status.py 
222
223
224
225
def status_api(config: dict, **kwargs):
    """Wrapper around status for python api."""
    cfg = Config.from_dict(config)
    return status(cfg=cfg, **kwargs)

recreate(config, runs, **kwargs)

Wrapper around create for python api.

Source code in duqtools/create.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
def recreate_api(config: dict, runs: Sequence[Path],
                 **kwargs) -> dict[str, tuple[Job, Run]]:
    """Wrapper around create for python api."""
    cfg = Config.from_dict(config)
    runs = recreate(cfg=cfg, runs=runs, **kwargs)

    if len(runs) == 0:
        raise CreateError('No runs were recreated, check logs for errors.')

    return {
        str(run.shortname): (Job(run.dirname, cfg=cfg), run)
        for run in runs
    }

submit(config, **kwargs)

Wrapper around submit for python api.

Source code in duqtools/submit.py 
273
274
275
276
def submit_api(config: dict, **kwargs):
    """Wrapper around submit for python api."""
    cfg = Config.from_dict(config)
    return submit(cfg=cfg, **kwargs)