Module csw.CommandResponse

Classes

class Accepted (runId: str)
Expand source code 
@dataclass
class Accepted(CommandResponse):
    """Represents a final response stating acceptance of a command received"""

    @staticmethod
    def _fromDict(obj: dict):
        return Accepted(
            runId=obj['runId']
        )

Represents a final response stating acceptance of a command received

Ancestors

Inherited members

class AssemblyBusyIssue (reason: str)
Expand source code 
class AssemblyBusyIssue(CommandIssue):
    """Returned when an Assembly receives a command and one is already executing"""

Returned when an Assembly receives a command and one is already executing

Ancestors

Inherited members

class Cancelled (runId: str)
Expand source code 
@dataclass
class Cancelled(CommandResponse):
    """Represents a negative response that describes the cancellation of command"""

    @staticmethod
    def _fromDict(obj: dict):
        return Cancelled(
            runId=obj['runId']
        )

Represents a negative response that describes the cancellation of command

Ancestors

Inherited members

class CommandError (submitResponse: Error | Invalid | Locked | Started | Completed | Cancelled)
Expand source code 
class CommandError(Exception):
    def __init__(self, submitResponse: SubmitResponse):
        super().__init__(submitResponse.message)

Common base class for all non-exit exceptions.

Ancestors

  • builtins.Exception
  • builtins.BaseException
class CommandIssue (reason: str)
Expand source code 
@dataclass
class CommandIssue:
    """Describes a command issue with appropriate reason for validation failure"""
    reason: str

    @staticmethod
    def _fromDict(obj: dict):
        return globals()[obj['_type']](obj['reason'])

Describes a command issue with appropriate reason for validation failure

Subclasses

Instance variables

var reason : str

The type of the None singleton.

class CommandResponse (runId: str)
Expand source code 
@dataclass
class CommandResponse:
    """
    Type of response to a command (submit, oneway or validate).
    Note that oneway and validate responses are limited to Accepted, Invalid or Locked.
    """
    runId: str

    def _asDict(self):
        """
        Returns: a dictionary corresponding to this object
        """
        return {
            "_type": self.__class__.__name__,
            'runId': self.runId,
        }

    @staticmethod
    def _fromDict(obj: dict):
        match obj["_type"]:
            case "Cancelled":
                return Cancelled._fromDict(obj)
            case "Error":
                return Error._fromDict(obj)
            case "Locked":
                return Locked._fromDict(obj)
            case "Started":
                return Started._fromDict(obj)
            case "Completed":
                return Completed._fromDict(obj)
            case "Accepted":
                return Accepted._fromDict(obj)
            case "Invalid":
                return Invalid._fromDict(obj)

    def isPositive(self) -> bool:
        return isinstance(self, Completed) or isinstance(self, Accepted)

    def isCompleted(self) -> bool:
        return isinstance(self, Completed)

    def isIntermediate(self) -> bool:
        return isinstance(self, Started)

    def isStarted(self) -> bool:
        return isinstance(self, Started)

    def isFinal(self) -> bool:
        return not isinstance(self, Started)

    def isNegative(self) -> bool:
        return not (self.isPositive() or self.isIntermediate())

    def isFailed(self) -> bool:
        return self.isNegative()

    def onFailedTerminate(self) -> Self:
        if self.isFailed():
            # noinspection PyTypeChecker
            raise CommandError(self)
        return self

    def onStarted(self, func: Callable[[Self], None]):
        if isinstance(self, Started):
            func(self)

    def onCompleted(self, func: Callable[[Self], None]):
        if isinstance(self, Completed):
            func(self)

    def onFailed(self, func: Callable[[Self], None]):
        if self.isFailed():
            func(self)

Type of response to a command (submit, oneway or validate). Note that oneway and validate responses are limited to Accepted, Invalid or Locked.

Subclasses

Instance variables

var runId : str

The type of the None singleton.

Methods

def isCompleted(self) ‑> bool
Expand source code 
def isCompleted(self) -> bool:
    return isinstance(self, Completed)
def isFailed(self) ‑> bool
Expand source code 
def isFailed(self) -> bool:
    return self.isNegative()
def isFinal(self) ‑> bool
Expand source code 
def isFinal(self) -> bool:
    return not isinstance(self, Started)
def isIntermediate(self) ‑> bool
Expand source code 
def isIntermediate(self) -> bool:
    return isinstance(self, Started)
def isNegative(self) ‑> bool
Expand source code 
def isNegative(self) -> bool:
    return not (self.isPositive() or self.isIntermediate())
def isPositive(self) ‑> bool
Expand source code 
def isPositive(self) -> bool:
    return isinstance(self, Completed) or isinstance(self, Accepted)
