3.4.1 - Improve channel command suggestions

.github/workflows/python-package.yml

@@ -68,7 +68,7 @@ jobs:
 
       - run: |
           mk python-release owner=vkottler \
-                            repo=runtimepy version=3.4.0
+                            repo=runtimepy version=3.4.1
         if: |
           matrix.python-version == '3.11'
           && matrix.system == 'ubuntu-latest'

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.4
-    hash=c6a4e8d162ebe8d48c6bce5ffcd9a663
+    hash=e249f3854eb9586ba137f0d4cb7089b9
     =====================================
 -->
 
-# runtimepy ([3.4.0](https://pypi.org/project/runtimepy/))
+# runtimepy ([3.4.1](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)
@@ -126,7 +126,7 @@ options:
 ```
 $ ./venv3.11/bin/runtimepy task -h
 
-usage: runtimepy task [-h] [-i] [-w] factory [configs ...]
+usage: runtimepy task [-h] [-i] [-w] [-r RATE] factory [configs ...]
 
 positional arguments:
   factory               name of task factory to create task with
@@ -139,6 +139,7 @@ options:
   -w, --wait-for-stop, --wait_for_stop
                         ensure that a 'wait_for_stop' application method is
                         run last
+  -r RATE, --rate RATE  rate (in Hz) that the task should run (default: 10)
 
 ```
 

local/configs/package.yaml

@@ -5,7 +5,7 @@ description: A framework for implementing Python services.
 entry: {{entry}}
 
 requirements:
-  - vcorelib>=2.8.3
+  - vcorelib>=3.2.0
   - websockets
   - "windows-curses; sys_platform == 'win32' and python_version < '3.12'"
 

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
 major: 3
 minor: 4
-patch: 0
+patch: 1
 entry: runtimepy

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "runtimepy"
-version = "3.4.0"
+version = "3.4.1"
 description = "A framework for implementing Python services."
 readme = "README.md"
 requires-python = ">=3.11"

runtimepy/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.4
-# hash=3ac128eb7b7140a6e7e096f51a3b3e43
+# hash=4df0a252c47915951eb613d1f294ee92
 # =====================================
 
 """
@@ -10,7 +10,7 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "3.4.0"
+VERSION = "3.4.1"
 
 # runtimepy-specific content.
 METRICS_NAME = "metrics"

runtimepy/channel/environment/base.py

@@ -40,6 +40,8 @@
 BoolChannelResult = _Tuple[_BoolChannel, _Optional[_RuntimeEnum]]
 IntChannelResult = _Tuple[_IntChannel, _Optional[_RuntimeEnum]]
 
+FieldOrChannel = _Union[_BitField, _AnyChannel]
+
 
 class BaseChannelEnvironment(_NamespaceMixin, FinalizeMixin):
     """A class integrating channel and enumeration registries."""
@@ -184,6 +186,22 @@ def exists(self, val: _RegistryKey) -> bool:
         """Determine if a channel exists."""
         return self.fields.has_field(val) or self.get(val) is not None
 
+    def field_or_channel(self, val: _RegistryKey) -> _Optional[FieldOrChannel]:
+        """Attempt to look up a field or channel for a given registry key."""
+
+        channel: _Optional[FieldOrChannel] = None
+
+        chan = self.get(val)
+        if chan is None:
+            # Check if the name is a field.
+            field = self.fields.get_field(val)
+            if field is not None:
+                channel = field
+        else:
+            channel, _ = chan
+
+        return channel
+
     def get(self, val: _RegistryKey) -> _Optional[ChannelResult]:
         """Attempt to get a channel and its enumeration (if it has one)."""
 

runtimepy/channel/environment/command/__init__.py

@@ -4,14 +4,14 @@
 
 # built-in
 from argparse import Namespace
-from typing import Any, Callable, Optional, Union, cast
+from typing import Any, Callable, 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.base import FieldOrChannel
 from runtimepy.channel.environment.command.parser import (
     ChannelCommand,
     CommandParser,
@@ -21,7 +21,6 @@
 from runtimepy.primitives.bool import Bool
 from runtimepy.primitives.field import BitField
 
-FieldOrChannel = Union[BitField, AnyChannel]
 CommandHook = Callable[[Namespace, Optional[FieldOrChannel]], None]
 
 
@@ -50,7 +49,19 @@ def get_suggestion(self, value: str) -> Optional[str]:
 
         args = self.parse(value)
         if args is not None:
-            result = self.env.namespace_suggest(args.channel, delta=False)
+            candidates = self.env.ns.length_sorted_suggestions(
+                args.channel, delta=False
+            )
+            if candidates:
+                result = candidates[0]
+
+                # Try to find a commandable suggestion.
+                for candidate in candidates:
+                    chan = self.env.field_or_channel(candidate)
+                    if chan is not None and chan.commandable:
+                        result = candidate
+                        break
+
             if result is not None:
                 result = args.command + " " + result
 
@@ -113,18 +124,9 @@ def handle_command(self, args: Namespace) -> CommandResult:
             if self.env.exists(args.channel):
                 return CommandResult(True, str(self.env.value(args.channel)))
 
-        chan = self.env.get(args.channel)
-
-        channel: FieldOrChannel
-
-        if chan is None:
-            # Check if the name is a field.
-            field = self.env.fields.get_field(args.channel)
-            if field is None:
-                return CommandResult(False, f"No channel '{args.channel}'.")
-            channel = field
-        else:
-            channel, _ = chan
+        channel = self.env.field_or_channel(args.channel)
+        if channel is None:
+            return CommandResult(False, f"No channel '{args.channel}'.")
 
         # Check if channel is commandable (or if a -f/--force flag is
         # set?).

runtimepy/requirements.txt

@@ -1,3 +1,3 @@
-vcorelib>=2.8.3
+vcorelib>=3.2.0
 websockets
 windows-curses; sys_platform == 'win32' and python_version < '3.12'

tasks/conf.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.4
-# hash=9f62028523c3b5a953733ca89dcc3018
+# hash=7d378a1752611508007a77d4ca39a5af
 # =====================================
 """
 A module for project-specific task registration.
@@ -20,14 +20,9 @@ def audit_local_tasks() -> None:
     """Ensure that shared task infrastructure is present."""
 
     local = Path(__file__).parent.joinpath("mklocal")
-
-    # Also link a top-level file.
     top_level = local.parent.parent.joinpath("mklocal")
-    if not top_level.is_symlink():
-        assert not top_level.exists()
-        top_level.symlink_to(local)
 
-    if local.is_symlink():
+    if local.is_symlink() and top_level.is_symlink():
         return
 
     # If it's not a symlink, it shouldn't be any other kind of file.
@@ -48,6 +43,11 @@ def audit_local_tasks() -> None:
     # Create the link.
     local.symlink_to(vmklib)
 
+    # Also link a top-level file.
+    if not top_level.is_symlink():
+        assert not top_level.exists()
+        top_level.symlink_to(local)
+
 
 def register(
     manager: TaskManager,