Tsakagur
05/09/2022, 3:30 PMpython
def main(
pipeline_name: str = typer.Option(default=None),
tag: str = typer.Option(default=None),
env: str = typer.Option(default="pipelines-aml"),
scope: Optional[Scope] = typer.Option(default=None, help= "Must be present if one of the following CLI option is."),
train_start_date: datetime = typer.Option(default=None, formats=[DATE_FORMAT]),
train_end_date: datetime = typer.Option(default=None, formats=[DATE_FORMAT]),
val_start_date: datetime = typer.Option(default=None, formats=[DATE_FORMAT]),
val_end_date: datetime = typer.Option(default=None, formats=[DATE_FORMAT]),
) -> None:
"""
Pico model training.
Emulates ``kedro run`` command.
"""
# Kedro session arguments
session_kwargs = {"env": env}
if scope and (train_start_date or train_end_date or val_start_date or val_end_date):
extra_params = {}
if train_start_date:
extra_params[f"{scope}_train_start_date"] = train_start_date
if train_end_date:
extra_params[f"{scope}_train_end_date"] = train_end_date
if val_start_date:
extra_params[f"{scope}_val_start_date"] = val_start_date
if val_end_date:
extra_params[f"{scope}_val_end_date"] = val_end_date
session_kwargs["extra_params"] = extra_params
print(f'{session_kwargs=}')
# Kedro run arguments
run_kwargs = {}
if pipeline_name:
run_kwargs["pipeline_name"] = pipeline_name
if tag:
run_kwargs["tags"] = [tag]
print(f'{run_kwargs=}')
metadata = bootstrap_project(Path.cwd())
with KedroSession.create(metadata.package_name, project_path=Path.cwd(), **session_kwargs) as session:
session.run(**run_kwargs)