"""
The ska_oso_oet.operator.ui package contains the REST API for operator actions.
"""
from fastapi import APIRouter, HTTPException, Response
from pubsub import pub
from pydantic import BaseModel
from ska_aaa_authhelpers import Role
from ska_oso_oet.auth import (
OPERATOR_ROLE_FOR_TELESCOPE,
OST_CLIENT_ID,
Permissions,
Scopes,
)
from ska_oso_oet.event import topics
from ska_oso_oet.procedure.application import ScriptContext
from ska_oso_oet.utils.ui import call_and_respond
operator_actions_router = APIRouter(
prefix="/operator-actions", tags=["Operator Actions"]
)
VALID_ACTION_TOPICS = {
"operator.wait_for_qa_ready.override": topics.operator.wait_for_qa_ready.override,
"operator.wait_for_qa_ready.enable": topics.operator.wait_for_qa_ready.enable,
"operator.wait_for_qa_ready.disable": topics.operator.wait_for_qa_ready.disable,
}
[docs]
class OperatorActionRequest(BaseModel):
topic: str
@operator_actions_router.get(
"",
summary="Get current operator override state",
description=(
"Returns the current state of persistent operator overrides. "
"Scan-level overrides are not tracked."
),
dependencies=[
Permissions(
roles={Role.SW_ENGINEER, OPERATOR_ROLE_FOR_TELESCOPE, Role.APP2APP},
scopes={Scopes.ACTIVITY_READ},
app_ids={OST_CLIENT_ID},
)
],
)
def get_override_state() -> ScriptContext:
result = call_and_respond(
topics.request.operator.override.state,
topics.operator.override.state,
)
return result
@operator_actions_router.post(
"",
status_code=202,
summary="Send an operator override action",
description=(
"Accepts an operator action command and publishes it within the OET "
"as a pub/sub message. Returns 202 immediately; the action propagates "
"asynchronously to running scripts."
),
dependencies=[
Permissions(
roles={Role.SW_ENGINEER, OPERATOR_ROLE_FOR_TELESCOPE, Role.APP2APP},
scopes={Scopes.ACTIVITY_EXECUTE},
app_ids={OST_CLIENT_ID},
)
],
)
def post_operator_action(request_body: OperatorActionRequest) -> Response:
topic_class = VALID_ACTION_TOPICS.get(request_body.topic)
if topic_class is None:
raise HTTPException(
status_code=400,
detail=f"Unknown topic: {request_body.topic}",
)
pub.sendMessage(topic_class, msg_src="FastAPIWorker")
return Response(status_code=202)