Implement profile support

src/lvmcryo/__main__.py

@@ -68,6 +68,7 @@ def list_commands(self, ctx):
 @cli.command("ln2")
 @cli_coro()
 async def ln2(
+    ctx: typer.Context,
     #
     # Arguments
     #
@@ -86,6 +87,16 @@ async def ln2(
     #
     # General options
     #
+    profile: Annotated[
+        Optional[str],
+        Option(
+            "--profile",
+            "-p",
+            envvar="LVMCRYO_PROFILE",
+            help="Profile to use. A list of valid profiles and parameters "
+            "can be printed with lvmcryo list-profiles.",
+        ),
+    ] = None,
     config_file: Annotated[
         Optional[pathlib.Path],
         Option(
@@ -434,6 +445,8 @@ async def ln2(
             cameras=cameras or [],
             config_file=config_file,
             dry_run=dry_run,
+            clear_lock=clear_lock,
+            with_traceback=with_traceback,
             interactive=interactive,
             no_prompt=no_prompt,
             notify=notify,
@@ -461,6 +474,11 @@ async def ln2(
             data_path=data_path,
             data_extra_time=data_extra_time,
             version=__version__,
+            profile=profile,
+            # We cannot pass the context directly so we pass a dict of the
+            # origin of each parameter to reject profile parameters that
+            # have been manually defined.
+            param_source={pp: ctx.get_parameter_source(pp) for pp in ctx.params},
         )
     except ValueError as err:
         err_console.print(f"[red]Error parsing configuration:[/] {err}")
@@ -469,7 +487,7 @@ async def ln2(
     if config.write_log and config.log_path:
         log.start_file_logger(str(config.log_path))
 
-        if write_json:
+        if config.write_json:
             json_path = config.log_path.with_suffix(".json")
             json_handler = add_json_handler(log, json_path)
 
@@ -497,8 +515,8 @@ async def ln2(
     if not config.notify:
         log.debug("Notifications are disabled and will not be emitted.")
 
-    if config_file is not None:
-        log.info(f"Using configuration file: {config_file!s}")
+    if config.config_file is not None:
+        log.info(f"Using configuration file: {config.config_file!s}")
 
     if not config.no_prompt:
         stdout_console.print(f"Action {config.action.value} will run with:")
@@ -521,7 +539,7 @@ async def ln2(
         alerts_route=config.internal_config["api_routes"]["alerts"],
     )
 
-    if LOCKFILE.exists() and clear_lock:
+    if LOCKFILE.exists() and config.clear_lock:
         log.warning("Lock file exists. Removing it because --clear-lock.")
         LOCKFILE.unlink()
 
@@ -544,7 +562,7 @@ async def ln2(
             "the lock."
         )
 
-        if notify:
+        if config.notify:
             log.warning("Sending failure notifications.")
             await notifier.notify_after_fill(
                 False,
@@ -568,7 +586,7 @@ async def ln2(
         handler.failed = True
 
         error = err
-        if with_traceback:
+        if config.with_traceback:
             raise
 
         raise typer.Exit(1)
@@ -624,7 +642,9 @@ async def ln2(
                     "error": str(error) if error is not None else None,
                     "action": action.value,
                     "log_file": str(config.log_path) if config.log_path else None,
-                    "json_file": str(json_path) if json_path and write_json else None,
+                    "json_file": str(json_path)
+                    if json_path and config.write_json
+                    else None,
                     "log_data": log_data,
                     "configuration": configuration_json,
                     "valve_times": handler.get_valve_times(as_string=True),
@@ -689,6 +709,21 @@ async def _close_valves_helper():
     return typer.Exit(0)
 
 
+@cli.command("list-profiles")
+def list_profiles():
+    """Lists the available profiles."""
+
+    from lvmcryo.config import get_internal_config
+
+    internal_config = get_internal_config()
+    profiles = internal_config["profiles"]
+
+    for profile in profiles:
+        info_console.print(profile, style="bold")
+        info_console.print(profiles[profile])
+        print()
+
+
 @cli.command("close-valves")
 @cli_coro()
 async def close_valves():

src/lvmcryo/config.py

@@ -16,6 +16,7 @@
 
 from typing import Annotated, Any, Self
 
+from click.core import ParameterSource
 from pydantic import (
     BaseModel,
     Field,
@@ -104,6 +105,8 @@ class Config(BaseModel):
     interactive: InteractiveMode = InteractiveMode.auto
     no_prompt: bool = False
     dry_run: bool = False
+    clear_lock: bool = False
+    with_traceback: bool = False
 
     use_thermistors: bool = True
     require_all_thermistors: bool = False
@@ -141,6 +144,9 @@ class Config(BaseModel):
 
     error: Annotated[bool, ExcludedField] = False
 
+    profile: str | None = None
+    param_source: Annotated[dict[str, ParameterSource | None], ExcludedField] = {}
+
     def model_post_init(self, __context: Any) -> None:
         self._internal_config = get_internal_config(self.config_file)
         return super().model_post_init(__context)
@@ -176,7 +182,7 @@ def validate_interactive(cls, value: InteractiveMode) -> InteractiveMode:
 
     @model_validator(mode="before")
     @classmethod
-    def check_ommitted_fields(cls, data: Any) -> Any:
+    def before_validator(cls, data: Any) -> Any:
         if not isinstance(data, dict):
             # Not sure if this is likely to happen.
             return data
@@ -185,6 +191,17 @@ def check_ommitted_fields(cls, data: Any) -> Any:
         config = get_internal_config(data.get("config_file", None))
         defaults = config.get("defaults", {})
 
+        # If a profile has been passed, we update the input parameters with those
+        # in the profile. But we want to do that only for cases in which the user
+        # has not explicitly passed the parameter as a flag.
+        if (profile := data["profile"]) is not None:
+            profile_data = config.get("profiles", {}).get(profile, {})
+            param_source = data.get("param_source", {})
+            for key in profile_data:
+                psource = param_source.get(key, None)
+                if psource != ParameterSource.COMMANDLINE:
+                    data[key] = profile_data[key]
+
         # Use internal configuration files to fill in missing fields.
         for field in [
             "min_purge_time",

src/lvmcryo/config.yaml

@@ -10,6 +10,17 @@ defaults:
   log_path: '/data/logs/lvmcryo/{timestamp}.log'
   data_path: '/data/logs/lvmcryo/{timestamp}.parquet'
 
+profiles:
+  production:
+    clear_lock: true
+    interactive: 'no'
+    notify: true
+    write_log: true
+    write_data: true
+    write_json: true
+    email_level: info
+    data_extra_time: 1200
+
 notifications:
   slack_channel: lvm-notifications
   email_recipients: