feat: add Launcher class with V1/V2 protocol support

pyflowlauncher/__init__.py

@@ -4,6 +4,7 @@
 from .result import Result, send_results
 from .method import Method
 from .response import handle_response
+from .launcher import FlowLauncherV1, FlowLauncherV2
 
 
 logger = logging.getLogger(__name__)
@@ -15,4 +16,6 @@
     "Result",
     "Method",
     "handle_response",
+    "FlowLauncherV1",
+    "FlowLauncherV2",
 ]

pyflowlauncher/api.py

@@ -77,3 +77,21 @@ def open_url(url: str, in_private: bool = False) -> JsonRPCRequest:
 def open_uri(uri: str) -> JsonRPCRequest:
     """Open a URI."""
     return _send_action("OpenAppUri", uri)
+
+
+class Api:
+    """Flow Launcher API calls, accessible via plugin.launcher.api."""
+    change_query = staticmethod(change_query)
+    shell_run = staticmethod(shell_run)
+    close_app = staticmethod(close_app)
+    hide_app = staticmethod(hide_app)
+    show_app = staticmethod(show_app)
+    show_msg = staticmethod(show_msg)
+    open_setting_dialog = staticmethod(open_setting_dialog)
+    start_loading_bar = staticmethod(start_loading_bar)
+    stop_loading_bar = staticmethod(stop_loading_bar)
+    reload_plugins = staticmethod(reload_plugins)
+    copy_to_clipboard = staticmethod(copy_to_clipboard)
+    open_directory = staticmethod(open_directory)
+    open_url = staticmethod(open_url)
+    open_uri = staticmethod(open_uri)

pyflowlauncher/icons.py

@@ -21,6 +21,24 @@ def _get_icon(icon_name: str, file_ext: str = "png") -> Optional[str]:
     return None
 
 
+class Icons:
+    """Flow Launcher icon paths, accessible via launcher.icons.<name>."""
+    _IMAGE_DIR = "Images"
+
+    def __init__(self, program_dir: Optional[Path]) -> None:
+        self._dir = Path(program_dir) / self._IMAGE_DIR if program_dir else None
+
+    def _get(self, name: str, ext: str = "png") -> Optional[str]:
+        if self._dir is None:
+            return None
+        return str(self._dir / f"{name}.{ext}")
+
+    def __getattr__(self, name: str) -> Optional[str]:
+        if name.startswith('__') and name.endswith('__'):
+            raise AttributeError(name)
+        return self._get(name)
+
+
 ADMIN = _get_icon("admin")
 APP = _get_icon("app")
 APP_ERROR = _get_icon("app_error")

pyflowlauncher/launcher.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import sys
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Any, Awaitable, Callable, Optional
+
+from .api import Api
+from .base import pyFlowLauncherObject
+from .icons import Icons
+from .jsonrpc import JsonRPCClient
+
+
+class Launcher(pyFlowLauncherObject, ABC):
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._settings: dict = {}
+        self.api = Api()
+        self._program_dir: Optional[Path] = self._find_program_dir()
+        self.icons = Icons(self._program_dir)
+
+    @property
+    def settings(self) -> dict:
+        return self._settings
+
+    @property
+    def program_dir(self) -> Optional[Path]:
+        return self._program_dir
+
+    def _find_program_dir(self) -> Optional[Path]:
+        env = os.getenv("FLOW_PROGRAM_DIRECTORY")
+        if env:
+            return Path(env)
+        appdata = os.getenv("APPDATA")
+        if appdata:
+            candidate = Path(appdata) / "FlowLauncher"
+            if candidate.exists():
+                return candidate
+        local = os.getenv("LOCALAPPDATA")
+        if local:
+            candidate = Path(local) / "FlowLauncher"
+            if candidate.exists():
+                return candidate
+        current = Path.cwd()
+        for _ in range(5):
+            if (current / "Flow.Launcher.exe").exists():
+                return current
+            current = current.parent
+        self.logger.warning("Could not locate Flow Launcher program directory.")
+        return None
+
+    @abstractmethod
+    async def run(self, dispatch: Callable[[str, list], Awaitable[Any]]) -> None: ...
+
+
+class FlowLauncherV1(Launcher):
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._client = JsonRPCClient()
+
+    async def run(self, dispatch: Callable[[str, list], Awaitable[Any]]) -> None:
+        request = self._client.recieve()
+        self._settings = request.get('settings', {})
+        result = await dispatch(request['method'], request.get('parameters', []))
+        if result:
+            self._client.send(result)
+
+
+class FlowLauncherV2(Launcher):
+
+    _LIFECYCLE_METHODS = {'initialize', 'reload_data'}
+
+    async def run(self, dispatch: Callable[[str, list], Awaitable[Any]]) -> None:
+        loop = asyncio.get_event_loop()
+        while True:
+            line = await loop.run_in_executor(None, sys.stdin.readline)
+            if not line:
+                break
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                request = json.loads(line)
+            except json.JSONDecodeError:
+                self.logger.warning("Bad JSON-RPC line: %r", line)
+                continue
+
+            request_id = request.get('id')
+            method = request.get('method', '')
+            incoming_settings = request.get('settings')
+            if incoming_settings is not None:
+                self._settings = incoming_settings
+
+            if method == 'close':
+                self._send({'id': request_id, 'result': {}, 'error': None})
+                break
+
+            if method in self._LIFECYCLE_METHODS:
+                self._send({'id': request_id, 'result': {}, 'error': None})
+                continue
+
+            params = request.get('parameters', [])
+            if method == 'query' and params and isinstance(params[0], dict):
+                params = [params[0].get('search') or params[0].get('trimmedQuery', '')]
+
+            try:
+                result = await dispatch(method, params)
+            except Exception:
+                self.logger.exception("Unhandled error dispatching %r", method)
+                self._send({'id': request_id, 'result': [], 'error': None, 'debugMessage': ''})
+                continue
+
+            self._send_response(request_id, method, result)
+
+    def _send(self, data: dict) -> None:
+        sys.stdout.write(json.dumps(data) + '\n')
+        sys.stdout.flush()
+
+    def _send_response(self, request_id: Any, method: str, result: Any) -> None:
+        if result is None:
+            self._send({'id': request_id, 'result': {}, 'error': None})
+            return
+        if isinstance(result, dict) and 'Result' in result:
+            payload = {
+                'id': request_id,
+                'result': result['Result'],
+                'settingsChange': result.get('SettingsChange'),
+                'debugMessage': '',
+            }
+        else:
+            payload = {'id': request_id, 'hide': True}
+        self._send(payload)

