Merge branch 'main' into 595-configure-codeql-and-sonarcloud

.github/workflows/_test.yml

@@ -24,19 +24,7 @@ jobs:
  run:
  runs-on: ${{ inputs.runs-on }}
 
- services:
- activemq:
- image: rmohr/activemq:5.14.5-alpine
- ports:
- - 61613:61613
-
  steps:
- - name: Start RabbitMQ
- uses: namoshek/rabbitmq-github-action@v1
- with:
- ports: "61614:61613"
- plugins: rabbitmq_stomp
-
  - name: Checkout
  uses: actions/checkout@v4
  with:

dev-requirements.txt

@@ -17,6 +17,7 @@ bidict==0.23.1
 bluesky==1.13.0a4
 bluesky-kafka==0.10.0
 bluesky-live==0.0.8
+bluesky-stomp==0.1.0
 boltons==24.0.0
 cachetools==5.5.0
 caproto==1.1.1
@@ -212,10 +213,10 @@ tzlocal==5.2
 urllib3==2.2.2
 uvicorn==0.30.6
 virtualenv==20.26.3
-watchfiles==0.23.0
+watchfiles==0.24.0
 wcwidth==0.2.13
 websocket-client==1.8.0
-websockets==13.0
+websockets==13.0.1
 widgetsnbextension==4.0.13
 workflows==2.27
 xarray==2024.7.0

docs/reference/openapi.yaml

@@ -104,12 +104,10 @@ components:
  description: Request to change the state of the worker.
  properties:
  defer:
- anyOf:
- - type: boolean
- - type: 'null'
  default: false
  description: Should worker defer Pausing until the next checkpoint
  title: Defer
+ type: boolean
  new_state:
  $ref: '#/components/schemas/WorkerState'
  reason:

helm/blueapi/templates/deployment.yaml → helm/blueapi/templates/statefulset.yaml

@@ -50,7 +50,7 @@ spec:
  image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
  imagePullPolicy: {{ .Values.image.pullPolicy }}
  resources:
- {{- toYaml .Values.resources | nindent 12 }}
+ {{- .Values.initResources | default .Values.resources | toYaml | nindent 12 }}
  command: [/bin/sh, -c]
  args:
  - |

helm/blueapi/values.yaml

@@ -60,6 +60,11 @@ resources:
  # cpu: 100m
  # memory: 128Mi
 
+initResources:
+ {}
+ # Can optionally specify separate resource constraints for the scratch setup container.
+ # If left empty this defaults to the same as resources above.
+
 nodeSelector: {}
 
 tolerations: []

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
  "dls-dodal>=1.24.0",
  "super-state-machine", # See GH issue 553
  "GitPython",
+ "bluesky-stomp>=0.1.0"
 ]
 dynamic = ["version"]
 license.file = "LICENSE"
@@ -90,9 +91,6 @@ addopts = """
 filterwarnings = ["error", "ignore::DeprecationWarning"]
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs src tests"
-markers = [
- "handler: marks tests that interact with the global handler object in handler.py",
-]
 asyncio_mode = "auto"
 
 [tool.coverage.run]

src/blueapi/cli/cli.py

@@ -6,6 +6,8 @@
 
 import click
 from bluesky.callbacks.best_effort import BestEffortCallback
+from bluesky_stomp.messaging import MessageContext, MessagingTemplate
+from bluesky_stomp.models import Broker
 from pydantic import ValidationError
 from requests.exceptions import ConnectionError
 
@@ -16,8 +18,6 @@
 from blueapi.client.rest import BlueskyRemoteControlError
 from blueapi.config import ApplicationConfig, ConfigLoader
 from blueapi.core import DataEvent
-from blueapi.messaging import MessageContext
-from blueapi.messaging.stomptemplate import StompMessagingTemplate
 from blueapi.service.main import start
 from blueapi.service.openapi import (
  DOCS_SCHEMA_LOCATION,
@@ -147,14 +147,20 @@ def listen_to_events(obj: dict) -> None:
  config: ApplicationConfig = obj["config"]
  if config.stomp is not None:
  event_bus_client = EventBusClient(
- StompMessagingTemplate.autoconfigured(config.stomp)
+ MessagingTemplate.for_broker(
+ broker=Broker(
+ host=config.stomp.host,
+ port=config.stomp.port,
+ auth=config.stomp.auth,
+ )
+ )
  )
  else:
  raise RuntimeError("Message bus needs to be configured")
 
  def on_event(
- context: MessageContext,
  event: WorkerEvent | ProgressEvent | DataEvent,
+ context: MessageContext,
  ) -> None:
  converted = json.dumps(event.dict(), indent=2)
  print(converted)

src/blueapi/client/client.py

@@ -1,9 +1,11 @@
 import time
 from concurrent.futures import Future
 
+from bluesky_stomp.messaging import MessageContext, MessagingTemplate
+from bluesky_stomp.models import Broker
+
 from blueapi.config import ApplicationConfig
 from blueapi.core.bluesky_types import DataEvent
-from blueapi.messaging import MessageContext, StompMessagingTemplate
 from blueapi.service.model import (
  DeviceModel,
  DeviceResponse,
@@ -38,7 +40,13 @@ def __init__(
  def from_config(cls, config: ApplicationConfig) -> "BlueapiClient":
  rest = BlueapiRestClient(config.api)
  if config.stomp is not None:
- template = StompMessagingTemplate.autoconfigured(config.stomp)
+ template = MessagingTemplate.for_broker(
+ broker=Broker(
+ host=config.stomp.host,
+ port=config.stomp.port,
+ auth=config.stomp.auth,
+ )
+ )
  events = EventBusClient(template)
  else:
  events = None
@@ -178,7 +186,7 @@ def run_task(
 
  complete: Future[WorkerEvent] = Future()
 
- def inner_on_event(ctx: MessageContext, event: AnyEvent) -> None:
+ def inner_on_event(event: AnyEvent, ctx: MessageContext) -> None:
  match event:
  case WorkerEvent(task_status=TaskStatus(task_id=test_id)):
  relates_to_task = test_id == task_id

src/blueapi/client/event_bus.py

@@ -1,7 +1,9 @@
 from collections.abc import Callable
 
+from bluesky_stomp.messaging import MessageContext, MessagingTemplate
+from bluesky_stomp.models import MessageTopic
+
 from blueapi.core import DataEvent
-from blueapi.messaging import MessageContext, MessagingTemplate
 from blueapi.worker import ProgressEvent, WorkerEvent
 
 
@@ -28,11 +30,11 @@ def __exit__(self, exc_type, exc_value, exc_traceback) -> None:
 
  def subscribe_to_all_events(
  self,
- on_event: Callable[[MessageContext, AnyEvent], None],
+ on_event: Callable[[AnyEvent, MessageContext], None],
  ) -> None:
  try:
  self.app.subscribe(
- self.app.destinations.topic("public.worker.event"),
+ MessageTopic(name="public.worker.event"),
  on_event,
  )
  except Exception as err: