"""Execution engine deployment."""
import logging
from ska_sdp_config import Config
from ska_sdp_config.config import Transaction
LOG = logging.getLogger("ska_sdp_scripting")
[docs]
class EEDeploy:
"""
Base class for execution engine deployment.
:param pb_id: processing block ID
:type pb_id: str
:param config: SDP configuration client
:type config: ska_sdp_config.Client
"""
def __init__(self, pb_id: str, config: Config):
self._pb_id = pb_id
self._config = config
self._deploy_id = None
[docs]
def update_deploy_status(self, status: str) -> None:
"""
Update deployment status.
:param status: status
:type status: str
"""
LOG.info(
"Updating deployment %s status to %s", self._deploy_id, status
)
for txn in self._config.txn():
state = txn.get_processing_block_state(self._pb_id)
deployments = state.get("deployments")
deployments[self._deploy_id] = status
state["deployments"] = deployments
txn.update_processing_block_state(self._pb_id, state)
[docs]
def get_id(self) -> str:
"""
Get the deployment ID.
:return: deployment ID
:rtype: str
"""
return self._deploy_id
[docs]
def remove(self, deploy_id: str) -> None:
"""
Remove the execution engine.
:param deploy_id: deployment ID
:type deploy_id: str
"""
for txn in self._config.txn():
deploy = txn.get_deployment(deploy_id)
txn.delete_deployment(deploy)
[docs]
def is_finished(self, txn: Transaction) -> bool:
"""
Check if the deployment is finished.
:param txn: configuration transaction
:type txn: ska_sdp_config.Transaction
:rtype: bool
"""
state = txn.get_processing_block_state(self._pb_id)
deployments = state.get("deployments")
if self._deploy_id in deployments:
if deployments[self._deploy_id] == "FINISHED":
deployment_lists = txn.list_deployments()
if self._deploy_id in deployment_lists:
self.remove(self._deploy_id)
return True
return False