pyflowlauncher/plugin.py

@@ -9,8 +9,9 @@
 from .base import pyFlowLauncherObject
 
 from .event import EventHandler
+from .launcher import Launcher, FlowLauncherV1, FlowLauncherV2
+from .jsonrpc import JsonRPCRequest
 from .response import handle_response
-from .jsonrpc import JsonRPCClient, JsonRPCRequest
 from .models.plugin_manifest import FILE_NAME
 from .manifest import Manifest
 
@@ -20,14 +21,21 @@
 
 class Plugin(pyFlowLauncherObject):
 
-    def __init__(self, methods: list[Method] | None = None) -> None:
+    def __init__(self, methods: list[Method] | None = None, launcher: Optional[Launcher] = None) -> None:
         super().__init__()
-        self._client = JsonRPCClient()
+        self._launcher: Launcher = launcher if launcher is not None else self._detect_launcher()
         self._event_handler = EventHandler()
-        self._settings: dict[str, Any] = {}
         if methods:
             self.add_methods(methods)
 
+    def _detect_launcher(self) -> Launcher:
+        try:
+            if (self.manifest.language or '').lower() == 'python_v2':
+                return FlowLauncherV2()
+        except FileNotFoundError:
+            pass
+        return FlowLauncherV1()
+
     def add_method(self, method: Method) -> str:
         """Add a method to the event handler."""
         @wraps(method)
@@ -65,25 +73,23 @@ def action(self, method: Method, parameters: Optional[List] = None) -> JsonRPCRe
         method_name = self.add_method(method)
         return {"method": method_name, "parameters": parameters or []}
 
+    @property
+    def launcher(self) -> Launcher:
+        return self._launcher
+
     @property
     def settings(self) -> dict:
-        if self._settings is None:
-            self._settings = {}
-        self._settings = self._client.recieve().get('settings', {})
-        return self._settings
+        return self._launcher.settings
 
     def run(self) -> None:
-        request = self._client.recieve()
-        method = request["method"]
-        parameters = request.get('parameters', [])
+        async def dispatch(method: str, parameters: list) -> Any:
+            return await self._event_handler.trigger_event(method, *parameters)
+
         if sys.version_info >= (3, 10, 0):
-            feedback = asyncio.run(self._event_handler.trigger_event(method, *parameters))
+            asyncio.run(self._launcher.run(dispatch))
         else:
             loop = asyncio.get_event_loop()
-            feedback = loop.run_until_complete(self._event_handler.trigger_event(method, *parameters))
-        if not feedback:
-            return
-        self._client.send(feedback)
+            loop.run_until_complete(self._launcher.run(dispatch))
 
     @property
     def run_dir(self) -> Path:

pyflowlauncher/response.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import inspect
 from typing import Any, Generator, Union