wip

Author: CamDavidsonPilon

CHANGELOG.md

@@ -8,6 +8,7 @@
 
  - Added boot-partition Wi-Fi setup for _new_ Pioreactor images. Users can place a `wifi.ini` file with `ssid` and `passphrase` values on the boot partition, and the image will try those credentials before starting local access point mode.
  - Improved plugin install and uninstall reliability by moving the lifecycle into the Pioreactor package instead of requiring `/usr/local/bin` helper scripts on the operating system.
+ - Added support for installing `.py` drop-in plugins from Pioreactor-managed USB drives on the Plugins page, alongside existing `.whl` plugin installs.
  - Reduced idle CPU usage from Huey on single-core Raspberry Pi leaders by starting the background task queue with fewer workers and a slower polling interval. Multi-core devices keep the existing Huey settings.
  - Reduced idle CPU usage from the monitor job on single-core Raspberry Pis by disabling button controls by default. The new `[monitor.config] enable_button` option supports `auto`, `true`, and `false`.
  - removed tty services from base images.

core/pioreactor/cli/pio.py

@@ -19,9 +19,6 @@
 from pioreactor import exc
 from pioreactor import whoami
 from pioreactor.cli.lazy_group import LazyGroup
-from pioreactor.http_response import summarize_error_response
-from pioreactor.mureq import get
-from pioreactor.mureq import HTTPErrorStatus
 
 lazy_subcommands = {
     "run": "pioreactor.cli.run.run",
@@ -929,6 +926,8 @@ def blink() -> None:
     """
     monitor job is required to be running.
     """
+    from pioreactor.http_response import summarize_error_response
+    from pioreactor.mureq import HTTPErrorStatus
     from pioreactor.pubsub import post_into_leader
 
     response = post_into_leader(f"/api/workers/{whoami.get_unit_name()}/blink")
@@ -1753,6 +1752,7 @@ def get_non_prerelease_tags_of_pioreactor(repo: str) -> list[str]:
     Returns a list of all the tag names associated with non-prerelease releases, sorted in descending order
     """
     from packaging.version import Version
+    from pioreactor.mureq import get
 
     url = f"https://api.github.com/repos/{repo}/releases"
     headers = {"Accept": "application/vnd.github.v3+json"}

core/pioreactor/logging.py

@@ -119,6 +119,7 @@ def __init__(
         self,
         topic_prefix: str,
         client: "Client",
+        owns_client: bool,
         qos: int = 0,
         retain: bool = False,
         **mqtt_kwargs: Any,
@@ -129,6 +130,7 @@ def __init__(
         self.retain = retain
         self.mqtt_kwargs = mqtt_kwargs
         self.client = client
+        self.owns_client = owns_client
 
     def emit(self, record: logging.LogRecord) -> None:
         payload = self.format(record)
@@ -152,7 +154,8 @@ def emit(self, record: logging.LogRecord) -> None:
         mqtt_msg.wait_for_publish(timeout=2)
 
     def close(self) -> None:
-        self.client.shutdown()
+        if self.owns_client:
+            self.client.shutdown()
         super().close()
 
 
@@ -230,6 +233,7 @@ def create_logger(
     logger.addHandler(console_handler)
 
     if to_mqtt:
+        mqtt_handler_owns_client = pub_client is None
         if pub_client is None:
             pub_client = create_client(
                 client_id=f"{name}-logging-{unit}-{experiment}",
@@ -242,7 +246,7 @@ def create_logger(
         topic_prefix = (
             f"pioreactor/{unit}/{experiment}/logs/{source}"  # NOTE: we later append the log-level, ex: /debug
         )
-        mqtt_to_db_handler = MQTTHandler(topic_prefix, pub_client)
+        mqtt_to_db_handler = MQTTHandler(topic_prefix, pub_client, owns_client=mqtt_handler_owns_client)
         mqtt_to_db_handler.setLevel(logging.DEBUG)
         mqtt_to_db_handler.setFormatter(CustomisedJSONFormatter())
 

core/pioreactor/plugin_management/__init__.py

@@ -1,21 +1,13 @@
 # -*- coding: utf-8 -*-
-import glob
 import importlib
 import importlib.metadata as entry_point
 import os
-from functools import cache
 from typing import Any
 
 import click
 from msgspec import Struct
-from pioreactor import pubsub
-from pioreactor.plugin_management.install_plugin import click_install_plugin
-from pioreactor.plugin_management.list_plugins import click_list_plugins
-from pioreactor.plugin_management.uninstall_plugin import click_uninstall_plugin
 from pioreactor.plugin_management.utils import discover_plugins_in_entry_points
 from pioreactor.plugin_management.utils import discover_plugins_in_local_folder
-from pioreactor.utils import networking
-from pioreactor.whoami import get_unit_name
 
 """
 How do plugins work? There are a few patterns we use to "register" plugins with the core app.
@@ -68,9 +60,31 @@ class Plugin(Struct):
 ]
 
 
