2.8.0 - Channel environment command processor

.github/workflows/python-package.yml

@@ -18,7 +18,6 @@ jobs:
     strategy:
       matrix:
         python-version:
-          - "3.10"
           - "3.11"
         system:
           - ubuntu-latest
@@ -37,7 +36,7 @@ jobs:
           python-version: ${{matrix.python-version}}
           cache: pip
 
-      - run: pip${{matrix.python-version}} install vmklib>=1.8.0
+      - run: pip${{matrix.python-version}} install vmklib
 
       - run: mk python-sa-types
 
@@ -68,7 +67,7 @@ jobs:
 
       - run: |
           mk python-release owner=vkottler \
-                            repo=runtimepy version=2.7.1
+                            repo=runtimepy version=2.8.0
         if: |
           matrix.python-version == '3.11'
           && matrix.system == 'ubuntu-latest'

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.3
-    hash=a72da168f214fb625163824388549911
+    hash=86fa83db6bac485c345afa670f2b159d
     =====================================
 -->
 
-# runtimepy ([2.7.1](https://pypi.org/project/runtimepy/))
+# runtimepy ([2.8.0](https://pypi.org/project/runtimepy/))
 
 [![python](https://img.shields.io/pypi/pyversions/runtimepy.svg)](https://pypi.org/project/runtimepy/)
 ![Build Status](https://github.com/vkottler/runtimepy/workflows/Python%20Package/badge.svg)
@@ -29,7 +29,6 @@
 
 This package is tested with the following Python minor versions:
 
-* [`python3.10`](https://docs.python.org/3.10/)
 * [`python3.11`](https://docs.python.org/3.11/)
 
 ## Platform Support

local/configs/python.yaml

@@ -3,7 +3,7 @@ author_info:
   name: Vaughn Kottler
   email: [email protected]
   username: vkottler
-versions: ["3.10", "3.11"]
+versions: ["3.11"]
 
 systems:
   - macos-latest

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
 major: 2
-minor: 7
-patch: 1
+minor: 8
+patch: 0
 entry: runtimepy

pyproject.toml

@@ -4,18 +4,17 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "runtimepy"
-version = "2.7.1"
+version = "2.8.0"
 description = "A framework for implementing Python services."
 readme = "README.md"
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 authors = [
   {name = "Vaughn Kottler", email = "[email protected]"}
 ]
 maintainers = [
   {name = "Vaughn Kottler", email = "[email protected]"}
 ]
 classifiers = [
-  "Programming Language :: Python :: 3.10",
   "Programming Language :: Python :: 3.11",
   "Operating System :: Microsoft :: Windows",
   "Operating System :: MacOS",

runtimepy/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.3
-# hash=962f3227b7f0033ff39fbfa1a78060f5
+# hash=6ef03f741188d3cc145268db1edd0d8d
 # =====================================
 
 """
@@ -10,7 +10,7 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "2.7.1"
+VERSION = "2.8.0"
 
 # runtimepy-specific content.
 METRICS_NAME = "metrics"

runtimepy/channel/environment/base.py

@@ -106,21 +106,35 @@ def set(self, key: _RegistryKey, value: ChannelValue) -> None:
         if isinstance(value, str):
             # Ensure that the channel has an associated enumeration.
             if enum is None:
-                raise ValueError(
-                    (
-                        f"Can't assign '{value}' to channel "
-                        f"'{self.channels.names.name(key)}'!"
+                resolved = False
+
+                is_int = chan.raw.kind.is_integer
+
+                if is_int or chan.raw.kind.is_float:
+                    kind = int if is_int else float
+                    try:
+                        value = kind(value)
+                        resolved = True
+                    except ValueError:
+                        pass
+
+                if not resolved:
+                    raise ValueError(
+                        (
+                            f"Can't assign '{value}' to channel "
+                            f"'{self.channels.names.name(key)}'!"
+                        )
                     )
-                )
 
-            value = (
-                enum.get_bool(value)
-                if chan.type.is_boolean
-                else enum.get_int(value)
-            )
+            else:
+                value = (
+                    enum.get_bool(value)
+                    if chan.type.is_boolean
+                    else enum.get_int(value)
+                )
 
         # Assign the value to the channel.
-        chan.raw.value = value
+        chan.raw.value = value  # type: ignore
 
     def apply(self, values: ValueMap) -> None:
         """Apply a map of values to the environment."""

runtimepy/channel/environment/command/__init__.py

@@ -0,0 +1,143 @@
+"""
+A module implementing UI command processing.
+"""
+
+# built-in
+from argparse import Namespace
+from typing import Any, Optional, cast
+
+# third-party
+from vcorelib.logging import LoggerType
+
+# internal
+from runtimepy.channel import AnyChannel
+from runtimepy.channel.environment import ChannelEnvironment
+from runtimepy.channel.environment.command.parser import (
+    ChannelCommand,
+    CommandParser,
+)
+from runtimepy.channel.environment.command.result import SUCCESS, CommandResult
+from runtimepy.mixins.environment import ChannelEnvironmentMixin
+from runtimepy.primitives.bool import Bool
+
+
+class ChannelCommandProcessor(ChannelEnvironmentMixin):
+    """A command processing interface for channel environments."""
+
+    def __init__(
+        self, env: ChannelEnvironment, logger: LoggerType, **kwargs
+    ) -> None:
+        """Initialize this instance."""
+
+        super().__init__(env=env, **kwargs)
+        self.logger = logger
+
+        self.parser_data: dict[str, Any] = {}
+        self.parser = CommandParser()
+        self.parser.data = self.parser_data
+
+        self.parser.initialize()
+
+    def get_suggestion(self, value: str) -> Optional[str]:
+        """Get an input suggestion."""
+
+        result = None
+
+        args = self.parse(value)
+        if args is not None:
+            result = self.env.namespace_suggest(args.channel, delta=False)
+            if result is not None:
+                result = args.command + " " + result
+
+        return result
+
+    def do_set(self, args: Namespace) -> CommandResult:
+        """Attempt to set a channel value."""
+
+        result = SUCCESS
+
+        if not args.extra:
+            return CommandResult(False, "No value specified.")
+
+        try:
+            self.env.set(args.channel, args.extra[0])
+        except (ValueError, KeyError) as exc:
+            result = CommandResult(False, str(exc))
+
+        return result
+
+    def do_toggle(self, args: Namespace, channel: AnyChannel) -> CommandResult:
+        """Attempt to toggle a channel."""
+
+        if args.command == ChannelCommand.TOGGLE:
+            if not channel.type.is_boolean:
+                return CommandResult(
+                    False,
+                    (
+                        f"Channel '{args.channel}' is "
+                        f"{channel.type}, not boolean."
+                    ),
+                )
+
+            cast(Bool, channel.raw).toggle()
+
+        return SUCCESS
+
+    def handle_command(self, args: Namespace) -> CommandResult:
+        """Handle a command from parsed arguments."""
+
+        result = SUCCESS
+
+        chan = self.env.get(args.channel)
+        if chan is None:
+            return CommandResult(False, f"No channel '{args.channel}'.")
+
+        channel, enum = chan
+        del enum
+
+        # Check if channel is commandable (or if a -f/--force flag is set?).
+        if not channel.commandable and not args.force:
+            return CommandResult(
+                False,
+                (
+                    f"Channel '{args.channel}' not "
+                    "commandable! (use -f/--force to bypass if necessary)"
+                ),
+            )
+
+        if args.command == ChannelCommand.TOGGLE:
+            result = self.do_toggle(args, channel)
+        elif args.command == ChannelCommand.SET:
+            result = self.do_set(args)
+
+        return result
+
+    def parse(self, value: str) -> Optional[Namespace]:
+        """Attempt to parse arguments."""
+
+        self.parser_data["error_message"] = None
+        args = self.parser.parse_args(value.split())
+        return args if not self.parser_data["error_message"] else None
+
+    def command(self, value: str) -> CommandResult:
+        """Process a command."""
+
+        args = self.parse(value)
+        success = args is not None
+
+        if not args or "help" in value:
+            self.logger.info(self.parser.format_help())
+
+        reason = None
+        if not success:
+            reason = self.parser_data["error_message"]
+            if "help" not in value:
+                self.logger.info("Try 'help'.")
+
+        result = CommandResult(success, reason)
+
+        if success:
+            assert args is not None
+            result = self.handle_command(args)
+
+        return result

runtimepy/channel/environment/command/parser.py

@@ -0,0 +1,58 @@
+"""
+A module implementing a channel-environment command argument parser wrapper.
+"""
+
+# built-in
+from argparse import ArgumentParser
+from enum import StrEnum
+from typing import Any
+
+
+class ChannelCommand(StrEnum):
+    """An enumeration for all channel command options."""
+
+    SET = "set"
+    TOGGLE = "toggle"
+
+
+class CommandParser(ArgumentParser):
+    """An argument parser wrapper."""
+
+    data: dict[str, Any]
+
+    def error(self, message: str):
+        """Pass error message to error handling."""
+        self.data["error_message"] = message
+
+    def exit(self, status: int = 0, message: str = None):
+        """Override exit behavior."""
+
+        del status
+
+        if message:
+            curr = self.data.get("error_message")
+            curr = curr + f" [{message}]" if curr else message
+            self.data["error_message"] = curr
+
+    def initialize(self) -> None:
+        """Initialize this command parser."""
+
+        self.add_argument(
+            "command",
+            type=ChannelCommand,
+            choices=set(ChannelCommand),
+            help="command to run",
+        )
+
+        self.add_argument(
+            "-f",
+            "--force",
+            action="store_true",
+            help="operate on a channel even if it's not commandable",
+        )
+
+        self.add_argument(
+            "channel", type=str, help="channel to perform action on"
+        )
+
+        self.add_argument("extra", nargs="*")

runtimepy/channel/environment/command/result.py

@@ -0,0 +1,30 @@
+"""
+A module implementing a command result interface.
+"""
+
+# built-in
+from typing import NamedTuple, Optional
+
+
+class CommandResult(NamedTuple):
+    """A container for command result data."""
+
+    success: bool
+    reason: Optional[str] = None
+
+    def __bool__(self) -> bool:
+        """Evaluate this instance as a boolean."""
+        return self.success
+
+    def __str__(self) -> str:
+        """Get this command result as a string."""
+
+        message = "(success)" if self.success else "(failure)"
+
+        if self.reason:
+            message += " " + self.reason
+
+        return message
+
+
+SUCCESS = CommandResult(True)