Module csw.CommandServiceRequest

Classes

class CommandServiceRequest (controlCommand: ControlCommand)

Expand source code

@dataclass_json
@dataclass
class CommandServiceRequest:
    """
    Represents a command that requires a response (of type CommandResponse).

    Args:
        controlCommand (ControlCommand): The command to send
    """
    controlCommand: ControlCommand

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

Represents a command that requires a response (of type CommandResponse).

Args

controlCommand : ControlCommand

The command to send

Subclasses

• Oneway
• Submit
• Validate

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A

def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A

def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var controlCommand : ControlCommand

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class ExecuteDiagnosticMode (startTime: UTCTime,
hint: str)

Expand source code

@dataclass
class ExecuteDiagnosticMode:
    startTime: UTCTime
    hint: str

    @staticmethod
    def _fromDict(obj):
        """
        Returns an ExecuteDiagnosticMode for the given dict.
        """
        startTime = UTCTime.from_str(obj['startTime'])
        hint = obj['hint']
        return ExecuteDiagnosticMode(startTime, hint)

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

ExecuteDiagnosticMode(startTime: csw.TMTTime.UTCTime, hint: str)

Instance variables

var hint : str

The type of the None singleton.

var startTime : UTCTime

The type of the None singleton.

class ExecuteOperationsMode

Expand source code

class ExecuteOperationsMode:

    @staticmethod
    def _fromDict(_):
        return ExecuteOperationsMode()

    def _asDict(self):
        return {
            "_type": self.__class__.__name__,
        }

class GoOffline

Expand source code

class GoOffline:

    @staticmethod
    def _fromDict(_):
        return GoOffline()

    def _asDict(self):
        return {
            "_type": self.__class__.__name__,
        }

class GoOnline

Expand source code

class GoOnline:

    @staticmethod
    def _fromDict(_):
        return GoOnline()

    def _asDict(self):
        return {
            "_type": self.__class__.__name__,
        }

class Oneway (controlCommand: ControlCommand)

Expand source code

@dataclass_json
@dataclass
class Oneway(CommandServiceRequest):
    pass

Oneway(controlCommand: csw.ParameterSetType.ControlCommand)

Ancestors

• CommandServiceRequest

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A

def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A

def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members

• CommandServiceRequest:
  • controlCommand

class QueryFinal (runId: str, timeout: datetime.timedelta)

Expand source code

@dataclass
class QueryFinal:
    """
    A message sent to query the final result of a long-running command.
    The response should be a CommandResponse.

    Args:
        runId (str): The command's runId
        timeout (timedelta) amount of time to wait
    """
    runId: str
    timeout: timedelta

    @staticmethod
    def _fromDict(obj):
        """
        Returns a ControlCommand for the given dict.
        """
        try:
            runId = obj['runId']
            timeout = timedelta(seconds=obj['timeoutInSeconds'])
            return QueryFinal(runId, timeout)
        except:
            traceback.print_exc()

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

A message sent to query the final result of a long-running command. The response should be a CommandResponse.

Args

runId : str

The command's runId

timeout (timedelta) amount of time to wait

Instance variables

var runId : str

The type of the None singleton.

var timeout : datetime.timedelta

The type of the None singleton.

class StateName (name: str)

Expand source code

@dataclass_json
@dataclass
class StateName:
    name: str

StateName(name: str)

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A

def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A

def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var name : str

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class Submit (controlCommand: ControlCommand)

Expand source code

@dataclass_json
@dataclass
class Submit(CommandServiceRequest):
    pass

Submit(controlCommand: csw.ParameterSetType.ControlCommand)

Ancestors

• CommandServiceRequest

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A

def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A

def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members

• CommandServiceRequest:
  • controlCommand

class SubscribeCurrentState (stateNames: List[str])

Expand source code

@dataclass
class SubscribeCurrentState:
    """
    Message used to subscribe to the current state of a component.

    Args:
        stateNames (List[str]) list of current state names to subscribe to
    """
    stateNames: List[str]

    @staticmethod
    def _fromDict(obj):
        """
        Returns a SubscribeCurrentState for the given dict.
        """
        if "names" in obj.keys():
            stateNames = obj.get("names")
        else:
            stateNames = []
        return SubscribeCurrentState(stateNames)

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

Message used to subscribe to the current state of a component.

Args

stateNames (List[str]) list of current state names to subscribe to

Instance variables

var stateNames : List[str]

The type of the None singleton.

class Validate (controlCommand: ControlCommand)

Expand source code

@dataclass_json
@dataclass
class Validate(CommandServiceRequest):
    pass

Validate(controlCommand: csw.ParameterSetType.ControlCommand)

Ancestors

• CommandServiceRequest

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A

def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A

def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members

• CommandServiceRequest:
  • controlCommand