cli: implement --watch option

This commit is contained in:
Chris Carroll
2024-09-07 16:28:45 -04:00
committed by Sina Atalay
parent f7e24bb24c
commit fa3f165d0e
3 changed files with 112 additions and 1 deletions

View File

@@ -79,6 +79,7 @@ dependencies = [
'typer==0.13.1', # to create the command-line interface
"markdown==3.7", # to convert Markdown to HTML
"PyMuPDF==1.24.14", # to convert PDF files to images
"watchdog==5.0.2", # to poll files for updates
]
classifiers = [
"Intended Audience :: Science/Research",

View File

@@ -11,7 +11,7 @@ import typer
from rich import print
from .. import __version__, data, renderer
from . import printer, utilities
from . import printer, utilities, watcher
app = typer.Typer(
rich_markup_mode="rich",
@@ -123,6 +123,14 @@ def cli_command_render(
help="Don't generate the PNG file.",
),
] = False,
watch: Annotated[
bool,
typer.Option(
"--watch",
"-w",
help="Automatically generate files on change.",
),
] = False,
# This is a dummy argument for the help message for
# extra_data_model_override_argumets:
_: Annotated[
@@ -136,6 +144,32 @@ def cli_command_render(
extra_data_model_override_argumets: typer.Context = None, # type: ignore
):
"""Render a CV from a YAML input file."""
if watch:
def rerun_command():
cli_command_render(
input_file_name=input_file_name,
use_local_latex_command=use_local_latex_command,
output_folder_name=output_folder_name,
latex_path=latex_path,
pdf_path=pdf_path,
markdown_path=markdown_path,
html_path=html_path,
png_path=png_path,
dont_generate_markdown=dont_generate_markdown,
dont_generate_html=dont_generate_html,
dont_generate_png=dont_generate_png,
watch=False,
extra_data_model_override_argumets=extra_data_model_override_argumets,
)
file_path = utilities.string_to_file_path(input_file_name)
if file_path is None:
raise FileNotFoundError(f"Unable to find path to {input_file_name}")
watcher.watch_file(file_path, rerun_command)
return
printer.welcome()
input_file_path: pathlib.Path = pathlib.Path(input_file_name).absolute()

76
rendercv/cli/watcher.py Normal file
View File

@@ -0,0 +1,76 @@
"""
The `rendercv.cli.watcher` module contains all the functions and classes that are used to watch files and emit callbacks.
"""
import os
import pathlib
import time
from hashlib import sha256
from typing import Callable
from watchdog.events import FileModifiedEvent, FileSystemEventHandler
from watchdog.observers import Observer
from typer import Exit
class ModifiedCVEventHandler(FileSystemEventHandler):
"""This class handles the file changes and triggers a specified `callback` ignoring duplicate changes.
Args:
file_path (pathlib.Path): The path of the file to watch for.
callback (Callable[..., None]): The function to be called on file modification. *CALLBACK MUST BE NON-BLOCKING*
"""
file_path: pathlib.Path
callback: Callable[..., None]
previous_hash: str = ""
def __init__(self, file_path: pathlib.Path, callback: Callable[..., None]):
self.callback = callback
self.file_path = file_path
# Handle an initial pass manually
self.on_modified(FileModifiedEvent(src_path=str(self.file_path)))
def on_modified(self, event: FileModifiedEvent) -> None:
if event.src_path != str(self.file_path):
# Ignore any events that aren't our file.
return
file_hash = sha256(open(event.src_path).read().encode("utf-8")).hexdigest()
if file_hash == self.previous_hash:
# Exit if file hash has not changed.
return
self.previous_hash = file_hash
try:
self.callback()
except Exit:
... # Suppress typer Exit so we can continue watching even if we see errors.
def watch_file(file_path: pathlib.Path, callback: Callable[..., None]):
"""Watch file located at `file_path` and trigger callback on file modification.
Args:
file_path (pathlib.Path): The path of the file to watch for.
callback (Callable[..., None]): The function to be called on file modification. *CALLBACK MUST BE NON-BLOCKING*
"""
event_handler = ModifiedCVEventHandler(file_path, callback)
observer = Observer()
# If on windows we have to poll the parent directory instead of the file.
if os.name == "nt":
observer.schedule(event_handler, str(file_path.parent), recursive=False)
else:
observer.schedule(event_handler, str(file_path), recursive=False)
observer.start()
try:
while True:
time.sleep(1)
finally:
observer.stop()
observer.join()