Monitor log files in real-time.
(
log_dir: Annotated[
str,
typer.Option(
"--log-dir",
"-d",
help="Path to the log directory. If provided, it will be used directly and ignore --config.",
),
] = "",
config: Annotated[
str,
typer.Option(
"--config",
"-c",
help="Path to the config file. If provided, it will automatically locate the log directory based on the config.",
),
] = "",
keyword: Annotated[
Optional[str],
typer.Option(
"--keyword",
"-k",
help="Only track log files containing the keyword in their filenames.",
),
] = None,
level: Annotated[
str,
typer.Option("--level", "-l", help="The minimum log level to display in real-time."),
] = "INFO",
last_n_lines: Annotated[
int,
typer.Option("--last-n-lines", "-n", help="Number of last lines to display when starting."),
] = 0,
search_pattern: Annotated[
Optional[str],
typer.Option(
"--search-pattern",
"-p",
help="The pattern to search in log files. Only search for history logs and display all lines containing the pattern.",
),
] = None,
no_color: Annotated[
bool,
typer.Option("--no-color", help="Disable colored output."),
] = False,
)
| 8 | |
| 9 | |
| 10 | def log_command( |
| 11 | log_dir: Annotated[ |
| 12 | str, |
| 13 | typer.Option( |
| 14 | "--log-dir", |
| 15 | "-d", |
| 16 | help="Path to the log directory. If provided, it will be used directly and ignore --config.", |
| 17 | ), |
| 18 | ] = "", |
| 19 | config: Annotated[ |
| 20 | str, |
| 21 | typer.Option( |
| 22 | "--config", |
| 23 | "-c", |
| 24 | help="Path to the config file. If provided, it will automatically locate the log directory based on the config.", |
| 25 | ), |
| 26 | ] = "", |
| 27 | keyword: Annotated[ |
| 28 | Optional[str], |
| 29 | typer.Option( |
| 30 | "--keyword", |
| 31 | "-k", |
| 32 | help="Only track log files containing the keyword in their filenames.", |
| 33 | ), |
| 34 | ] = None, |
| 35 | level: Annotated[ |
| 36 | str, |
| 37 | typer.Option("--level", "-l", help="The minimum log level to display in real-time."), |
| 38 | ] = "INFO", |
| 39 | last_n_lines: Annotated[ |
| 40 | int, |
| 41 | typer.Option("--last-n-lines", "-n", help="Number of last lines to display when starting."), |
| 42 | ] = 0, |
| 43 | search_pattern: Annotated[ |
| 44 | Optional[str], |
| 45 | typer.Option( |
| 46 | "--search-pattern", |
| 47 | "-p", |
| 48 | help="The pattern to search in log files. Only search for history logs and display all lines containing the pattern.", |
| 49 | ), |
| 50 | ] = None, |
| 51 | no_color: Annotated[ |
| 52 | bool, |
| 53 | typer.Option("--no-color", help="Disable colored output."), |
| 54 | ] = False, |
| 55 | ) -> None: |
| 56 | """Monitor log files in real-time.""" |
| 57 | from trinity.manager.log_manager import LogManager |
| 58 | |
| 59 | if not config and not log_dir: |
| 60 | raise typer.BadParameter("Either --config or --log-dir must be provided.") |
| 61 | if not log_dir: |
| 62 | cfg = load_config(config) |
| 63 | checkpoint_job_dir = cfg.get_checkpoint_job_dir() |
| 64 | # we do not use check_and_update here because user may use this command |
| 65 | # in another environment |
| 66 | log_dir = os.path.join(checkpoint_job_dir, "log") |
| 67 |
nothing calls this directly
no test coverage detected