Escape Sequences: The Key to Data Transparency

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

"""
PPP Escape Sequence Encoder
 
This module implements the transmitter-side escape processing
for PPP asynchronous framing, as specified in RFC 1662.
"""
 
from typing import List, Callable
 
class PPPEscapeEncoder:
    """
    Encodes frame data by inserting escape sequences for reserved bytes.
    
    This encoder handles:
    - Flag byte (0x7E) escaping (mandatory)
    - Escape byte (0x7D) escaping (mandatory)  
    - Control character escaping (ACCM-controlled)
    - Optional high-byte escaping (0x80-0xFF range, if configured)
    """
    
    FLAG_BYTE = 0x7E
    ESCAPE_BYTE = 0x7D
    XOR_MASK = 0x20
    
    def __init__(self, accm: int = 0xFFFFFFFF):
        """
        Initialize the encoder with an Async Control Character Map.
        
        Args:
            accm: 32-bit bitmap controlling which control characters
                  (0x00-0x1F) require escaping. Default is all escaped.
        """
        self.accm = accm
        self._build_escape_table()
    
    def _build_escape_table(self):
        """
        Pre-build a lookup table for O(1) escape decisions.
        
        Using a lookup table is faster than computing escape decisions
        for each byte, especially at high data rates.
        """
        self._must_escape = [False] * 256
        
        # Flag and escape bytes are always escaped
        self._must_escape[self.FLAG_BYTE] = True
        self._must_escape[self.ESCAPE_BYTE] = True
        
        # Control characters based on ACCM
        for i in range(32):
            if self.accm & (1 << i):
                self._must_escape[i] = True
        
        # 0x7F (DEL) is often escaped as well
        # (Some implementations include this in the escape set)
        # self._must_escape[0x7F] = True
    
    def update_accm(self, accm: int):
        """
        Update the ACCM after LCP negotiation.
        
        This would be called when the link configuration changes,
        such as after successful LCP negotiation reduces the escape set.
        """
        self.accm = accm
        self._build_escape_table()
    
    def encode(self, data: bytes) -> bytes:
        """
        Encode frame data with escape sequences.
        
        Args:
            data: Raw frame content (header + payload + FCS, no flags)
        
        Returns:
            Escaped data suitable for transmission between flags
        
        Note:
            This does NOT add surrounding flag bytes.
            Caller must add 0x7E before and after the returned data.
        """
        result = []
        
        for byte in data:
            if self._must_escape[byte]:
                # Insert escape sequence
                result.append(self.ESCAPE_BYTE)
                result.append(byte ^ self.XOR_MASK)
            else:
                # Pass through unchanged
                result.append(byte)
        
        return bytes(result)
    
    def encode_streaming(self, data: bytes, 
                         output_callback: Callable[[int], None]):
        """
        Stream-oriented encoding for memory-constrained environments.
        
        Instead of building a complete output buffer, this method
        calls the output callback for each byte as it's generated.
        
        Args:
            data: Raw frame content
            output_callback: Function called with each output byte
        """
        for byte in data:
            if self._must_escape[byte]:
                output_callback(self.ESCAPE_BYTE)
                output_callback(byte ^ self.XOR_MASK)
            else:
                output_callback(byte)
    
    def calculate_encoded_length(self, data: bytes) -> int:
        """
        Calculate the length of encoded data without actually encoding.
        
        Useful for buffer allocation when output size must be known
        before encoding begins.
        """
        extra = sum(1 for byte in data if self._must_escape[byte])
        return len(data) + extra
 
 
