Skip to content

Delphyne Command Line Interface

DelphyneCLI

The Delphyne Command Line Interface.

The delphyne package features a delphyne command line application that is automatically generated from the DelphyneCLI class using fire. In particular, this application can be used to check demonstration files, execute command files, and launch the Delphyne language server.

Source code in src/delphyne/__main__.py 
 26
 27
 28
 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
120
121
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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class DelphyneCLI:
    """
    The Delphyne Command Line Interface.

    The `delphyne` package features a `delphyne` command line
    application that is automatically generated from the `DelphyneCLI`
    class using [fire](https://github.com/google/python-fire). In
    particular, this application can be used to check demonstration
    files, execute command files, and launch the Delphyne language
    server.
    """

    def __init__(
        self,
        *,
        workspace_dir: Path | None = None,
        ensure_no_error: bool = False,
        ensure_no_warning: bool = False,
    ):
        """
        Arguments:
            workspace_dir: The workspace directory. If not provided, it
                is deduced for each demonstration or command file by
                considering the closest transitive parent directory that
                contains a `delphyne.yaml` file. If no such directory
                exists, the current working directory is used.
            ensure_no_error: Exit with a nonzero code if an error is
                produced.
            ensure_no_warning: Exit with a nonzero code if a warning is
                produced.
        """
        self.workspace_dir = workspace_dir
        self.ensure_no_error = ensure_no_error
        self.ensure_no_warning = ensure_no_warning

    def _process_diagnostics(
        self,
        warnings: list[str],
        errors: list[str],
        use_stderr: bool = False,
        show_summary: bool = True,
    ):
        show = partial(print, file=sys.stderr if use_stderr else sys.stdout)
        num_errors = len(errors)
        num_warnings = len(warnings)
        if show_summary:
            show(f"{num_errors} error(s), {num_warnings} warning(s)")
        if errors or warnings:
            show("")
        for e in errors:
            show(f"Error: {e}")
        for w in warnings:
            show(f"Warning: {w}")
        if self.ensure_no_error and num_errors > 0:
            exit(1)
        if self.ensure_no_warning and num_warnings > 0:
            exit(1)

    def _workspace_dir_for(self, file: Path) -> Path:
        """
        Get the workspace directory for a given file.
        """
        workspace_dir = self.workspace_dir
        if workspace_dir is None:
            workspace_dir = find_workspace_dir(file)
        if workspace_dir is None:
            workspace_dir = Path.cwd()
        return workspace_dir

    def check(self, file: str):
        """
        Check a demonstration file.
        """
        file_path = Path(file)
        workspace_dir = self._workspace_dir_for(file_path)
        config = load_config(workspace_dir, local_config_from=file_path)
        feedback = check_demo_file(
            file_path, config.strategy_dirs, config.modules
        )
        self._process_diagnostics(feedback.warnings, feedback.errors)

    def run(
        self,
        file: str,
        *,
        cache: bool = False,
        update: bool = False,
        no_output: bool = False,
        no_header: bool = False,
        no_status: bool = False,
        filter: list[str] | None = None,
        clear: bool = False,
    ):
        """
        Execute a command file.

        Print an updated command file with an `outcome` section added on
        stdout. Print other information on stderr.

        Arguments:
            file: Path to the command file to execute.
            cache: Enable caching (assuming the command supports it).
            update: Update the command file in place with the outcome.
            no_output: Do not print on stdout.
            no_header: Only print the `outcome` section on stdout.
            no_status: Do not show the progress bar.
            filter: Only show the provided subset of fields for the
                `outcome.result` section.
            clear: When this option is passed, all other options are
                ignored and the `clear` method is called.
        """
        file_path = Path(file)
        workspace_dir = self._workspace_dir_for(file_path)
        config = load_config(workspace_dir, local_config_from=file_path)
        config = replace(
            config,
            status_refresh_period=STATUS_REFRESH_PERIOD_IN_SECONDS,
            result_refresh_period=None,
        )
        if clear:
            self.clear(file)
            return
        if update:
            no_output = True
            no_header = False
        if cache and not config.cache_root:
            config = replace(config, cache_root=file_path.parent / "cache")
        with open(file, "r") as f:
            spec = ty.pydantic_load(CommandSpec, yaml.safe_load(f))
        cmd, args = spec.load(config.base)
        if cache:
            assert hasattr(args, "cache_dir"), (
                "Command does not have a `cache_dir` argument."
            )
            assert hasattr(args, "cache_format"), (
                "Command does not have a `cache_format` argument."
            )
            if not args.cache_dir:
                args.cache_dir = file_path.stem
            args.cache_format = "db"
        progress = StatusIndicator(sys.stderr, show=not no_status)
        res = std.run_command(cmd, args, config, on_status=progress.on_status)
        progress.done()
        res_type = std.command_optional_result_wrapper_type(cmd)
        res_python: Any = ty.pydantic_dump(res_type, res)
        if filter and res_python["result"] is not None:
            res_python["result"] = {
                k: v for k, v in res_python["result"].items() if k in filter
            }
        if no_header:
            output = pretty_yaml(res_python)
        else:
            with open(file_path, "r") as f:
                header = command_file_header(f.read())
            output = header.rstrip() + "\n"
            output += pretty_yaml({"outcome": res_python})
        if not no_output:
            print(output)
        if update:
            with open(file_path, "w") as f:
                f.write(output)
        errors = [d[1] for d in res.diagnostics if d[0] == "error"]
        warnings = [d[1] for d in res.diagnostics if d[0] == "warning"]
        self._process_diagnostics(
            warnings,
            errors,
            use_stderr=True,
            show_summary=self.ensure_no_error or self.ensure_no_warning,
        )

    def clear(self, file: str):
        """
        Clear the outcome of a command file by updating it in place.
        """
        path = Path(file)
        with open(path, "r") as f:
            content = f.read()
        new_content = command_file_header(content)
        with open(path, "w") as f:
            f.write(new_content)

    def serve(self):
        """
        Launch an instance of the Delphyne language server.
        """
        from delphyne.server.__main__ import main

        main()

