base.py

from .exceptions import PacketLostError, StalePacketError
from .packets import (
    S7Header,
    S7Packet,
    SetupCommunicationParameter,
    SetupCommunicationRequest,
)


class BaseS7Comm:
    MAX_VARS = 20  # Max vars that can be transferred with MultiRead/MultiWrite

    def __init__(self, pdu_length: int = 480) -> None:
        self._pdu_reference = -1
        self.pdu_length = pdu_length

    @property
    def pdu_reference(self) -> int:
        self._pdu_reference = (self._pdu_reference + 1) & 0xFFFF
        return self._pdu_reference

    @staticmethod
    def _create_communication_request(
        max_amq_caller_ack: int = 0x0001,
        max_amq_callee_ack: int = 0x0001,
        pdu_length: int = 0x01E0,
    ) -> SetupCommunicationRequest:
        parameter = SetupCommunicationParameter(
            max_amq_caller_ack=max_amq_caller_ack,
            max_amq_callee_ack=max_amq_callee_ack,
            pdu_length=pdu_length,
        )
        return SetupCommunicationRequest(parameter=parameter)

    @staticmethod
    def _create_packet(request: S7Packet, pdu_reference: int) -> bytes:
        parameter = request.serialize_parameter()
        data = request.serialize_data()
        header = S7Header(
            pdu_reference=pdu_reference,
            message_type=request.MESSAGE_TYPE,
            parameter_length=len(parameter),
            data_length=len(data),
        )
        request.header = header
        return bytes(header.serialize() + parameter + data)

    def _validate_pdu_reference(self, response: S7Packet) -> None:
        """Raises if PDU reference is invalid."""
        assert response.header is not None
        if response.header.pdu_reference > self._pdu_reference:
            raise PacketLostError(f"Expected {self._pdu_reference}, got {response.header.pdu_reference}")
        elif response.header.pdu_reference < self._pdu_reference:
            raise StalePacketError(f"Stale packet: {response.header.pdu_reference}")