def isStarted(self) ‑> bool
Expand source code 
def isStarted(self) -> bool:
    return isinstance(self, Started)
def onCompleted(self, func: Callable[[Self], None])
Expand source code 
def onCompleted(self, func: Callable[[Self], None]):
    if isinstance(self, Completed):
        func(self)
def onFailed(self, func: Callable[[Self], None])
Expand source code 
def onFailed(self, func: Callable[[Self], None]):
    if self.isFailed():
        func(self)
def onFailedTerminate(self) ‑> Self
Expand source code 
def onFailedTerminate(self) -> Self:
    if self.isFailed():
        # noinspection PyTypeChecker
        raise CommandError(self)
    return self
def onStarted(self, func: Callable[[Self], None])
Expand source code 
def onStarted(self, func: Callable[[Self], None]):
    if isinstance(self, Started):
        func(self)
class Completed (runId: str,
result: Result = <factory>)
Expand source code 
@dataclass
class Completed(CommandResponse):
    """Represents a positive response stating completion of command"""
    result: Result = field(default_factory=lambda: Result([]))

    # noinspection PyProtectedMember
    def _asDict(self):
        """
        Returns: dict
            a dictionary corresponding to this object
        """
        return {
            "_type": self.__class__.__name__,
            'runId': self.runId,
            'result': self.result._asDict()
        }

    @staticmethod
    def _fromDict(obj: dict):
        return Completed(
            runId=obj['runId'],
            result=Result._fromDict(obj['result']),
        )

Represents a positive response stating completion of command

Ancestors

Instance variables

var result : Result

The type of the None singleton.

Inherited members

class Error (runId: str, message: str)
Expand source code 
@dataclass
class Error(CommandResponse):
    """Represents a negative response that describes an error in executing the command"""
    message: str

    def _asDict(self):
        """
        Returns: dict
            a dictionary corresponding to this object
        """
        return {
            "_type": self.__class__.__name__,
            'runId': self.runId,
            'message': self.message
        }

    @staticmethod
    def _fromDict(obj: dict):
        return Error(
            runId=obj['runId'],
            message=obj['message']
        )

Represents a negative response that describes an error in executing the command

Ancestors

Instance variables

var message : str

The type of the None singleton.

Inherited members

class HCDBusyIssue (reason: str)
Expand source code 
class HCDBusyIssue(CommandIssue):
    """Returned when the HCD is busy and can't process a command"""

Returned when the HCD is busy and can't process a command

Ancestors

Inherited members

class IdNotAvailableIssue (reason: str)
Expand source code 
class IdNotAvailableIssue(CommandIssue):
    """Returned when a CommandResponse associated with runId is not available"""

Returned when a CommandResponse associated with runId is not available

Ancestors

Inherited members

class Invalid (runId: str,
issue: CommandIssue)
Expand source code 
@dataclass
class Invalid(CommandResponse):
    issue: CommandIssue

    def _asDict(self):
        """
        Returns: dict
            a dictionary for this object
        """
        return {
            "_type": self.__class__.__name__,
            'runId': self.runId,
            'issue': {
                "_type": self.issue.__class__.__name__,
                "reason": self.issue.reason
            }
        }

    @staticmethod
    def _fromDict(obj: dict):
        return Invalid(
            runId=obj['runId'],
            issue=CommandIssue._fromDict(obj['issue']),
        )

Invalid(runId: str, issue: csw.CommandResponse.CommandIssue)

Ancestors

Instance variables

var issue : CommandIssue

The type of the None singleton.

Inherited members

class Locked (runId: str)
Expand source code 
@dataclass
class Locked(CommandResponse):
    """Represents a negative response stating that a component is Locked and command was not validated or executed"""

    def _asDict(self):
        """
        Returns: dict
            a dictionary corresponding to this object
        """
        return {
            "_type": self.__class__.__name__,
            'runId': self.runId,
        }

    @staticmethod
    def _fromDict(obj: dict):
        return Locked(
            runId=obj['runId'],
        )

Represents a negative response stating that a component is Locked and command was not validated or executed

Ancestors

Inherited members

class MissingKeyIssue (reason: str)
Expand source code 
class MissingKeyIssue(CommandIssue):
    """Returned when a command is missing a required key/parameter"""

Returned when a command is missing a required key/parameter

Ancestors

Inherited members

class OtherIssue (reason: str)
Expand source code 
class OtherIssue(CommandIssue):
    """A required Sequencer is not available"""

A required Sequencer is not available

Ancestors

Inherited members

class ParameterValueOutOfRangeIssue (reason: str)
Expand source code 
class ParameterValueOutOfRangeIssue(CommandIssue):
    """Parameter of a command is out of range"""

