Uh oh!
There was an error while loading. Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork 0
updating summarize cli for consistency with other subcmds. #360
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Uh oh!
There was an error while loading. Please reload this page.
Changes from all commits
5342210c166f704df2d3fd85526fad5d0d3a9893dec5c3d29File filter
Filter by extension
Conversations
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -2,7 +2,129 @@ | ||
| Package containing CLI modules. | ||
| """ | ||
| from mmif.utils.cli import describe | ||
| import contextlib | ||
| import io | ||
| import os | ||
| import sys | ||
| from typing import Iterator, Optional, TextIO, cast | ||
| @contextlib.contextmanager | ||
| def open_cli_io_arg(path_or_dash: Optional[str], | ||
| mode: str = 'r', | ||
| encoding: Optional[str] = None, | ||
| errors: Optional[str] = None, | ||
| default_stdin: bool = False, | ||
| ) -> Iterator[TextIO]: | ||
| """ | ||
| Context manager for opening files with stdin/stdout support. | ||
| This function is intended for plain text streams (e.g. JSON/MMIF) and does | ||
| not support binary modes (e.g., 'rb', 'wb'). | ||
keighrim marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| This is a native replacement for argparse.FileType which is deprecated as | ||
| of Python 3.14 due to resource leak issues. Unlike FileType, this defers | ||
| file opening until actually needed and ensures proper cleanup via context | ||
| manager. | ||
| Handles the common CLI pattern where: | ||
| - '-' means stdin (read mode) or stdout (write mode) | ||
| - None means "argument not provided"; when default_stdin=True, it falls back | ||
| to stdin/stdout | ||
| - Regular paths open actual files with proper resource management | ||
| :param path_or_dash: File path, '-' for stdin/stdout, or None for no argument | ||
| :param mode: File mode ('r' for reading, 'w' for writing). Binary modes are | ||
| not supported. | ||
| :param encoding: Optional file encoding | ||
| :param errors: Optional error handling strategy for encoding | ||
| :param default_stdin: If True and path_or_dash is None, default to stdin | ||
| (mode 'r') or stdout (mode 'w') | ||
| :returns: Context manager yielding text-mode file handle | ||
| :rtype: Iterator[TextIO] | ||
| Example usage:: | ||
| # Read from file or stdin | ||
| with open_cli_io_arg(args.input, 'r', default_stdin=True) as f: | ||
| content = f.read() | ||
| # Write to file or stdout | ||
| with open_cli_io_arg(args.output, 'w', default_stdin=True) as f: | ||
| f.write(content) | ||
| """ | ||
| # Valid text modes for file operations | ||
| _READ_FLAGS = frozenset({'r', '+'}) | ||
| _WRITE_FLAGS = frozenset({'w', 'a', 'x', '+'}) | ||
| if 'b' in mode: | ||
| raise ValueError( | ||
| f"Binary mode '{mode}' is not supported. " | ||
| "Use text modes ('r', 'w', 'a', 'x') instead." | ||
| ) | ||
| needs_read = bool(set(mode) & _READ_FLAGS) | ||
| needs_write = bool(set(mode) & _WRITE_FLAGS) | ||
| should_use_stdio = path_or_dash == '-' or ( | ||
| path_or_dash is None and default_stdin | ||
| ) | ||
| file_handle: Optional[TextIO] = None | ||
| should_close = False | ||
| try: | ||
| if should_use_stdio: | ||
| if needs_read and needs_write: | ||
| raise ValueError( | ||
| f"Mode '{mode}' not supported with stdin/stdout " | ||
| "(use read or write only)" | ||
| ) | ||
| if needs_read: | ||
| # Check for missing input when stdin is a terminal | ||
| if ( | ||
| path_or_dash is None | ||
| and default_stdin | ||
| and sys.stdin.isatty() | ||
| ): | ||
| raise SystemExit("error: No input provided.") | ||
| file_handle = sys.stdin | ||
| elif needs_write: | ||
| file_handle = sys.stdout | ||
| else: | ||
| raise ValueError( | ||
| f"Mode '{mode}' not supported with stdin/stdout " | ||
| "(use 'r' or 'w')" | ||
| ) | ||
| elif isinstance(path_or_dash, str): | ||
| if needs_read and not os.path.exists(path_or_dash): | ||
| raise FileNotFoundError(f"Input path does not exist: {path_or_dash}") | ||
| file_handle = cast(TextIO, io.open(path_or_dash, mode, encoding=encoding, errors=errors)) | ||
| should_close = True | ||
| elif path_or_dash is None: | ||
| # None without default_stdin means no file specified | ||
| raise ValueError( | ||
| "No file path provided. Use '-' for stdin/stdout or set default_stdin=True." | ||
| ) | ||
| else: | ||
| raise TypeError( | ||
| f"Invalid type for path_or_dash: {type(path_or_dash).__name__}. " | ||
| "Expected str or None." | ||
| ) | ||
| yield file_handle | ||
| finally: | ||
| if should_close and file_handle is not None: | ||
| file_handle.close() | ||
| # keep imports of CLI modules for historical reasons | ||
| # keep them here in the bottom to avoid circular imports | ||
| from mmif.utils.cli import rewind | ||
| from mmif.utils.cli import source | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,10 +1,12 @@ | ||
| import argparse | ||
| import json | ||
| import os | ||
| import sys | ||
| import textwrap | ||
| from pathlib import Path | ||
| from typing import Union, cast | ||
| from mmif.utils.cli import open_cli_io_arg | ||
| from mmif.utils.workflow_helper import generate_workflow_identifier, describe_single_mmif, \ | ||
| describe_mmif_collection | ||
| # gen_param_hash is imported for backward compatibility | ||
| @@ -26,11 +28,6 @@ def generate_pipeline_identifier(mmif_file: Union[str, Path]) -> str: | ||
| def describe_argparser(): | ||
| """ | ||
| Returns two strings: one-line description of the argparser, and | ||
| additional material, which will be shown in `clams --help` and | ||
| `clams <subcmd> --help`, respectively. | ||
| """ | ||
| oneliner = ( | ||
| 'Describe the workflow specification from a MMIF file or a ' | ||
| 'collection of MMIF files.' | ||
| @@ -74,13 +71,12 @@ def prep_argparser(**kwargs): | ||
| "MMIF_FILE", | ||
| nargs="?", | ||
| type=str, | ||
| default=None if sys.stdin.isatty() else sys.stdin, | ||
| default=None, | ||
| help='input MMIF file, a directory of MMIF files, or STDIN if `-` or not provided.' | ||
| ) | ||
| parser.add_argument( | ||
| "-o", "--output", | ||
| type=argparse.FileType("w"), | ||
| default=sys.stdout, | ||
| type=str, default=None, | ||
| help='output file path, or STDOUT if not provided.' | ||
| ) | ||
| parser.add_argument( | ||
| @@ -105,16 +101,13 @@ def main(args): | ||
| """ | ||
| output = {} | ||
| # if input is a directory | ||
| if isinstance(args.MMIF_FILE, str) and Path(args.MMIF_FILE).is_dir(): | ||
| if isinstance(args.MMIF_FILE, (str, os.PathLike)) and Path(args.MMIF_FILE).is_dir(): | ||
| output = describe_mmif_collection(args.MMIF_FILE) | ||
| # if input is a file or stdin | ||
| else: | ||
| # Read MMIF content | ||
| if hasattr(args.MMIF_FILE, 'read'): | ||
| mmif_content = args.MMIF_FILE.read() | ||
| else: | ||
| with open(args.MMIF_FILE, 'r') as f: | ||
| mmif_content = f.read() | ||
| with open_cli_io_arg(args.MMIF_FILE, 'r', default_stdin=True) as input_file: | ||
keighrim marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| mmif_content = input_file.read() | ||
| # For file input, we need to handle the path | ||
| # If input is from stdin, create a temp file | ||
| @@ -132,11 +125,9 @@ def main(args): | ||
| tmp_path.unlink() | ||
| if output: | ||
| if args.pretty: | ||
| json.dump(output, args.output, indent=2) | ||
| else: | ||
| json.dump(output, args.output) | ||
| args.output.write('\n') | ||
| with open_cli_io_arg(args.output, 'w', default_stdin=True) as output_file: | ||
| json.dump(output, output_file, indent=2 if args.pretty else None) | ||
| output_file.write('\n') | ||
| if __name__ == "__main__": | ||
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.