<commit_before>import sys
from dataclasses import dataclass
from textwrap import indent
from typing import Union, cast

from x690 import decode
from x690.types import Integer, OctetString, Sequence

if sys.stdout.isatty():
    # Add some colour for TTYs
    INDENT_STRING = " \u001b[38;5;22m│\u001b[0m "
else:
    INDENT_STRING = " │ "


@dataclass(frozen=True)
class V3Flags:
    """
    This class represents the SNMP message flags.

    At the time of this writing, this only contains the flags auth, priv and
    reportable. The remaining bytes are (as of this state) reserved by the SNMP
    standard.
    """

    auth: bool = False
    priv: bool = False
    reportable: bool = False

    @staticmethod
    def decode(blob: OctetString) -> "V3Flags":
        """
        Converts an OctetString instance into a more pythonic instance
        """
        flags = int.from_bytes(blob.pythonize(), "big")
        reportable = bool(flags & 0b100)
        priv = bool(flags & 0b010)
        auth = bool(flags & 0b001)
        return V3Flags(auth, priv, reportable)

    def __bytes__(self) -> bytes:
        value = 0
        value |= int(self.reportable) << 2
        value |= int(self.priv) << 1
        value |= int(self.auth)
        return bytes([value])


@dataclass(frozen=True)
class HeaderData:
    message_id: int
    message_max_size: int
    flags: V3Flags
    security_model: int

    def as_snmp_type(self) -> Sequence:
        return Sequence(
            [
                Integer(self.message_id),
                Integer(self.message_max_size),
                OctetString(bytes(self.flags)),
                Integer(self.security_model),
            ]
        )

    def __bytes__(self) -> bytes:
        return bytes(self.as_snmp_type())

    def pretty(self, depth: int = 0) -> str:
        lines = []
        lines.append("Global Data")
        lines.append(f"{INDENT_STRING}Message ID: {self.message_id}")
        lines.append(
            f"{INDENT_STRING}Message max size: {self.message_max_size}"
        )
        lines.append(f"{INDENT_STRING}Flags: {self.flags}")
        lines.append(f"{INDENT_STRING}Security Model: {self.security_model}")
        return indent("\n".join(lines), INDENT_STRING * depth)


@dataclass(frozen=True)
class ScopedPDU:
    context_engine_id: OctetString
    context_name: OctetString
    data: "PDU"

    def __bytes__(self) -> bytes:
        return bytes(self.as_snmp_type())

    @staticmethod
    def decode(data: bytes, slc: slice = slice(None)) -> "ScopedPDU":
        sequence, _ = decode(data, enforce_type=Sequence, strict=False)<commit_msg>Pass slice-start to PDU decoding<commit_after>import sys
from dataclasses import dataclass
from textwrap import indent
from typing import Union, cast

from x690 import decode
from x690.types import Integer, OctetString, Sequence

if sys.stdout.isatty():
    # Add some colour for TTYs
    INDENT_STRING = " \u001b[38;5;22m│\u001b[0m "
else:
    INDENT_STRING = " │ "


@dataclass(frozen=True)
class V3Flags:
    """
    This class represents the SNMP message flags.

    At the time of this writing, this only contains the flags auth, priv and
    reportable. The remaining bytes are (as of this state) reserved by the SNMP
    standard.
    """

    auth: bool = False
    priv: bool = False
    reportable: bool = False

    @staticmethod
    def decode(blob: OctetString) -> "V3Flags":
        """
        Converts an OctetString instance into a more pythonic instance
        """
        flags = int.from_bytes(blob.pythonize(), "big")
        reportable = bool(flags & 0b100)
        priv = bool(flags & 0b010)
        auth = bool(flags & 0b001)
        return V3Flags(auth, priv, reportable)

    def __bytes__(self) -> bytes:
        value = 0
        value |= int(self.reportable) << 2
        value |= int(self.priv) << 1
        value |= int(self.auth)
        return bytes([value])


@dataclass(frozen=True)
class HeaderData:
    message_id: int
    message_max_size: int
    flags: V3Flags
    security_model: int

    def as_snmp_type(self) -> Sequence:
        return Sequence(
            [
                Integer(self.message_id),
                Integer(self.message_max_size),
                OctetString(bytes(self.flags)),
                Integer(self.security_model),
            ]
        )

    def __bytes__(self) -> bytes:
        return bytes(self.as_snmp_type())

    def pretty(self, depth: int = 0) -> str:
        lines = []
        lines.append("Global Data")
        lines.append(f"{INDENT_STRING}Message ID: {self.message_id}")
        lines.append(
            f"{INDENT_STRING}Message max size: {self.message_max_size}"
        )
        lines.append(f"{INDENT_STRING}Flags: {self.flags}")
        lines.append(f"{INDENT_STRING}Security Model: {self.security_model}")
        return indent("\n".join(lines), INDENT_STRING * depth)


@dataclass(frozen=True)
class ScopedPDU:
    context_engine_id: OctetString
    context_name: OctetString
    data: "PDU"

    def __bytes__(self) -> bytes:
        return bytes(self.as_snmp_type())

    @staticmethod
    def decode(data: bytes, slc: slice = slice(None)) -> "ScopedPDU":
        sequence, _ = decode(
            data,
            start_index=slc.start or 0,
            enforce_type=Sequence,
            strict=False,
        )
{'fraction_non_alphanumeric': 0.10612321562734786, 'fraction_numerical': 0.014462809917355372, 'mean_word_length': 2.512532981530343, 'pattern_counts': {'":': 4, '<': 11, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2305978', 'n_tokens_mistral': 1732, 'n_tokens_neox': 1639, 'n_words': 419}