Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 4)
b41bee3 verified | #!/usr/bin/env python3 | |
| """ | |
| Aureon Harmonic Binary Protocol | |
| -------------------------------- | |
| Provides a compact binary transport for harmonic thoughts. Messages are encoded as: | |
| • 2-byte magic header (HB) | |
| • 1-byte version | |
| • 5-byte control header (type/direction/grade/coherence/confidence/symbol flags) | |
| • Variable-length body containing packed harmonic frequency/amplitude samples | |
| This enables the ThoughtBus and Harmonic Signal Chain to exchange compact payloads | |
| without losing the sacred frequency context. | |
| 👑💰 QUEEN'S SACRED 1.88% LAW - ENCODED IN ALL BINARY TRANSMISSIONS 💰👑 | |
| • MIN_COP = 1.0188 (1.88% minimum realized profit) | |
| • All execution decisions must carry the Queen's profit mandate | |
| • The sacred number is encoded as flag 0x88 in the binary header | |
| """ | |
| from __future__ import annotations | |
| from aureon.core.aureon_baton_link import link_system as _baton_link; _baton_link(__name__) | |
| import base64 | |
| import struct | |
| import zlib | |
| from dataclasses import dataclass | |
| from enum import IntEnum | |
| from typing import Iterable, List, Optional, Tuple | |
| from aureon.harmonic.aureon_harmonic_alphabet import HarmonicAlphabet, HarmonicTone | |
| # 👑 QUEEN'S SACRED 1.88% LAW - THE PROTOCOL KNOWS! | |
| QUEEN_MIN_COP = 1.0188 # 1.88% minimum realized profit | |
| QUEEN_MIN_PROFIT_PCT = 1.88 # The sacred number as percentage | |
| QUEEN_PROFIT_FLAG = 0x88 # The sacred flag in binary headers (1.88 → 0x88) | |
| MAGIC = b"HB" | |
| PROTOCOL_VERSION = 1 | |
| _alphabet = HarmonicAlphabet() | |
| class BinaryMessageType(IntEnum): | |
| UNDEFINED = 0 | |
| MARKET_SCAN = 1 | |
| EXECUTION_DECISION = 2 | |
| OUTCOME_SIGNAL = 3 | |
| TELEMETRY = 4 | |
| ALERT = 5 | |
| QUEEN_PROFIT_GATE = 6 # 👑 NEW! - Message type for 1.88% profit gate | |
| QUEEN_PROFIT_APPROVED = 7 # 👑 NEW! - Queen approved trade (≥1.88%) | |
| QUEEN_PROFIT_BLOCKED = 8 # 👑 NEW! - Queen blocked trade (<1.88%) | |
| class BinaryDirection(IntEnum): | |
| DOWN = 0 # Queen → Whale | |
| UP = 1 # Whale → Queen | |
| LATERAL = 2 | |
| class BinaryHeader: | |
| message_type: BinaryMessageType = BinaryMessageType.UNDEFINED | |
| direction: BinaryDirection = BinaryDirection.DOWN | |
| grade: int = 0 # 0-15 | |
| coherence: int = 15 # 0-15 nibble | |
| confidence: int = 15 # 0-15 nibble | |
| flags: int = 0 # 👑 Use QUEEN_PROFIT_FLAG (0x88) to indicate 1.88% compliance | |
| symbol_id: int = 0 | |
| def pack(self) -> bytes: | |
| b0 = self.message_type & 0xFF | |
| b1 = ((self.direction & 0x0F) << 4) | (self.grade & 0x0F) | |
| b2 = ((self.coherence & 0x0F) << 4) | (self.confidence & 0x0F) | |
| b3 = self.flags & 0xFF | |
| b4 = self.symbol_id & 0xFF | |
| return bytes([b0, b1, b2, b3, b4]) | |
| def unpack(cls, data: bytes) -> "BinaryHeader": | |
| if len(data) < 5: | |
| raise ValueError("Binary header requires 5 bytes") | |
| b0, b1, b2, b3, b4 = data[:5] | |
| direction = BinaryDirection((b1 >> 4) & 0x0F) | |
| grade = b1 & 0x0F | |
| coherence = (b2 >> 4) & 0x0F | |
| confidence = b2 & 0x0F | |
| return cls( | |
| message_type=BinaryMessageType(b0), | |
| direction=direction, | |
| grade=grade, | |
| coherence=coherence, | |
| confidence=confidence, | |
| flags=b3, | |
| symbol_id=b4, | |
| ) | |
| class HarmonicBinaryPacket: | |
| header: BinaryHeader | |
| body: bytes | |
| def to_bytes(self) -> bytes: | |
| return MAGIC + bytes([PROTOCOL_VERSION]) + self.header.pack() + self.body | |
| def from_bytes(cls, payload: bytes) -> "HarmonicBinaryPacket": | |
| if len(payload) < len(MAGIC) + 1 + 5: | |
| raise ValueError("Payload too small for harmonic binary packet") | |
| if payload[:2] != MAGIC: | |
| raise ValueError("Invalid harmonic binary magic") | |
| version = payload[2] | |
| if version != PROTOCOL_VERSION: | |
| raise ValueError(f"Unsupported harmonic binary version: {version}") | |
| header = BinaryHeader.unpack(payload[3:8]) | |
| body = payload[8:] | |
| return cls(header=header, body=body) | |
| def to_base64(self) -> str: | |
| return base64.b64encode(self.to_bytes()).decode("ascii") | |
| def from_base64(cls, data: str) -> "HarmonicBinaryPacket": | |
| return cls.from_bytes(base64.b64decode(data.encode("ascii"))) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Symbol Table Helpers | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Integration with Persistent Symbol Table | |
| try: | |
| from aureon.harmonic.aureon_harmonic_symbol_table import get_symbol_id, get_symbol_from_id | |
| except ImportError: | |
| # Fallback if module missing (e.g. during bootstrap) | |
| def get_symbol_id(symbol: str) -> int: | |
| if not symbol: return 0 | |
| return zlib.crc32(symbol.encode("utf-8")) & 0xFF | |
| def get_symbol_from_id(sid: int) -> Optional[str]: | |
| return None | |
| def compute_symbol_id(symbol: Optional[str]) -> int: | |
| if not symbol: | |
| return 0 | |
| return get_symbol_id(symbol) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Encoding / Decoding Helpers | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| FREQUENCY_SCALE = 10 # tenths of Hz | |
| AMPLITUDE_SCALE = 100 # hundredths | |
| TONE_RECORD_SIZE = 3 # 2 bytes freq + 1 byte amp | |
| def _pack_tones(tones: Iterable[HarmonicTone]) -> bytes: | |
| body = bytearray() | |
| for tone in tones: | |
| freq_val = max(0, min(0xFFFF, int(round(tone.frequency * FREQUENCY_SCALE)))) | |
| amp = tone.amplitude | |
| if amp < 0: | |
| amp = 0.0 | |
| amp_val = max(0, min(0xFF, int(round(min(2.55, amp) * AMPLITUDE_SCALE)))) | |
| body.extend(struct.pack(">H", freq_val)) | |
| body.append(amp_val) | |
| return bytes(body) | |
| def _unpack_tones(data: bytes) -> List[Tuple[float, float]]: | |
| tones: List[Tuple[float, float]] = [] | |
| if len(data) % TONE_RECORD_SIZE != 0: | |
| raise ValueError("Invalid tone body length") | |
| for offset in range(0, len(data), TONE_RECORD_SIZE): | |
| freq_raw, = struct.unpack(">H", data[offset:offset + 2]) | |
| amp_raw = data[offset + 2] | |
| tones.append((freq_raw / FREQUENCY_SCALE, amp_raw / AMPLITUDE_SCALE)) | |
| return tones | |
| def encode_text_packet( | |
| text: str, | |
| message_type: BinaryMessageType = BinaryMessageType.UNDEFINED, | |
| direction: BinaryDirection = BinaryDirection.DOWN, | |
| grade: int = 0, | |
| coherence: float = 1.0, | |
| confidence: float = 1.0, | |
| flags: int = 0, | |
| symbol: Optional[str] = None, | |
| ) -> HarmonicBinaryPacket: | |
| tones = _alphabet.encode_text(text) | |
| coherence_nibble = max(0, min(15, int(round(coherence * 15)))) | |
| confidence_nibble = max(0, min(15, int(round(confidence * 15)))) | |
| header = BinaryHeader( | |
| message_type=message_type, | |
| direction=direction, | |
| grade=max(0, min(15, grade)), | |
| coherence=coherence_nibble, | |
| confidence=confidence_nibble, | |
| flags=flags, | |
| symbol_id=compute_symbol_id(symbol), | |
| ) | |
| body = _pack_tones(tones) | |
| return HarmonicBinaryPacket(header=header, body=body) | |
| def decode_packet(packet_bytes: bytes) -> Tuple[BinaryHeader, str]: | |
| packet = HarmonicBinaryPacket.from_bytes(packet_bytes) | |
| tones = _unpack_tones(packet.body) | |
| decoded = _alphabet.decode_signal(tones) | |
| return packet.header, decoded | |
| def decode_packet_from_base64(data: str) -> Tuple[BinaryHeader, str]: | |
| packet = HarmonicBinaryPacket.from_base64(data) | |
| tones = _unpack_tones(packet.body) | |
| decoded = _alphabet.decode_signal(tones) | |
| return packet.header, decoded |