def demonstrate_escape_encoding():
    """
    Demonstrate escape sequence encoding with examples.
    """
    encoder = PPPEscapeEncoder(accm=0xFFFFFFFF)  # Full ACCM
    
    print("PPP Escape Sequence Encoding Demonstration")
    print("=" * 50)
    
    # Example 1: Data with flag byte
    data1 = bytes([0x01, 0x7E, 0x02])  # Contains flag
    encoded1 = encoder.encode(data1)
    print(f"
Example 1: Data containing flag byte")
    print(f"  Original: {data1.hex(' ')}")
    print(f"  Encoded:  {encoded1.hex(' ')}")
    print(f"  Expansion: {len(data1)} → {len(encoded1)} bytes")
    
    # Example 2: Data with escape byte
    data2 = bytes([0x01, 0x7D, 0x02])  # Contains escape
    encoded2 = encoder.encode(data2)
    print(f"
Example 2: Data containing escape byte")
    print(f"  Original: {data2.hex(' ')}")
    print(f"  Encoded:  {encoded2.hex(' ')}")
    
    # Example 3: Data with escape followed by what would decode as flag
    data3 = bytes([0x7D, 0x5E])  # Literal 0x7D 0x5E
    encoded3 = encoder.encode(data3)
    print(f"
Example 3: Data that looks like escape sequence")
    print(f"  Original: {data3.hex(' ')}")
    print(f"  Encoded:  {encoded3.hex(' ')}")
    print(f"  (0x7D becomes 7D 5D, then 0x5E passes through)")
    
    # Example 4: Control characters
    data4 = bytes([0x11, 0x13, 0x41])  # XON, XOFF, 'A'
    encoded4 = encoder.encode(data4)
    print(f"
Example 4: Control characters (XON, XOFF, 'A')")
    print(f"  Original: {data4.hex(' ')}")
    print(f"  Encoded:  {encoded4.hex(' ')}")
    
    # Example 5: Worst-case expansion
    data5 = bytes([0x7E] * 10)  # All flags
    encoded5 = encoder.encode(data5)
    print(f"
Example 5: Worst-case (all flag bytes)")
    print(f"  Original: {data5.hex(' ')}")
    print(f"  Encoded:  {encoded5.hex(' ')}")
    print(f"  Expansion: {len(data5)} → {len(encoded5)} bytes (2x)")
    
    # ACCM comparison
    print(f"
{'='*50}")
    print("ACCM Impact Analysis")
    
    # Data with control characters
    test_data = bytes(range(256))  # All possible bytes
    
    full_accm_encoder = PPPEscapeEncoder(accm=0xFFFFFFFF)
    min_accm_encoder = PPPEscapeEncoder(accm=0x00000000)
    
    full_encoded = full_accm_encoder.encode(test_data)
    min_encoded = min_accm_encoder.encode(test_data)
    
    print(f"
All 256 byte values:")
    print(f"  Full ACCM: {len(test_data)} → {len(full_encoded)} bytes")
    print(f"  Min ACCM:  {len(test_data)} → {len(min_encoded)} bytes")
    print(f"  Overhead: Full={len(full_encoded)-len(test_data)}, Min={len(min_encoded)-len(test_data)}")
 
 
if __name__ == "__main__":
    demonstrate_escape_encoding()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184

"""
PPP Escape Sequence Decoder
 
This module implements the receiver-side escape processing
for PPP asynchronous framing.
"""
 
from enum import Enum, auto
from typing import Optional, Tuple
 
class DecoderState(Enum):
    NORMAL = auto()    # Processing regular bytes
    ESCAPED = auto()   # Previous byte was 0x7D
 
class DecodeError(Exception):
    """Raised when escape sequence decoding fails."""
    pass
 
class PPPEscapeDecoder:
    """
    Decodes escape sequences to recover original frame data.
    
    This decoder handles the inverse of PPP escape encoding:
    - [0x7D 0x5E] → 0x7E
    - [0x7D 0x5D] → 0x7D
    - [0x7D 0xXX] → 0xXX XOR 0x20 (for other escaped bytes)
    """
    
    ESCAPE_BYTE = 0x7D
    XOR_MASK = 0x20
    
    def __init__(self):
        self.state = DecoderState.NORMAL
    
    def reset(self):
        """Reset decoder state for a new frame."""
        self.state = DecoderState.NORMAL
    
    def decode(self, data: bytes) -> bytes:
        """
        Decode escape sequences in received frame data.
        
        Args:
            data: Escaped frame data (between flags, flags removed)
        
        Returns:
            Original unescaped data
        
        Raises:
            DecodeError: If frame ends with incomplete escape sequence
        """
        self.reset()
        result = []
        
        for byte in data:
            decoded = self._process_byte(byte)
            if decoded is not None:
                result.append(decoded)
        
        # Check for incomplete escape at end of frame
        if self.state == DecoderState.ESCAPED:
            raise DecodeError("Frame ends with incomplete escape sequence")
        
        return bytes(result)
    
    def _process_byte(self, byte: int) -> Optional[int]:
        """
        Process a single byte through the decoder state machine.
        
        Returns the decoded byte if one is ready, None if more input needed.
        """
        if self.state == DecoderState.NORMAL:
            if byte == self.ESCAPE_BYTE:
                # Start escape sequence
                self.state = DecoderState.ESCAPED
                return None
            else:
                # Regular byte - pass through
                return byte
        
        else:  # ESCAPED state
            # Decode the escaped byte
            self.state = DecoderState.NORMAL
            return byte ^ self.XOR_MASK
    
    def decode_streaming(self, byte: int) -> Tuple[bool, Optional[int]]:
        """
        Stream-oriented decoding for interrupt-driven receivers.
        
        This method processes one byte at a time, maintaining state
        between calls. Suitable for byte-at-a-time UART reception.
        
        Args:
            byte: Single received byte
        
        Returns:
            (output_ready, decoded_byte) tuple
            output_ready: True if decoded_byte is valid
            decoded_byte: The decoded byte value, or None
        """
        decoded = self._process_byte(byte)
        return (decoded is not None, decoded)
 
 
def demonstrate_escape_decoding():
    """
    Demonstrate escape sequence decoding with examples.
    """
    decoder = PPPEscapeDecoder()
    
    print("PPP Escape Sequence Decoding Demonstration")
    print("=" * 50)
    
    # Example 1: Basic escape sequence
    encoded1 = bytes([0x01, 0x7D, 0x5E, 0x02])  # Contains escaped flag
    decoded1 = decoder.decode(encoded1)
    print(f"
Example 1: Escaped flag byte")
    print(f"  Received: {encoded1.hex(' ')}")
    print(f"  Decoded:  {decoded1.hex(' ')}")
    print(f"  (7D 5E became 7E)")
    
    # Example 2: Escaped escape byte
    encoded2 = bytes([0x01, 0x7D, 0x5D, 0x02])
    decoded2 = decoder.decode(encoded2)
    print(f"
Example 2: Escaped escape byte")
    print(f"  Received: {encoded2.hex(' ')}")
    print(f"  Decoded:  {decoded2.hex(' ')}")
    print(f"  (7D 5D became 7D)")
    
    # Example 3: Multiple escape sequences
    encoded3 = bytes([0x7D, 0x5E, 0x7D, 0x5D, 0x7D, 0x31])
    decoded3 = decoder.decode(encoded3)
    print(f"
Example 3: Multiple escapes")
    print(f"  Received: {encoded3.hex(' ')}")
    print(f"  Decoded:  {decoded3.hex(' ')}")
    print(f"  (3 escape sequences → 3 bytes)")
    
    # Example 4: Verify round-trip encoding
    from escape_encoder import PPPEscapeEncoder  # Import encoder
    
    original = bytes([0x7E, 0x7D, 0x00, 0x11, 0x13, 0x41])
    encoder = PPPEscapeEncoder()
    encoded = encoder.encode(original)
    decoded = decoder.decode(encoded)
    
    print(f"
Example 4: Round-trip verification")
    print(f"  Original: {original.hex(' ')}")
    print(f"  Encoded:  {encoded.hex(' ')}")
    print(f"  Decoded:  {decoded.hex(' ')}")
    print(f"  Match: {original == decoded}")
    
    # Example 5: Stream decoding
    print(f"
Example 5: Stream (byte-by-byte) decoding")
    stream_decoder = PPPEscapeDecoder()
    stream_data = bytes([0x41, 0x7D, 0x5E, 0x42])
    print(f"  Input stream: {stream_data.hex(' ')}")
    print(f"  Byte-by-byte processing:")
    
    for byte in stream_data:
        ready, decoded = stream_decoder.decode_streaming(byte)
        hex_byte = f"0x{byte:02X}"
        if ready:
            print(f"    Received {hex_byte} → Output 0x{decoded:02X}")
        else:
            print(f"    Received {hex_byte} → (buffered, awaiting next byte)")
    
    # Example 6: Error case
    print(f"
Example 6: Error case (incomplete escape)")
    try:
        bad_data = bytes([0x41, 0x7D])  # Ends with escape byte
        decoder.decode(bad_data)
    except DecodeError as e:
        print(f"  Data: {bad_data.hex(' ')}")
        print(f"  Error: {e}")
 
 
if __name__ == "__main__":
    demonstrate_escape_decoding()