__init__ 

__init__(
    *,
    workspace_dir: Path | None = None,
    ensure_no_error: bool = False,
    ensure_no_warning: bool = False,
)

Parameters:

Name Type Description Default
workspace_dir Path | None

The workspace directory. If not provided, it is deduced for each demonstration or command file by considering the closest transitive parent directory that contains a delphyne.yaml file. If no such directory exists, the current working directory is used.

 None
ensure_no_error bool

Exit with a nonzero code if an error is produced.

 False
ensure_no_warning bool

Exit with a nonzero code if a warning is produced.

 False
Source code in src/delphyne/__main__.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(
    self,
    *,
    workspace_dir: Path | None = None,
    ensure_no_error: bool = False,
    ensure_no_warning: bool = False,
):
    """
    Arguments:
        workspace_dir: The workspace directory. If not provided, it
            is deduced for each demonstration or command file by
            considering the closest transitive parent directory that
            contains a `delphyne.yaml` file. If no such directory
            exists, the current working directory is used.
        ensure_no_error: Exit with a nonzero code if an error is
            produced.
        ensure_no_warning: Exit with a nonzero code if a warning is
            produced.
    """
    self.workspace_dir = workspace_dir
    self.ensure_no_error = ensure_no_error
    self.ensure_no_warning = ensure_no_warning

check 

check(file: str)

Check a demonstration file.

Source code in src/delphyne/__main__.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
def check(self, file: str):
    """
    Check a demonstration file.
    """
    file_path = Path(file)
    workspace_dir = self._workspace_dir_for(file_path)
    config = load_config(workspace_dir, local_config_from=file_path)
    feedback = check_demo_file(
        file_path, config.strategy_dirs, config.modules
    )
    self._process_diagnostics(feedback.warnings, feedback.errors)

run 

run(
    file: str,
    *,
    cache: bool = False,
    update: bool = False,
    no_output: bool = False,
    no_header: bool = False,
    no_status: bool = False,
    filter: list[str] | None = None,
    clear: bool = False,
)

Execute a command file.

Print an updated command file with an outcome section added on stdout. Print other information on stderr.

Parameters:

