hello-world.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
If this outputs anything, the universe allowed it.
"""

#############################
# IMPORT THE LAST REMAINS
#############################

import ast, ctypes, hashlib, gc, atexit
import multiprocessing, unicodedata
import pickle, gzip, base64
from abc import ABC, abstractmethod
from typing import Any, Protocol, TypeVar, Generic, Optional
from dataclasses import dataclass, field, replace as dataclass_replace
from enum import Enum
import json, tempfile, threading, os, pathlib

#############################
# CONFIGURATION (DRY)
#############################

@dataclass(frozen=True)
class SystemConfig:
    """Single source of truth for system configuration"""
    program_source: str = '"Hello" + ", " + "World"'
    normalization_form: str = "NFC"
    consensus_nodes: int = 2
    hash_algorithm: str = "sha256"
    encoding: str = "utf-8"
    compression_level: int = 9
    log_file_path: Optional[str] = None

DEFAULT_CONFIG = SystemConfig()

#############################
# SOLID: INTERFACE SEGREGATION
# Define minimal interfaces
#############################

class IEvaluator(ABC):
    """Single Responsibility: Evaluate AST nodes"""
    @abstractmethod
    def eval(self, node: ast.AST) -> Any:
        pass

class IEncoder(ABC):
    """Single Responsibility: Encode/decode messages"""
    @abstractmethod
    def encode(self, msg: Any) -> bytes:
        pass

    @abstractmethod
    def decode(self, blob: bytes) -> Any:
        pass

class INormalizer(ABC):
    """Single Responsibility: Normalize text"""
    @abstractmethod
    def normalize(self, text: str) -> str:
        pass

class IOutputWriter(ABC):
    """Single Responsibility: Write output"""
    @abstractmethod
    def write(self, message: str) -> None:
        pass

class IHashStrategy(ABC):
    """Single Responsibility: Hash computation"""
    @abstractmethod
    def compute_hash(self, data: str) -> bytes:
        pass

class ITransactionLog(ABC):
    """ACID: Transaction logging"""
    @abstractmethod
    def begin_transaction(self, tx_id: str) -> None:
        pass

    @abstractmethod
    def commit_transaction(self, tx_id: str, result: Any) -> None:
        pass

    @abstractmethod
    def rollback_transaction(self, tx_id: str) -> None:
        pass

#############################
# SOLID: SINGLE RESPONSIBILITY
# Each class has one reason to change
#############################

class TinyVM(IEvaluator):
    """Liskov Substitution: Can be replaced by any IEvaluator"""

    def eval(self, node: ast.AST) -> Any:
        if isinstance(node, ast.Module):
            return self._eval_module(node)
        if isinstance(node, ast.Expr):
            return self._eval_expression(node)
        if isinstance(node, ast.Constant):
            return self._eval_constant(node)
        if isinstance(node, ast.BinOp):
            return self._eval_binop(node)
        raise RuntimeError(f"Unsupported AST: {ast.dump(node)}")

    def _eval_module(self, node: ast.Module) -> Any:
        """DRY: Extracted method"""
        val = None
        for stmt in node.body:
            val = self.eval(stmt)
        return val

    def _eval_expression(self, node: ast.Expr) -> Any:
        """DRY: Extracted method"""
        return self.eval(node.value)

    def _eval_constant(self, node: ast.Constant) -> Any:
        """DRY: Extracted method"""
        return node.value

    def _eval_binop(self, node: ast.BinOp) -> Any:
        """DRY: Extracted method, Open/Closed: extensible"""
        if isinstance(node.op, ast.Add):
            return self.eval(node.left) + self.eval(node.right)
        raise RuntimeError(f"Unsupported operation: {ast.dump(node.op)}")

class CompressionEncoder(IEncoder):
    """Dependency Inversion: Depends on abstraction"""

    def __init__(self, compression_level: int = 9):
        self._compression_level = compression_level

    def encode(self, msg: Any) -> bytes:
        """DRY: Reusable encoding logic"""
        blob = gzip.compress(
            pickle.dumps(msg),
            compresslevel=self._compression_level
        )
        return base64.b64encode(blob)

    def decode(self, blob: bytes) -> Any:
        """DRY: Reusable decoding logic"""
        return pickle.loads(gzip.decompress(base64.b64decode(blob)))

class UnicodeNormalizer(INormalizer):
    """Single Responsibility: Unicode normalization only"""

    def __init__(self, form: str = "NFC"):
        self._form = form

    def normalize(self, text: str) -> str:
        return unicodedata.normalize(self._form, text)

class COutputWriter(IOutputWriter):
    """Single Responsibility: C-level output only"""

    def __init__(self):
        self._libc = ctypes.CDLL(None)
        self._puts = self._libc.puts
        self._puts.argtypes = [ctypes.c_char_p]

    def write(self, message: str) -> None:
        self._puts(message.encode())

class SHA256HashStrategy(IHashStrategy):
    """Open/Closed: Can add new hash strategies without modifying existing code"""

    def compute_hash(self, data: str) -> bytes:
        return hashlib.sha256(data.encode()).digest()

class FileTransactionLog(ITransactionLog):
    """ACID: Atomicity, Consistency, Isolation, Durability

    Multiprocessing-safe implementation using file path instead of file handle.
    Each process opens the file independently.
    """

    def __init__(self, log_path: Optional[str] = None):
        if log_path is None:
            fd, log_path = tempfile.mkstemp(suffix='.txlog', text=True)
            os.close(fd)  # Close fd, we'll open by path

        self._log_path = log_path
        self._lock = multiprocessing.Lock()  # Process-safe lock
        atexit.register(self._cleanup)

    def begin_transaction(self, tx_id: str) -> None:
        """ACID: Durability - Write to disk"""
        with self._lock:  # ACID: Isolation (process-safe)
            entry = {"action": "BEGIN", "tx_id": tx_id}
            self._write_entry(entry)

    def commit_transaction(self, tx_id: str, result: Any) -> None:
        """ACID: Atomicity - All or nothing"""
        with self._lock:  # ACID: Isolation (process-safe)
            entry = {
                "action": "COMMIT",
                "tx_id": tx_id,
                "result": str(result)
            }
            self._write_entry(entry)

    def rollback_transaction(self, tx_id: str) -> None:
        """ACID: Consistency - Maintain valid state"""
        with self._lock:  # ACID: Isolation (process-safe)
            entry = {"action": "ROLLBACK", "tx_id": tx_id}
            self._write_entry(entry)

    def _write_entry(self, entry: dict) -> None:
        """DRY: Centralized write logic"""
        with open(self._log_path, 'a') as f:
            f.write(json.dumps(entry) + "\n")
            f.flush()
            os.fsync(f.fileno())  # Force write to disk

    def _cleanup(self):
        """DRY: Centralized cleanup"""
        try:
            if os.path.exists(self._log_path):
                os.unlink(self._log_path)
        except Exception:
            pass  # Best effort cleanup

#############################
# MULTIPROCESSING WORKER
# Standalone function for process boundaries
#############################

def _consensus_worker(
    result_queue: multiprocessing.Queue,
    node_id: int,
    config: SystemConfig
) -> None:
    """Standalone worker function that can be pickled for multiprocessing.

    ACID: Execute with transaction support.
    SOLID: Dependency Inversion - recreates dependencies in child process.
    """
    # Recreate all dependencies in the child process
    tx_log = FileTransactionLog(config.log_file_path)
    interpreter = InterpreterService(
        ComponentFactory.create_evaluator(),
        ComponentFactory.create_encoder(config),
        ComponentFactory.create_normalizer(config),
        config
    )

    tx_id = f"node_{node_id}"
    tx_log.begin_transaction(tx_id)

    try:
        result = interpreter.interpret(config.program_source)
        tx_log.commit_transaction(tx_id, result)
        result_queue.put(result)
    except Exception as e:
        tx_log.rollback_transaction(tx_id)
        raise RuntimeError(f"Node {node_id} failed: {e}")

#############################
# SOLID: DEPENDENCY INVERSION
# High-level modules depend on abstractions
#############################

class InterpreterService:
    """Orchestrates interpretation with dependency injection"""

    def __init__(
        self,
        evaluator: IEvaluator,
        encoder: IEncoder,
        normalizer: INormalizer,
        config: SystemConfig = DEFAULT_CONFIG
    ):
        # Dependency Inversion: Depend on abstractions
        self._evaluator = evaluator
        self._encoder = encoder
        self._normalizer = normalizer
        self._config = config

    def interpret(self, source: str) -> str:
        """DRY: Single interpretation logic"""
        tree = ast.parse(source)
        raw_result = self._evaluator.eval(tree)
        encoded = self._encoder.encode(raw_result)
        decoded = self._encoder.decode(encoded)
        return self._normalizer.normalize(decoded)

class ConsensusEngine:
    """Manages consensus with ACID guarantees

    Multiprocessing-safe implementation using standalone worker functions.
    """

    def __init__(
        self,
        hash_strategy: IHashStrategy,
        config: SystemConfig = DEFAULT_CONFIG
    ):
        self._hash_strategy = hash_strategy
        self._config = config

    def achieve_consensus(self) -> str:
        """ACID: Consistency - All nodes must agree"""
        queues = [multiprocessing.Queue() for _ in range(self._config.consensus_nodes)]
        processes = []

        # Start all consensus nodes with standalone worker
        for i, queue in enumerate(queues):
            process = multiprocessing.Process(
                target=_consensus_worker,
                args=(queue, i, self._config)
            )
            processes.append(process)
            process.start()

        # Wait for all nodes
        for process in processes:
            process.join()

        # Collect results
        results = [queue.get() for queue in queues]

        # Verify consensus (ACID: Consistency)
        hashes = [self._hash_strategy.compute_hash(r) for r in results]
        if not all(h == hashes[0] for h in hashes):
            raise RuntimeError("Multiverse disagreement: Consensus failed")

        return results[0]

#############################
# SOLID: FACTORY PATTERN
# Open/Closed: Add new implementations without modifying factory
#############################

class ComponentFactory:
    """DRY: Centralized component creation"""

    @staticmethod
    def create_evaluator() -> IEvaluator:
        return TinyVM()

    @staticmethod
    def create_encoder(config: SystemConfig) -> IEncoder:
        return CompressionEncoder(config.compression_level)

    @staticmethod
    def create_normalizer(config: SystemConfig) -> INormalizer:
        return UnicodeNormalizer(config.normalization_form)

    @staticmethod
    def create_output_writer() -> IOutputWriter:
        return COutputWriter()

    @staticmethod
    def create_hash_strategy() -> IHashStrategy:
        return SHA256HashStrategy()

    @staticmethod
    def create_transaction_log(log_path: Optional[str] = None) -> ITransactionLog:
        return FileTransactionLog(log_path)

    @staticmethod
    def create_interpreter_service(config: SystemConfig) -> InterpreterService:
        """DRY: Assemble dependencies in one place"""
        return InterpreterService(
            ComponentFactory.create_evaluator(),
            ComponentFactory.create_encoder(config),
            ComponentFactory.create_normalizer(config),
            config
        )

    @staticmethod
    def create_consensus_engine(config: SystemConfig) -> ConsensusEngine:
        """DRY: Assemble dependencies in one place"""
        return ConsensusEngine(
            ComponentFactory.create_hash_strategy(),
            config
        )

#############################
# APPLICATION FACADE
# SOLID: Simplified interface to complex subsystem
#############################

class UniverseSimulator:
    """High-level facade with ACID transaction support"""

    def __init__(self, config: SystemConfig = DEFAULT_CONFIG):
        # Create shared log file for all processes
        if config.log_file_path is None:
            fd, log_path = tempfile.mkstemp(suffix='.txlog', text=True)
            os.close(fd)
            # Create new config with log path
            config = dataclass_replace(config, log_file_path=log_path)

        self._config = config
        self._consensus_engine = ComponentFactory.create_consensus_engine(config)
        self._output_writer = ComponentFactory.create_output_writer()

    def collapse_reality(self) -> None:
        """Main execution with full ACID guarantees"""
        try:
            result = self._consensus_engine.achieve_consensus()
            self._output_writer.write(result)
        except Exception as e:
            raise RuntimeError(f"Reality collapse failed: {e}")

#############################
# EXIT RITUAL
#############################

@atexit.register
def clean():
    """DRY: Centralized cleanup"""
    gc.collect()

#############################
# MAIN: COLLAPSE REALITY
#############################

def main():
    """Dependency Inversion: Program to abstractions"""
    simulator = UniverseSimulator(DEFAULT_CONFIG)
    simulator.collapse_reality()

#############################
# DO NOT LOOK AWAY
#############################

if __name__ == "__main__":
    main()