Parameter of a command is out of range

Ancestors

Inherited members

class RequiredAssemblyUnavailableIssue (reason: str)
Expand source code 
class RequiredAssemblyUnavailableIssue(CommandIssue):
    """A required Assembly is not available"""

A required Assembly is not available

Ancestors

Inherited members

class RequiredHCDUnavailableIssue (reason: str)
Expand source code 
class RequiredHCDUnavailableIssue(CommandIssue):
    """A required HCD is not available"""

A required HCD is not available

Ancestors

Inherited members

class RequiredSequencerUnavailableIssue (reason: str)
Expand source code 
class RequiredSequencerUnavailableIssue(CommandIssue):
    """Returned when some other issue occurred apart from those already defined"""

Returned when some other issue occurred apart from those already defined

Ancestors

Inherited members

class RequiredServiceUnavailableIssue (reason: str)
Expand source code 
class RequiredServiceUnavailableIssue(CommandIssue):
    """A required service is not available"""

A required service is not available

Ancestors

Inherited members

class Result (paramSet: List[Parameter] = <factory>)
Expand source code 
@dataclass
class Result:
    """A result containing parameters for command response"""
    paramSet: List[Parameter] = field(default_factory=list)

    # noinspection PyProtectedMember
    def _asDict(self):
        """
        Returns: dict
            a dictionary corresponding to this object
        """
        return {
            'paramSet': list(map(lambda p: p._asDict(), self.paramSet))
        }

    # noinspection PyProtectedMember
    @staticmethod
    def _fromDict(obj):
        """
        Returns a Result for the given dict.
        """
        paramSet = list(map(lambda p: Parameter._fromDict(p), obj['paramSet']))
        return Result(paramSet)

A result containing parameters for command response

Instance variables

var paramSet : List[Parameter]

The type of the None singleton.

class Started (runId: str)
Expand source code 
@dataclass
class Started(CommandResponse):
    """Represents an intermediate response stating a long-running command has been started"""

    def _asDict(self):
        """
        Returns: dict
            a dictionary corresponding to this object
        """
        return {
            "_type": self.__class__.__name__,
            'runId': self.runId,
        }

    @staticmethod
    def _fromDict(obj: dict):
        return Started(
            runId=obj['runId'],
        )

Represents an intermediate response stating a long-running command has been started

Ancestors

Inherited members

class UnresolvedLocationsIssue (reason: str)
Expand source code 
class UnresolvedLocationsIssue(CommandIssue):
    """Returned when some required location is not available"""

Returned when some required location is not available

Ancestors

Inherited members

class UnsupportedCommandInStateIssue (reason: str)
Expand source code 
class UnsupportedCommandInStateIssue(CommandIssue):
    """A command is unsupported in the current state"""

A command is unsupported in the current state

Ancestors

Inherited members

class UnsupportedCommandIssue (reason: str)
Expand source code 
class UnsupportedCommandIssue(CommandIssue):
    """A command is unsupported by component"""

A command is unsupported by component

Ancestors

Inherited members

class WrongCommandTypeIssue (reason: str)
Expand source code 
class WrongCommandTypeIssue(CommandIssue):
    """Returned when the given command type is not expected"""

Returned when the given command type is not expected

Ancestors

Inherited members

class WrongInternalStateIssue (reason: str)
Expand source code 
class WrongInternalStateIssue(CommandIssue):
    """The component is in the wrong internal state to handle a command"""

The component is in the wrong internal state to handle a command

Ancestors

Inherited members

class WrongNumberOfParametersIssue (reason: str)
Expand source code 
class WrongNumberOfParametersIssue(CommandIssue):
    """Returned when a command does not have the correct number of parameters"""

Returned when a command does not have the correct number of parameters

Ancestors

Inherited members

class WrongParameterTypeIssue (reason: str)
Expand source code 
class WrongParameterTypeIssue(CommandIssue):
    """Returned when the parameter for a key is not the correct type (i.e. int vs double, etc.)"""

Returned when the parameter for a key is not the correct type (i.e. int vs double, etc.)

Ancestors

Inherited members

class WrongPrefixIssue (reason: str)
Expand source code 
class WrongPrefixIssue(CommandIssue):
    """Returned when an Assembly receives a configuration with a prefix that it doesn't support"""

Returned when an Assembly receives a configuration with a prefix that it doesn't support

Ancestors

Inherited members

class WrongUnitsIssue (reason: str)
Expand source code 
class WrongUnitsIssue(CommandIssue):
    """Returned when a parameter value does not have the correct units"""

Returned when a parameter value does not have the correct units

Ancestors

Inherited members