Name Type Description Default
file str

Path to the command file to execute.

 required
cache bool

Enable caching (assuming the command supports it).

 False
update bool

Update the command file in place with the outcome.

 False
no_output bool

Do not print on stdout.

 False
no_header bool

Only print the outcome section on stdout.

 False
no_status bool

Do not show the progress bar.

 False
filter list[str] | None

Only show the provided subset of fields for the outcome.result section.

 None
clear bool

When this option is passed, all other options are ignored and the clear method is called.

 False
Source code in src/delphyne/__main__.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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
def run(
    self,
    file: str,
    *,
    cache: bool = False,
    update: bool = False,
    no_output: bool = False,
    no_header: bool = False,
    no_status: bool = False,
    filter: list[str] | None = None,
    clear: bool = False,
):
    """
    Execute a command file.

    Print an updated command file with an `outcome` section added on
    stdout. Print other information on stderr.

    Arguments:
        file: Path to the command file to execute.
        cache: Enable caching (assuming the command supports it).
        update: Update the command file in place with the outcome.
        no_output: Do not print on stdout.
        no_header: Only print the `outcome` section on stdout.
        no_status: Do not show the progress bar.
        filter: Only show the provided subset of fields for the
            `outcome.result` section.
        clear: When this option is passed, all other options are
            ignored and the `clear` method is called.
    """
    file_path = Path(file)
    workspace_dir = self._workspace_dir_for(file_path)
    config = load_config(workspace_dir, local_config_from=file_path)
    config = replace(
        config,
        status_refresh_period=STATUS_REFRESH_PERIOD_IN_SECONDS,
        result_refresh_period=None,
    )
    if clear:
        self.clear(file)
        return
    if update:
        no_output = True
        no_header = False
    if cache and not config.cache_root:
        config = replace(config, cache_root=file_path.parent / "cache")
    with open(file, "r") as f:
        spec = ty.pydantic_load(CommandSpec, yaml.safe_load(f))
    cmd, args = spec.load(config.base)
    if cache:
        assert hasattr(args, "cache_dir"), (
            "Command does not have a `cache_dir` argument."
        )
        assert hasattr(args, "cache_format"), (
            "Command does not have a `cache_format` argument."
        )
        if not args.cache_dir:
            args.cache_dir = file_path.stem
        args.cache_format = "db"
    progress = StatusIndicator(sys.stderr, show=not no_status)
    res = std.run_command(cmd, args, config, on_status=progress.on_status)
    progress.done()
    res_type = std.command_optional_result_wrapper_type(cmd)
    res_python: Any = ty.pydantic_dump(res_type, res)
    if filter and res_python["result"] is not None:
        res_python["result"] = {
            k: v for k, v in res_python["result"].items() if k in filter
        }
    if no_header:
        output = pretty_yaml(res_python)
    else:
        with open(file_path, "r") as f:
            header = command_file_header(f.read())
        output = header.rstrip() + "\n"
        output += pretty_yaml({"outcome": res_python})
    if not no_output:
        print(output)
    if update:
        with open(file_path, "w") as f:
            f.write(output)
    errors = [d[1] for d in res.diagnostics if d[0] == "error"]
    warnings = [d[1] for d in res.diagnostics if d[0] == "warning"]
    self._process_diagnostics(
        warnings,
        errors,
        use_stderr=True,
        show_summary=self.ensure_no_error or self.ensure_no_warning,
    )

clear 

clear(file: str)

Clear the outcome of a command file by updating it in place.

Source code in src/delphyne/__main__.py 
196
197
198
199
200
201
202
203
204
205
def clear(self, file: str):
    """
    Clear the outcome of a command file by updating it in place.
    """
    path = Path(file)
    with open(path, "r") as f:
        content = f.read()
    new_content = command_file_header(content)
    with open(path, "w") as f:
        f.write(new_content)

serve 

serve()

Launch an instance of the Delphyne language server.

Source code in src/delphyne/__main__.py 
207
208
209
210
211
212
213
def serve(self):
    """
    Launch an instance of the Delphyne language server.
    """
    from delphyne.server.__main__ import main

    main()