Source code for ecom.message

from abc import ABC, abstractmethod
from typing import Optional, Any, List, Dict, Iterator
from dataclasses import dataclass
from collections import OrderedDict

from ecom.datatypes import TypeInfo, EnumType, ArrayType, DynamicSizeError


@dataclass(frozen=True)
class MessageDatapointType:
    """ A base for all message datapoint types. """
    name: str
    """ The name of this datapoint. """

    type: TypeInfo
    """ The type information of this datapoint. """

    description: Optional[str] = None
    """ The description of this datapoint. """


@dataclass(frozen=True)
class MessageType:
    """ A base for all message types. """
    id: EnumType
    """ An enum value representing this message type. """

    data: List[MessageDatapointType]
    """ A list of datapoints that will be transmitted with this message. """


@dataclass(frozen=True)
class TelecommandResponseType:
    """ A type of telecommand response. """
    name: str
    """ The name of the response value. """

    typeInfo: TypeInfo
    """ The type of the response value. """

    description: Optional[str] = None
    """ A description of the response value. """


@dataclass(frozen=True)
class DependantTelecommandResponseType(TelecommandResponseType, ABC):
    """ A telecommand response whose type depends on the value of a datapoint of the telecommand. """
    provider: 'TelecommandDatapointType' = None
    """ The datapoint that this response is dependent on. """

    @abstractmethod
    def configureWith(self, providerValue: Any) -> TelecommandResponseType:
        """
        Create an instance of the dependant response for the value of the provider.

        :param providerValue: The value of the provider that this response depends on.
        :return: The configured TelecommandResponseType instance.
        """
        raise NotImplemented()


@dataclass(frozen=True)
class TelemetryDatapointType(MessageDatapointType):
    """ A telemetry response data type. """


@dataclass(frozen=True)
class TelemetryType(MessageType):
    """ A type of telemetry message. """
    id: EnumType
    """ An enum value representing this telemetry type. See CommunicationDatabase.telemetryTypeEnum. """

    data: List[TelemetryDatapointType]
    """ A list of datapoints that will be transmitted with the telemetry. """


@dataclass(frozen=True)
class Message:
    """ A generic message. """
    type: EnumType
    """
    The type of the message.
    See CommunicationDatabase.telemetryTypeEnum and CommunicationDatabase.telecommandTypeEnum.
    """

    data: Dict[str, Any]
    """ The data of this message. """

    header: Dict[str, Any]
    """ The data from the header of this message. """


class Telemetry(Message):
    """ A telemetry message. This is a message that has been sent to the base. """


class Telecommand(Message):
    """ A telecommand message. This is a message that has been sent from the base. """


@dataclass(frozen=True)
class TelecommandDatapointType(MessageDatapointType):
    """ A datapoint of a telecommand. """
    default: Optional = None
    """ The default value for the datapoint. """


@dataclass(frozen=True)
class DependantTelecommandDatapointType(TelecommandDatapointType, ABC):
    """ A datapoint whose type depends on the value of another datapoint. """
    provider: TelecommandDatapointType = None
    """ The argument that this datapoint is dependent on. """

    @abstractmethod
    def configureWith(self, providerValue: Any) -> TelecommandDatapointType:
        """
        Create an instance of this dependant telecommand datapoint for the value of the provider.

        :param providerValue: The value of the provider that this datapoint depends on.
        """
        raise NotImplemented()


@dataclass(frozen=True)
class TelecommandType(MessageType):
    """ A telecommand message. """
    id: EnumType
    """ An enum value representing this telecommand type. See CommunicationDatabase.telecommandTypeEnum. """

    data: List[TelecommandDatapointType]
    """ The datapoints of the telecommand. """

    response: Optional[TelecommandResponseType]
    """ The type information of the return value of the telecommand. """

    description: Optional[str]
    """ A description of the telecommand. """

    isDebug: bool
    """ Whether the telecommand is a debugging command. """


def iterateRequiredDatapoints(telecommand: TelecommandType) -> Iterator[TelecommandDatapointType]:
    """
    Iterate over the datapoints of the given telecommand that are required to serialize the telecommand.
    Parameters not included in the resulting iterator can be deduced from other parameters.

    :param telecommand: The telecommand type whose required datapoints should be iterated.
    :return: An iterator over the required datapoints of the telecommand type.
    """
    parameters = OrderedDict()
    for parameter in telecommand.data:
        parameters[parameter.name] = parameter
        if not isinstance(parameter, DependantTelecommandDatapointType) \
                and issubclass(parameter.type.type, ArrayType):
            try:
                len(parameter.type.type)
            except DynamicSizeError as error:
                parameters.pop(error.sizeMember, None)
    return (parameter for parameter in parameters.values())