DJ-Goanna-Coding/oppo-node

1"""2Signal noise filter: packet-inspection for an incoming IoT data stream.3 4The filter performs cheap, deterministic validation and drops malformed5packets before they reach downstream consumers. It does not attempt any6semantic interpretation of packet payloads.7 8A "packet" is any mapping that contains at least the keys declared in9``REQUIRED_FIELDS``. Additional fields are preserved unchanged.10"""11 12from __future__ import annotations13 14import logging15from typing import Any, Iterable, Iterator, Mapping16 17from .constants import MAX_PACKET_BYTES18 19logger = logging.getLogger(__name__)20 21REQUIRED_FIELDS: tuple[str, ...] = ("device_id", "timestamp", "payload")22 23 24class PacketValidationError(ValueError):25    """Raised when a packet fails structural validation."""26 27 28class SignalNoiseFilter:29    """Drop malformed packets from an IoT stream.30 31    Parameters32    ----------33    max_packet_bytes:34        Maximum accepted size of the serialised payload in bytes. Packets35        whose ``payload`` exceeds this size are dropped.36    required_fields:37        Iterable of field names that must be present and non-empty in every38        packet. Defaults to :data:`REQUIRED_FIELDS`.39    """40 41    def __init__(42        self,43        max_packet_bytes: int = MAX_PACKET_BYTES,44        required_fields: Iterable[str] = REQUIRED_FIELDS,45    ) -> None:46        if max_packet_bytes <= 0:47            raise ValueError("max_packet_bytes must be positive")48        self.max_packet_bytes = max_packet_bytes49        self.required_fields = tuple(required_fields)50        self.dropped_count = 051        self.accepted_count = 052 53    def validate(self, packet: Any) -> None:54        """Validate a single packet. Raises :class:`PacketValidationError`.55 56        The packet must be a mapping with all required fields present and57        with a payload no larger than ``max_packet_bytes`` bytes when58        encoded as UTF-8.59        """60        if not isinstance(packet, Mapping):61            raise PacketValidationError(62                f"expected mapping, got {type(packet).__name__}"63            )64 65        for field in self.required_fields:66            if field not in packet:67                raise PacketValidationError(f"missing required field: {field!r}")68            value = packet[field]69            if value is None or (isinstance(value, str) and not value):70                raise PacketValidationError(f"field {field!r} is empty")71 72        payload = packet["payload"]73        try:74            encoded = (75                payload76                if isinstance(payload, (bytes, bytearray))77                else str(payload).encode("utf-8")78            )79        except (UnicodeEncodeError, TypeError) as exc:80            raise PacketValidationError(f"payload not encodable: {exc}") from exc81 82        if len(encoded) > self.max_packet_bytes:83            raise PacketValidationError(84                f"payload size {len(encoded)} exceeds max {self.max_packet_bytes}"85            )86 87    def is_valid(self, packet: Any) -> bool:88        """Return ``True`` if the packet passes validation, otherwise ``False``."""89        try:90            self.validate(packet)91        except PacketValidationError:92            return False93        return True94 95    def filter_stream(self, packets: Iterable[Any]) -> Iterator[Mapping[str, Any]]:96        """Yield only the valid packets from ``packets``.97 98        Invalid packets are logged at ``DEBUG`` level and counted in99        :attr:`dropped_count`.100        """101        for packet in packets:102            try:103                self.validate(packet)104            except PacketValidationError as exc:105                self.dropped_count += 1106                logger.debug("dropping malformed packet: %s", exc)107                continue108            self.accepted_count += 1109            yield packet110