app_deploy: streamline type and channel config

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>

moonraker/components/update_manager/app_deploy.py

@@ -35,19 +35,6 @@ if TYPE_CHECKING:
     from ..machine import Machine
     from ..file_manager.file_manager import FileManager
 
-SUPPORTED_CHANNELS = {
-    AppType.WEB: [Channel.STABLE, Channel.BETA],
-    AppType.ZIP: [Channel.STABLE, Channel.BETA],
-    AppType.GIT_REPO: list(Channel),
-    AppType.PYTHON: list(Channel)
-}
-TYPE_TO_CHANNEL = {
-    AppType.WEB: Channel.STABLE,
-    AppType.ZIP: Channel.STABLE,
-    AppType.GIT_REPO: Channel.DEV,
-    AppType.PYTHON: Channel.STABLE
-}
-
 DISTRO_ALIASES = [distro.id()]
 DISTRO_ALIASES.extend(distro.like().split())
 
@@ -57,23 +44,18 @@ class AppDeploy(BaseDeploy):
     ) -> None:
         super().__init__(config, cmd_helper, prefix=prefix)
         self.config = config
-        type_choices = list(TYPE_TO_CHANNEL.keys())
+        type_choices = {str(t): t for t in AppType.valid_types()}
-        self.type = AppType.from_string(config.get('type'))
+        self.type = config.getchoice("type", type_choices)
-        if self.type not in type_choices:
+        channel_choices = {str(chnl): chnl for chnl in list(Channel)}
-            str_types = [str(t) for t in type_choices]
+        self.channel = config.getchoice(
-            raise config.error(
+            "channel", channel_choices, str(self.type.default_channel)
-                f"Section [{config.get_name()}], Option 'type: {self.type}': "
-                f"value must be one of the following choices: {str_types}"
-            )
-        self.channel = Channel.from_string(
-            config.get("channel", str(TYPE_TO_CHANNEL[self.type]))
         )
         self.channel_invalid: bool = False
-        if self.channel not in SUPPORTED_CHANNELS[self.type]:
+        if self.channel not in self.type.supported_channels:
-            str_channels = [str(c) for c in SUPPORTED_CHANNELS[self.type]]
+            str_channels = [str(c) for c in self.type.supported_channels]
             self.channel_invalid = True
             invalid_channel = self.channel
-            self.channel = TYPE_TO_CHANNEL[self.type]
+            self.channel = self.type.default_channel
             self.server.add_warning(
                 f"[{config.get_name()}]: Invalid value '{invalid_channel}' for "
                 f"option 'channel'. Type '{self.type}' supports the following "

moonraker/components/update_manager/common.py

@@ -14,7 +14,8 @@ from ...utils import source_info
 from typing import (
     TYPE_CHECKING,
     Dict,
-    Union
+    Union,
+    List
 )
 
 if TYPE_CHECKING:
@@ -47,6 +48,8 @@ BASE_CONFIG: Dict[str, Dict[str, str]] = {
     }
 }
 
+OPTION_OVERRIDES = ("channel", "pinned_commit", "refresh_interval")
+
 class AppType(ExtendedEnum):
     NONE = 1
     WEB = 2
@@ -66,6 +69,26 @@ class AppType(ExtendedEnum):
         else:
             return AppType.NONE
 
+    @classmethod
+    def valid_types(cls) -> List[AppType]:
+        all_types = list(cls)
+        all_types.remove(AppType.NONE)
+        return all_types
+
+    @property
+    def supported_channels(self) -> List[Channel]:
+        if self == AppType.NONE:
+            return []
+        elif self in [AppType.WEB, AppType.ZIP]:
+            return [Channel.STABLE, Channel.BETA]  # type: ignore
+        else:
+            return list(Channel)
+
+    @property
+    def default_channel(self) -> Channel:
+        if self == AppType.GIT_REPO:
+            return Channel.DEV  # type: ignore
+        return Channel.STABLE  # type: ignore
 
 class Channel(ExtendedEnum):
     STABLE = 1
@@ -84,19 +107,16 @@ def get_base_configuration(config: ConfigHelper) -> ConfigHelper:
         "moonraker", "update_manager.klipper_exec", KLIPPER_DEFAULT_EXEC
     ).result()
     base_cfg["klipper"]["type"] = str(AppType.detect(base_cfg["klipper"]["path"]))
-    channel = config.get("channel", "dev")
+    default_channel = config.get("channel", None)
-    base_cfg["moonraker"]["channel"] = channel
+    # Check for configuration overrides
-    base_cfg["klipper"]["channel"] = channel
+    for app_name in base_cfg.keys():
-    if config.has_section("update_manager moonraker"):
+        if default_channel is not None:
-        mcfg = config["update_manager moonraker"]
+            base_cfg[app_name]["channel"] = default_channel
-        base_cfg["moonraker"]["channel"] = mcfg.get("channel", channel)
+        override_section = f"update_manager {app_name}"
-        commit = mcfg.get("pinned_commit", None)
+        if not config.has_section(override_section):
-        if commit is not None:
+            continue
-            base_cfg["moonraker"]["pinned_commit"] = commit
+        app_cfg = config[override_section]
-    if config.has_section("update_manager klipper"):
+        for opt in OPTION_OVERRIDES:
-        kcfg = config["update_manager klipper"]
+            if app_cfg.has_option(opt):
-        base_cfg["klipper"]["channel"] = kcfg.get("channel", channel)
+                base_cfg[app_name][opt] = app_cfg.get(opt)
-        commit = kcfg.get("pinned_commit", None)
-        if commit is not None:
-            base_cfg["klipper"]["pinned_commit"] = commit
     return config.read_supplemental_dict(base_cfg)