import os from typing import List, Callable, Optional from .models import Packet, Page, TeletextService, decode_hamming_8_4 def load_t42(file_path: str, progress_callback: Optional[Callable[[int, int], None]] = None) -> TeletextService: service = TeletextService() total_bytes = os.path.getsize(file_path) # Auto-detect alignment # Scan first 4096 bytes for the offset that yields the most Row 0 headers best_offset = 0 max_headers = 0 with open(file_path, 'rb') as f: head_data = f.read(4096 + 42) for offset in range(42): headers = 0 for i in range(offset, len(head_data) - 42, 42): chunk = head_data[i:i+42] try: # We check if Mag/Row decode to something sensible # AND if it's Row 0, we check if Page Num decodes without errors (ideally) # But for now, just counting Row 0s is good enough. mag, row = decode_packet_header(chunk[0], chunk[1]) if row == 0 and mag != 8: # Mag 8 is often just nulls headers += 1 except: pass if headers > max_headers: max_headers = headers best_offset = offset # Each packet is 42 bytes total_packets = (total_bytes - best_offset) // 42 processed_packets = 0 with open(file_path, 'rb') as f: f.seek(best_offset) while True: chunk = f.read(42) if not chunk: break if len(chunk) < 42: break processed_packets += 1 if progress_callback and processed_packets % 100 == 0: progress_callback(processed_packets, total_packets) packet = Packet(chunk) service.all_packets.append(packet) # Logic to group into pages. if packet.row == 0: # Start of a new page header. # Byte 2-9 of header contain Page Number, Subcode, Control bits etc. # We need to parse the header to identify the page. # Header format (after Mag/Row): # Bytes: P1 P2 S1 S2 S3 S4 C1 C2 ... # All Hamming 8/4 encoded. # For now, let's just create a new page entry for every Header we see, # or find the existing one if we want to support updates (but T42 usually is a stream capture). # If it's an editor file, it's likely sequential. p_num, sub_code, language = parse_header(packet.data) # Create new page new_page = Page(magazine=packet.magazine, page_number=p_num, sub_code=sub_code, language=language) new_page.packets.append(packet) service.pages.append(new_page) else: # Add to the "current" page of this magazine. # We need to track the current active page for each magazine. # A simplistic approach: add to the last page added that matches the magazine ?? # Robust approach: Maintain a dict of current_pages_by_magazine. # Let's find the last page in service that matches the packet's magazine # This is O(N) but N (pages) is small. target_page = None for p in reversed(service.pages): if p.magazine == packet.magazine: target_page = p break if target_page: target_page.packets.append(packet) else: # Packet without a header? Orphaned. Just keep in all_packets pass return service def encode_hamming_8_4(value): # Value is 4 bits (0-15) d1 = (value >> 0) & 1 d2 = (value >> 1) & 1 d3 = (value >> 2) & 1 d4 = (value >> 3) & 1 # Parity bits (Odd parity default? Or standard Hamming?) # Teletext spec: # P1 = 1 + D1 + D2 + D4 (mod 2) -> Inverse of even parity check? # Actually, simpler to look up or calculate. # Let's match typical implementation: # P1 (b0) covers 1,3,7 (D1, D2, D4) # P2 (b2) covers 1,5,7 (D1, D3, D4) # P3 (b4) covers 3,5,7 (D2, D3, D4) # P4 (b6) covers all. # Teletext uses ODD parity for the hamming bits usually? # "Hamming 8/4 with odd parity" p1 = 1 ^ d1 ^ d2 ^ d4 p2 = 1 ^ d1 ^ d3 ^ d4 p3 = 1 ^ d2 ^ d3 ^ d4 res = (p1 << 0) | (d1 << 1) | \ (p2 << 2) | (d2 << 3) | \ (p3 << 4) | (d3 << 5) | \ (d4 << 7) # P4 (bit 6) makes total bits odd # Count set bits so far set_bits = bin(res).count('1') p4 = 1 if (set_bits % 2 == 0) else 0 res |= (p4 << 6) return res def decode_packet_header(b1, b2): """ Decodes the Magazine and Row from the first 2 bytes of a T42 packet. """ d1 = decode_hamming_8_4(b1) d2 = decode_hamming_8_4(b2) mag = (d1 & 0b0111) if mag == 0: mag = 8 row = (d2 << 1) | ((d1 >> 3) & 1) return mag, row def save_t42(file_path: str, service: TeletextService, progress_callback: Optional[Callable[[int, int], None]] = None): total_packets = len(service.all_packets) processed = 0 with open(file_path, 'wb') as f: for packet in service.all_packets: processed += 1 if progress_callback and processed % 100 == 0: progress_callback(processed, total_packets) # Check if we can reuse the original header (preserving parity/integrity) use_original_header = False if hasattr(packet, 'original_data') and len(packet.original_data) >= 2: # Try to decode the original header try: orig_mag, orig_row = decode_packet_header(packet.original_data[0], packet.original_data[1]) if orig_mag == packet.magazine and orig_row == packet.row: use_original_header = True except: pass if use_original_header: header = packet.original_data[:2] else: # Reconstruct header bytes mag = packet.magazine if mag == 8: mag = 0 # 0 encoded as 8 # Bits: # B1 data: M1(0) M2(1) M3(2) R1(3) m1 = (mag >> 0) & 1 m2 = (mag >> 1) & 1 m3 = (mag >> 2) & 1 r1 = (packet.row >> 0) & 1 b1_val = m1 | (m2 << 1) | (m3 << 2) | (r1 << 3) b1_enc = encode_hamming_8_4(b1_val) # B2 data: R2(0) R3(1) R4(2) R5(3) r2 = (packet.row >> 1) & 1 r3 = (packet.row >> 2) & 1 r4 = (packet.row >> 3) & 1 r5 = (packet.row >> 4) & 1 b2_val = r2 | (r3 << 1) | (r4 << 2) | (r5 << 3) b2_enc = encode_hamming_8_4(b2_val) header = bytes([b1_enc, b2_enc]) f.write(header + packet.data) def parse_header(data: bytearray): # Data is 40 bytes. # Bytes 0-7 are Page Num (2), Subcode (4), Control (2) - ALL Hamming encoded. # 0: Page Units (PU) # 1: Page Tens (PT) pu = decode_hamming_8_4(data[0]) pt = decode_hamming_8_4(data[1]) # Use BCD/Hex-like storage: High nibble is Tens, Low nibble is Units. # This preserves Hex pages (A-F) without colliding with decimal pages. # E.g. Page 1FF -> Tens=F(15), Units=F(15) -> 0xFF (255) # Page 12E -> Tens=2, Units=E(14) -> 0x2E (46) # Page 134 -> Tens=3, Units=4 -> 0x34 (52) # 0x2E != 0x34. No collision. page_num = ((pt & 0xF) << 4) | (pu & 0xF) # Subcode: S1, S2, S3, S4 # S1 (low), S2, S3, S4 (high) s1 = decode_hamming_8_4(data[2]) s2 = decode_hamming_8_4(data[3]) s3 = decode_hamming_8_4(data[4]) s4 = decode_hamming_8_4(data[5]) # Subcode logic is a bit complex with specific bit mapping for "Time" vs "Subcode" # But usually just combining them gives the raw subcode value. # S1: bits 0-3 # S2: bits 4-6 (bit 4 is C4) -> actually S2 has 3 bits of subcode + 1 control bit usually? # Let's simplify and just concat them for a unique identifier. sub_code = s1 | (s2 << 4) | (s3 << 8) | (s4 << 12) # Control bits C12, C13, C14 are in Byte 8 (index 8) # They determine the National Option (Language) c_bits_2 = decode_hamming_8_4(data[8]) # Fix for Language Detection: # It seems C12 and C13 are swapped in the Hamming decoding or file format relative to expected values. # C12 is bit 0, C13 is bit 1. # We swap them so D1 maps to C13 (Swedish bit) and D2 maps to C12 (German bit). # Original: language = c_bits_2 & 0b111 language = ((c_bits_2 & 1) << 1) | ((c_bits_2 & 2) >> 1) | (c_bits_2 & 4) return page_num, sub_code, language def save_tti(file_path: str, page: Page): """ Saves a single Page object to a TTI file. """ with open(file_path, 'w', encoding='latin-1') as f: # Header Info # DS - Source? Description? f.write(f"DS,Teletext Editor Export\n") f.write(f"SP,{file_path}\n") # PN - Page Number mpp00 # Typically TTI uses decimal integer for mppss? # Or mppss as hex digits? # Standard convention: mppss where m is 1-8, pp is 00-FF, ss is 00-99 # Example: Page 100 -> PN,10000 # Example: Page 1F0 -> PN,1F000 f.write(f"PN,{page.magazine}{page.page_number:02X}00\n") # SC - Subcode ssss f.write(f"SC,{page.sub_code:04X}\n") # PS - Page Status # 8000 is typical for "Transmission" f.write(f"PS,8000\n") # RE - Region (Language) f.write(f"RE,{page.language}\n") # Lines # We need to construct the 40-char string for each row # Row 0 is special (Header) # Get all packets for this page # Map row -> packet rows = {} for p in page.packets: rows[p.row] = p for r in range(26): # 0 to 25 if r in rows: packet = rows[r] # Packet data is 40 bytes (after the 2-byte header we stripped in Packet class? No wait) # Packet.data in our model IS the 40 bytes of character data (we strip MRAG in __post_init__) # So we just decode it as latin-1 to get the chars # However, we must ensure we don't have newlines or nulls breaking the text file structure # TTI format usually accepts raw bytes 0x00-0xFF if strictly handled, but often expects # mapped control codes. # Standard VBIT2 TTI handling treats it as a binary-safe string if mapped to char. # Row 0 special handling: The first 8 bytes of Row 0 are usually header flags in the packet, # but visually they are "P100 ". # TTI usually expects the visual representation for the line content. # But for transmission, we want the raw bytes. # OL,r,String data_str = packet.data.decode('latin-1') f.write(f"OL,{r},{data_str}\n") else: # Empty line? Usually omitted or written as empty pass