Run BlueskyContext and Worker in subprocess (#343)

Create SubprocessHandler as additional BlueskyHandler implementation to
forward calls to a subprocess. This is to allow a new REST endpoint with
the result of reloading the plans and devices without having to restart
the service.

Additionally the API version is incremented due to new schema and a
reference to delete_task replaced with clear_pending_task which was
missed in a previous PR.

Further improvements required in later PRs.

docs/user/reference/openapi.yaml

@@ -33,6 +33,18 @@ components:
       - devices
       title: DeviceResponse
       type: object
+    EnvironmentResponse:
+      additionalProperties: false
+      description: State of internal environment.
+      properties:
+        initialized:
+          description: blueapi context initialized
+          title: Initialized
+          type: boolean
+      required:
+      - initialized
+      title: EnvironmentResponse
+      type: object
     HTTPValidationError:
       properties:
         detail:
@@ -197,7 +209,7 @@ components:
       type: object
 info:
   title: BlueAPI Control
-  version: 0.0.4
+  version: 0.0.5
 openapi: 3.0.2
 paths:
   /devices:
@@ -237,6 +249,26 @@ paths:
                 $ref: '#/components/schemas/HTTPValidationError'
           description: Validation Error
       summary: Get Device By Name
+  /environment:
+    delete:
+      operationId: delete_environment_environment_delete
+      responses:
+        '200':
+          content:
+            application/json:
+              schema: {}
+          description: Successful Response
+      summary: Delete Environment
+    get:
+      operationId: get_environment_environment_get
+      responses:
+        '200':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/EnvironmentResponse'
+          description: Successful Response
+      summary: Get Environment
   /plans:
     get:
       description: Retrieve information about all available plans.

src/blueapi/cli/rest.py

@@ -72,7 +72,7 @@ def create_task(self, task: RunPlan) -> TaskResponse:
             data=task.dict(),
         )
 
-    def delete_task(self, task_id: str) -> TaskResponse:
+    def clear_pending_task(self, task_id: str) -> TaskResponse:
         return self._request_and_deserialize(
             f"/tasks/{task_id}", TaskResponse, method="DELETE"
         )

src/blueapi/service/handler.py

@@ -28,6 +28,7 @@ class Handler(BlueskyHandler):
     _worker: Worker
     _config: ApplicationConfig
     _messaging_template: MessagingTemplate
+    _initialized: bool = False
 
     def __init__(
         self,
@@ -64,6 +65,7 @@ def start(self) -> None:
         )
 
         self._messaging_template.connect()
+        self._initialized = True
 
     def _publish_event_streams(
         self, streams_to_destinations: Mapping[EventStream, str]
@@ -79,6 +81,7 @@ def _publish_event_stream(self, stream: EventStream, destination: str) -> None:
         )
 
     def stop(self) -> None:
+        self._initialized = False
         self._worker.stop()
         if self._messaging_template.is_connected():
             self._messaging_template.disconnect()
@@ -134,6 +137,10 @@ def pending_tasks(self) -> List[TrackableTask]:
     def get_pending_task(self, task_id: str) -> Optional[TrackableTask]:
         return self._worker.get_pending_task(task_id)
 
+    @property
+    def initialized(self) -> bool:
+        return self._initialized
+
 
 HANDLER: Optional[Handler] = None
 
@@ -145,7 +152,6 @@ def setup_handler(
 
     provider = None
     plan_wrappers = []
-
     if config:
         visit_service_client: VisitServiceClientBase
         if config.env.data_writing.visit_service_url is not None:

src/blueapi/service/handler_base.py

@@ -83,3 +83,21 @@ def pending_tasks(self) -> List[TrackableTask]:
     def get_pending_task(self, task_id: str) -> Optional[TrackableTask]:
         """Returns a task matching the task ID supplied,
         if the worker knows of it"""
+
+    @abstractmethod
+    def start(self):
+        """Start the handler"""
+
+    @abstractmethod
+    def stop(self):
+        """Stop the handler"""
+
+    @property
+    @abstractmethod
+    def initialized(self) -> bool:
+        """Handler initialization state"""
+
+
+class HandlerNotStartedError(Exception):
+    def __init__(self, message):
+        super().__init__(message)

src/blueapi/service/main.py

@@ -1,27 +1,61 @@
 from contextlib import asynccontextmanager
-from typing import Dict, Set
-
-from fastapi import Body, Depends, FastAPI, HTTPException, Request, Response, status
+from typing import Dict, Optional, Set
+
+from fastapi import (
+    BackgroundTasks,
+    Body,
+    Depends,
+    FastAPI,
+    HTTPException,
+    Request,
+    Response,
+    status,
+)
 from pydantic import ValidationError
 from starlette.responses import JSONResponse
 from super_state_machine.errors import TransitionError
 
 from blueapi.config import ApplicationConfig
 from blueapi.worker import RunPlan, TrackableTask, WorkerState
 
-from .handler import get_handler, setup_handler, teardown_handler
 from .handler_base import BlueskyHandler
 from .model import (
     DeviceModel,
     DeviceResponse,
+    EnvironmentResponse,
     PlanModel,
     PlanResponse,
     StateChangeRequest,
     TaskResponse,
     WorkerTask,
 )
+from .subprocess_handler import SubprocessHandler
+
+REST_API_VERSION = "0.0.5"
+
+HANDLER: Optional[BlueskyHandler] = None
+
+
+def get_handler() -> BlueskyHandler:
+    if HANDLER is None:
+        raise ValueError()
+    return HANDLER
+
+
+def setup_handler(config: Optional[ApplicationConfig] = None):
+    global HANDLER
+    handler = SubprocessHandler(config)
+    handler.start()
+
+    HANDLER = handler
 
-REST_API_VERSION = "0.0.4"
+
+def teardown_handler():
+    global HANDLER
+    if HANDLER is None:
+        return
+    HANDLER.stop()
+    HANDLER = None
 
 
 @asynccontextmanager
@@ -49,6 +83,26 @@ async def on_key_error_404(_: Request, __: KeyError):
     )
 
 
+@app.get("/environment", response_model=EnvironmentResponse)
+def get_environment(
+    handler: BlueskyHandler = Depends(get_handler),
+) -> EnvironmentResponse:
+    return EnvironmentResponse(initialized=handler.initialized)
+
+
+@app.delete("/environment")
+async def delete_environment(
+    background_tasks: BackgroundTasks,
+    handler: BlueskyHandler = Depends(get_handler),
+):
+    def restart_handler(handler: BlueskyHandler):
+        handler.stop()
+        handler.start()
+
+    if handler.initialized:
+        background_tasks.add_task(restart_handler, handler)
+
+
 @app.get("/plans", response_model=PlanResponse)
 def get_plans(handler: BlueskyHandler = Depends(get_handler)):
     """Retrieve information about all available plans."""

src/blueapi/service/model.py

@@ -128,3 +128,11 @@ class StateChangeRequest(BlueapiBaseModel):
         description="The reason for the current run to be aborted",
         default=None,
     )
+
+
+class EnvironmentResponse(BlueapiBaseModel):
+    """
+    State of internal environment.
+    """
+
+    initialized: bool = Field(description="blueapi context initialized")