+def __getattr__(attr: str) -> Any:
+    if attr == "click_install_plugin":
+        from pioreactor.plugin_management.install_plugin import click_install_plugin
+
+        return click_install_plugin
+    elif attr == "click_list_plugins":
+        from pioreactor.plugin_management.list_plugins import click_list_plugins
+
+        return click_list_plugins
+    elif attr == "click_uninstall_plugin":
+        from pioreactor.plugin_management.uninstall_plugin import click_uninstall_plugin
+
+        return click_uninstall_plugin
+    raise AttributeError(attr)
+
+
 def get_plugin_api_url(py_file: str) -> str:
+    from pioreactor.config import config
+    from pioreactor.utils.networking import resolve_to_address
+    from pioreactor.whoami import get_unit_name
+
     endpoint = f"/unit_api/plugins/installed/{py_file}"
-    return pubsub.create_webserver_path(networking.resolve_to_address(get_unit_name()), endpoint)
+    port = config.getint("ui", "port", fallback=80)
+    proto = config.get("ui", "proto", fallback="http")
+    return f"{proto}://{resolve_to_address(get_unit_name())}:{port}/{endpoint.removeprefix('/')}"
 
 
 def get_plugins() -> dict[str, Plugin]:
@@ -89,7 +103,6 @@ def get_plugins() -> dict[str, Plugin]:
     # get entry point plugins
     # Users can use Python's entry point system to create rich plugins, see
     # example here: https://github.com/Pioreactor/pioreactor-air-bubbler
-
     for plugin in discover_plugins_in_entry_points():
         try:
             md = entry_point.metadata(plugin.name)
@@ -120,11 +133,12 @@ def get_plugins() -> dict[str, Plugin]:
         try:
             module = importlib.import_module(module_name)
             plugin_name = getattr(module, "__plugin_name__", module_name)
+            plugin_homepage = getattr(module, "__plugin_homepage__", None)
             plugins[plugin_name] = Plugin(
                 module,
                 getattr(module, "__plugin_summary__", BLANK),
                 getattr(module, "__plugin_version__", BLANK),
-                getattr(module, "__plugin_homepage__", get_plugin_api_url(py_file.name)),
+                plugin_homepage if plugin_homepage is not None else get_plugin_api_url(py_file.name),
                 getattr(module, "__plugin_author__", BLANK),
                 f"plugins/{py_file.name}",
             )

core/pioreactor/plugin_management/utils.py

@@ -33,5 +33,4 @@ def discover_plugins_in_local_folder() -> list[Path]:
 
 
 def discover_plugins_in_entry_points() -> list[entry_point.EntryPoint]:
-    eps = entry_point.entry_points()
-    return list(eps.select(group="pioreactor.plugins"))
+    return list(entry_point.entry_points(group="pioreactor.plugins"))

core/pioreactor/pubsub.py

@@ -9,18 +9,21 @@
 from time import sleep
 from typing import Any
 from typing import Callable
+from typing import TYPE_CHECKING
 
 from msgspec import Struct
 from msgspec.json import decode as loads
 from paho.mqtt.client import Client as PahoClient
 from paho.mqtt.enums import CallbackAPIVersion
 from paho.mqtt.enums import MQTTErrorCode
-from pioreactor import mureq
 from pioreactor import types as pt
 from pioreactor.config import config
 from pioreactor.config import leader_address
 from pioreactor.config import mqtt_address
 
