#
# This file is part of the SKA Tango Base project
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE.txt for more info.
#
from __future__ import annotations
import typing
import warnings
from .. import type_hints
[docs]
class NoValueType:
"""Type of the :obj:`NoValue` singleton."""
NoValue = NoValueType()
"""Used as the default initial value for a stored :obj:`Signal`.
Required to distinguish from ``None``, which is a valid value to emit.
"""
T = typing.TypeVar("T")
[docs]
class Signal(typing.Generic[T]):
"""
Descriptor to provide access to a signal on a shared bus.
Can only be used on classes which model
:py:class:`~ska_tango_base.type_hints.SharingObserverProtocol`.
Whenever the ``Signal`` is set, a signal is emitted with the value provided.
By default, the ``Signal`` descriptor is not gettable, however, if
``stored`` is set then the last value emitted will be returned on ``__get__``.
**DEPRECATED**: The ``Signal`` can be assigned a special ``NoValue`` to clear the
cached value. The ``NoValue`` is emitted on the bus. Instead use the ``del`` keyword
on a ``Signal`` to clear the cache without emitting anything. Assigning ``NoValue``
to a ``Signal`` will raise an exception in the future. ``NoValue`` will only be
passed as the default ``initial_value``.
The relative name of the signal is taken from the name of the ``Signal`` on
the owner class, and can be overridden by providing ``name`` kwarg to the
initialiser.
The relative name is prefixed by the
:py:attr:`~ska_tango_base.type_hints.SharingObserverProtocol.observer_prefix`
of the owner object before being emitted on the bus.
:param name: Name of the signal, if ``None`` this objects name is used.
:param stored: If ``True``, store the latest emitted value on the parent object.
:param initial_value: If not ``NoValue``, emit this value when connected to a
new bus.
"""
[docs]
def __init__(
self,
/,
name: str | None = None,
stored: bool = False,
initial_value: typing.Any = NoValue,
) -> None:
"""Initialise the object."""
self._relative_name = name
self._stored = stored
self._initial_value = initial_value
def __set_name__(self, owner: typing.Any, name: str) -> None:
"""Set the name of the descriptor."""
if self._relative_name is None:
self._relative_name = name
@property
def relative_name(self) -> str:
"""Name of signal, relative to parent object."""
return typing.cast(str, self._relative_name)
def _absolute_name_for(self, obj: type_hints.SharingObserverProtocol) -> str:
"""Return the absolute name of the signal when relative to obj."""
return f"{obj.observer_prefix}{self.relative_name}"
@typing.overload
def __get__(self, obj: None, objtype: typing.Any = None) -> Signal[T]: ...
@typing.overload
def __get__(
self, obj: type_hints.SharingObserverProtocol, objtype: typing.Any = None
) -> T: ...
def __get__(
self,
obj: type_hints.SharingObserverProtocol | None,
objtype: typing.Any = None,
) -> T | Signal[T]:
"""
Return the last emitted value or self.
:raises AttributeError: If the signal has not been created with ``stored=True``.
:raises AttributeError: If the signal has never been emitted.
"""
if obj is None:
return self
name = self._absolute_name_for(obj)
if not self._stored:
raise AttributeError(
f"'{obj.__class__.__name__}' object does not store emitted values for "
f"signal '{name}' on its shared bus.",
)
try:
value = obj.shared_bus.get_last_emitted_value(name)
except KeyError as e:
raise AttributeError(
f"'{obj.__class__.__name__}' object has no last emitted value stored "
f"for signal '{name}' on its shared bus.",
) from e
return typing.cast(T, value)
def __set__(
self: Signal[T],
obj: type_hints.SharingObserverProtocol,
value: T | None | NoValueType,
) -> None:
"""Emit the signal."""
name = self._absolute_name_for(obj)
if value is NoValue:
warnings.warn(
"Assigning 'NoValue' to a 'Signal' is deprecated since ska-tango-base "
"1.5.0. It will raise an exception in the next major release. To "
"delete the stored value, use the 'del' keyword on the signal instead. "
"To change a linked attribute's quality to 'ATTR_INVALID', assign "
"'None' instead.",
DeprecationWarning,
)
try:
obj.shared_bus.delete_last_emitted_value(name)
except KeyError:
pass
obj.shared_bus.emit(name, value, store=self._stored and value is not NoValue)
def __delete__(self: Signal[T], obj: type_hints.SharingObserverProtocol) -> None:
"""
Delete the last emitted value stored for the signal.
:raises AttributeError: If the signal has not been created with ``stored=True``.
:raises AttributeError: If the signal has never been emitted or has already been
deleted.
"""
name = self._absolute_name_for(obj)
if not self._stored:
raise AttributeError(
f"'{obj.__class__.__name__}' object does not store emitted values for "
f"signal '{name}' on its shared bus.",
)
try:
obj.shared_bus.delete_last_emitted_value(name)
except KeyError as e:
raise AttributeError(
f"'{obj.__class__.__name__}' object has no last emitted value stored "
f"for signal '{name}' on its shared bus.",
) from e