+if TYPE_CHECKING:
+    from pioreactor import mureq
+
 
 def add_hash_suffix(s: str) -> str:
     """Adds random 4-character hash to the end of a string.
@@ -428,12 +431,14 @@ def create_webserver_path(address: str, endpoint: str) -> str:
     return f"{proto}://{address}:{port}/{endpoint}"
 
 
-def get_from(address: str, endpoint: str, **kwargs: Any) -> mureq.Response:
+def get_from(address: str, endpoint: str, **kwargs: Any) -> "mureq.Response":
     # pioreactor cluster specific
+    from pioreactor import mureq
+
     return mureq.get(create_webserver_path(address, endpoint), **kwargs)
 
 
-def get_from_leader(endpoint: str, timeout: int = 5, **kwargs: Any) -> mureq.Response:
+def get_from_leader(endpoint: str, timeout: int = 5, **kwargs: Any) -> "mureq.Response":
     return get_from(leader_address, endpoint, timeout=timeout, **kwargs)
 
 
@@ -443,8 +448,10 @@ def put_into(
     body: bytes | None = None,
     json: dict[str, Any] | Struct | None = None,
     **kwargs: Any,
-) -> mureq.Response:
+) -> "mureq.Response":
     # pioreactor cluster specific
+    from pioreactor import mureq
+
     return mureq.put(create_webserver_path(address, endpoint), body=body, json=json, **kwargs)
 
 
@@ -454,7 +461,7 @@ def put_into_leader(
     json: dict[str, Any] | Struct | None = None,
     timeout: int = 5,
     **kwargs: Any,
-) -> mureq.Response:
+) -> "mureq.Response":
     return put_into(leader_address, endpoint, body=body, json=json, timeout=timeout, **kwargs)
 
 
@@ -464,8 +471,10 @@ def patch_into(
     body: bytes | None = None,
     json: dict[str, Any] | Struct | None = None,
     **kwargs: Any,
-) -> mureq.Response:
+) -> "mureq.Response":
     # pioreactor cluster specific
+    from pioreactor import mureq
+
     return mureq.patch(create_webserver_path(address, endpoint), body=body, json=json, **kwargs)
 
 
@@ -475,7 +484,7 @@ def patch_into_leader(
     json: dict[str, Any] | Struct | None = None,
     timeout: int = 5,
     **kwargs: Any,
-) -> mureq.Response:
+) -> "mureq.Response":
     return patch_into(leader_address, endpoint, body=body, json=json, timeout=timeout, **kwargs)
 
 
@@ -485,8 +494,10 @@ def post_into(
     body: bytes | None = None,
     json: dict[str, Any] | Struct | None = None,
     **kwargs: Any,
-) -> mureq.Response:
+) -> "mureq.Response":
     # pioreactor cluster specific
+    from pioreactor import mureq
+
     return mureq.post(create_webserver_path(address, endpoint), body=body, json=json, **kwargs)
 
 
@@ -496,14 +507,16 @@ def post_into_leader(
     json: dict[str, Any] | Struct | None = None,
     timeout: int = 5,
     **kwargs: Any,
-) -> mureq.Response:
+) -> "mureq.Response":
     return post_into(leader_address, endpoint, body=body, json=json, timeout=timeout, **kwargs)
 
 
-def delete_from(address: str, endpoint: str, **kwargs: Any) -> mureq.Response:
+def delete_from(address: str, endpoint: str, **kwargs: Any) -> "mureq.Response":
     # pioreactor cluster specific
+    from pioreactor import mureq
+
     return mureq.delete(create_webserver_path(address, endpoint), **kwargs)
 
 
-def delete_from_leader(endpoint: str, timeout: int = 5, **kwargs: Any) -> mureq.Response:
+def delete_from_leader(endpoint: str, timeout: int = 5, **kwargs: Any) -> "mureq.Response":
     return delete_from(leader_address, endpoint, timeout=timeout, **kwargs)

core/pioreactor/structs.py

@@ -726,6 +726,10 @@ class InstallPluginFromUsbRequest(Struct, forbid_unknown_fields=True):
     filepath: str
 
 
+class InstallPluginFromLeaderCopyRequest(Struct, forbid_unknown_fields=True):
+    filename: str
+
+
 class CreateCalibrationRequest(Struct, forbid_unknown_fields=True):
     calibration_data: str
     set_as_active: bool = False

core/pioreactor/utils/usb.py

@@ -64,19 +64,20 @@ def as_dict(self) -> dict[str, str]:
         return {"path": str(self.path), "version": self.version}
 
 
-class UsbPluginWheel(msgspec.Struct, frozen=True):
+class UsbPluginArtifact(msgspec.Struct, frozen=True):
     path: Path
     name: str
     version: str | None
+    kind: str
 
     def as_dict(self) -> dict[str, str | None]:
-        return {"path": str(self.path), "name": self.name, "version": self.version}
+        return {"path": str(self.path), "name": self.name, "version": self.version, "kind": self.kind}
 
 
 class UsbScan(msgspec.Struct, frozen=True):
     mountpoint: Path
     updates: tuple[UsbUpdateArchive, ...]
-    plugins: tuple[UsbPluginWheel, ...]
+    plugins: tuple[UsbPluginArtifact, ...]
     writable: bool
     free_bytes: int
 
@@ -208,8 +209,8 @@ def scan_usb_mount(mountpoint: Path) -> UsbScan:
         )
     )
     plugins = tuple(
-        UsbPluginWheel(path=path, name=name, version=version)
-        for path, name, version in _find_plugin_wheels(mountpoint)
+        UsbPluginArtifact(path=path, name=name, version=version, kind=kind)
+        for path, name, version, kind in _find_plugin_artifacts(mountpoint)
     )
     usage = shutil.disk_usage(mountpoint)
     return UsbScan(
@@ -310,7 +311,17 @@ def choose_usb_mountpoint(mountpoint: str | None = None) -> Path:
 def resolve_usb_plugin_wheel(filepath: str) -> Path:
     path = Path(filepath)
     if path.suffix != ".whl":
-        raise ValueError("USB plugin installs currently support .whl files only.")
+        raise ValueError("USB wheel plugin installs support .whl files only.")
+
+    return resolve_usb_plugin_artifact(filepath)
+
+
+def resolve_usb_plugin_artifact(filepath: str) -> Path:
+    path = Path(filepath)
+    if path.suffix not in (".whl", ".py"):
+        raise ValueError("USB plugin installs currently support .whl and .py files only.")
+    if path.suffix == ".py" and not is_valid_python_plugin_filename(path.name):
+        raise ValueError(f"{path.name} is not a valid Python plugin filename.")
     if not path.exists():
         raise ValueError(f"{filepath} does not exist.")
 
@@ -451,20 +462,23 @@ def _pioreactor_managed_mountpoints(partition: UsbPartition) -> list[Path]:
     ]
 
 
-def _find_plugin_wheels(mountpoint: Path) -> list[tuple[Path, str, str | None]]:
+def _find_plugin_artifacts(mountpoint: Path) -> list[tuple[Path, str, str | None, str]]:
     candidates = [mountpoint, mountpoint / "pioreactor" / "plugins"]
     seen: set[Path] = set()
-    plugins: list[tuple[Path, str, str | None]] = []
+    plugins: list[tuple[Path, str, str | None, str]] = []
 
     for directory in candidates:
         if not directory.exists():
             continue
-        for path in sorted(directory.glob("*.whl")):
-            if path in seen:
+        for path in sorted(directory.iterdir()):
+            if path in seen or path.suffix not in (".whl", ".py"):
                 continue
             seen.add(path)
-            name, version = parse_wheel_name(path.name)
-            plugins.append((path, name, version))
+            if path.suffix == ".whl":
+                name, version = parse_wheel_name(path.name)
+                plugins.append((path, name, version, "wheel"))
+            elif is_valid_python_plugin_filename(path.name):
+                plugins.append((path, path.stem, None, "python_file"))
 
     return plugins
 
@@ -476,6 +490,11 @@ def parse_wheel_name(filename: str) -> tuple[str, str | None]:
     return parts[0].replace("_", "-"), parts[1]
 
 
+def is_valid_python_plugin_filename(filename: str) -> bool:
+    path = Path(filename)
+    return path.name == filename and path.suffix == ".py" and path.stem.isidentifier()
+
+
 def _as_bool(value: object) -> bool:
     return value is True or value == 1 or value == "1"
 

core/pioreactor/web/api.py

@@ -2493,7 +2493,7 @@ def install_plugin_across_cluster(pioreactor_unit: str) -> DelayedResponseReturn
 @api_bp.route("/units/<pioreactor_unit>/plugins/install-from-leader-usb", methods=["POST", "PATCH"])
 def install_plugin_from_leader_usb_on_machine(pioreactor_unit: str) -> DelayedResponseReturnValue:
     """
-    Install one wheel plugin from the leader's Pioreactor-managed USB mount onto selected unit(s).
+    Install one plugin artifact from the leader's Pioreactor-managed USB mount onto selected unit(s).
 
     JSON body:
     {