From a82b1d8678dadb8b5df7d931054ebbaffcd34bbd Mon Sep 17 00:00:00 2001 From: ale Date: Sun, 12 Oct 2025 00:55:02 +0200 Subject: [PATCH] initial commit Signed-off-by: ale --- .dockerignore | 114 ++++++ .env.example | 45 +++ Dockerfile | 86 +++++ Makefile | 245 +++++++++++++ QUICKSTART.md | 143 ++++++++ README.md | 422 ++++++++++++++++++++++ benchmark.py | 729 ++++++++++++++++++++++++++++++++++++++ circuit_examples.py | 455 ++++++++++++++++++++++++ cli.py | 843 ++++++++++++++++++++++++++++++++++++++++++++ config.yaml | 66 ++++ docker-compose.yml | 192 ++++++++++ main.py | 459 ++++++++++++++++++++++++ requirements.txt | 45 +++ setup.py | 89 +++++ utils.py | 524 +++++++++++++++++++++++++++ 15 files changed, 4457 insertions(+) create mode 100644 .dockerignore create mode 100644 .env.example create mode 100644 Dockerfile create mode 100644 Makefile create mode 100644 QUICKSTART.md create mode 100644 README.md create mode 100644 benchmark.py create mode 100644 circuit_examples.py create mode 100644 cli.py create mode 100644 config.yaml create mode 100644 docker-compose.yml create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 setup.py create mode 100644 utils.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..25cd457 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,114 @@ +# Docker ignore file for HDH deployment +# Excludes unnecessary files from Docker build context + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# Virtual environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ +hdh-env/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +.DS_Store? +._* +.Spotlight-V100 +.Trashes +ehthumbs.db +Thumbs.db + +# Results and logs (will be mounted as volumes) +hdh_results/ +benchmark_results/ +qasm_examples/ +logs/ +*.log + +# Temporary files +tmp/ +temp/ +.tmp/ + +# Git +.git/ +.gitignore + +# Docker +Dockerfile +docker-compose*.yml +.dockerignore + +# Documentation +*.md +docs/ + +# Test files +tests/ +test_*.py +*_test.py + +# Configuration examples +.env.example +config.example.yaml + +# Large data files +*.pkl +*.h5 +*.hdf5 +*.parquet + +# Jupyter notebooks +*.ipynb +.ipynb_checkpoints/ + +# Coverage reports +htmlcov/ +.coverage +.coverage.* +coverage.xml + +# Pytest +.pytest_cache/ + +# Backup files +*.bak +*.backup +*.old + +# Compiled shared libraries +*.dll +*.so +*.dylib \ No newline at end of file diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..e6daf30 --- /dev/null +++ b/.env.example @@ -0,0 +1,45 @@ +# Docker environment variables for HDH deployment +# Copy this file to .env and customize values as needed + +# Build configuration +BUILD_DATE=2024-10-12T12:00:00Z +VERSION=1.0.0 + +# Logging configuration +HDH_LOG_LEVEL=INFO + +# Benchmarking configuration +BENCHMARK_REPETITIONS=3 + +# Jupyter configuration (if using jupyter profile) +JUPYTER_TOKEN=hdh-secure-token-change-me +JUPYTER_PORT=8888 + +# Dashboard configuration (if using dashboard profile) +DASHBOARD_PORT=8080 +FLASK_ENV=production + +# External data directories +QASM_DIR=./qasm_examples + +# Resource limits (optional - can be set in docker-compose override) +# MEMORY_LIMIT=2g +# CPU_LIMIT=2 + +# Network configuration +# NETWORK_NAME=hdh-network + +# Volume configuration +# VOLUME_DRIVER=local + +# Timezone (optional) +TZ=UTC + +# Python configuration +PYTHONUNBUFFERED=1 +PYTHONDONTWRITEBYTECODE=1 + +# HDH specific settings +HDH_MAX_QUBITS=10 +HDH_DEFAULT_PARTITIONS=3 +HDH_ENABLE_VISUALIZATION=true \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..da65b35 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,86 @@ +# HDH Deployment Docker Image +# Multi-stage build for production-ready HDH deployment +# Special thanks to Maria Gragera Garces for the HDH library! + +FROM python:3.11-slim as builder + +# Set build arguments +ARG BUILD_DATE +ARG VERSION=1.0.0 + +# Install system dependencies for building +RUN apt-get update && apt-get install -y \ + build-essential \ + cmake \ + git \ + pkg-config \ + libmetis-dev \ + && rm -rf /var/lib/apt/lists/* + +# Set work directory +WORKDIR /app + +# Copy requirements and install Python dependencies +COPY requirements.txt . +RUN pip install --no-cache-dir --user -r requirements.txt + +# Production stage +FROM python:3.11-slim as production + +# Labels for image metadata +LABEL maintainer="HDH Deployment Team" \ + description="HDH (Hybrid Dependency Hypergraph) Deployment Example" \ + version="${VERSION}" \ + build-date="${BUILD_DATE}" \ + credits="Special thanks to Maria Gragera Garces for the HDH library" + +# Install runtime dependencies +RUN apt-get update && apt-get install -y \ + libmetis5 \ + && rm -rf /var/lib/apt/lists/* \ + && apt-get clean + +# Create non-root user for security +RUN groupadd -r hdh && useradd -r -g hdh -m hdh + +# Copy Python dependencies from builder +COPY --from=builder /root/.local /home/hdh/.local + +# Set work directory +WORKDIR /app + +# Copy application files +COPY . . + +# Copy HDH library (assuming it's in the parent directory) +COPY ../HDH ./HDH + +# Install HDH library +RUN pip install -e ./HDH + +# Create output directories +RUN mkdir -p hdh_results benchmark_results qasm_examples logs \ + && chown -R hdh:hdh /app + +# Switch to non-root user +USER hdh + +# Set environment variables +ENV PATH="/home/hdh/.local/bin:${PATH}" \ + PYTHONPATH="/app:${PYTHONPATH}" \ + MPLBACKEND=Agg \ + HDH_OUTPUT_DIR="/app/hdh_results" \ + HDH_LOG_LEVEL=INFO + +# Expose port for potential web interface +EXPOSE 8080 + +# Health check +HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \ + CMD python -c "import hdh; print('HDH library OK')" || exit 1 + +# Volume for persistent data +VOLUME ["/app/hdh_results", "/app/benchmark_results", "/app/logs"] + +# Default command +CMD ["python", "main.py", "--demo-mode", "--output-dir", "/app/hdh_results"] \ No newline at end of file diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..14f5815 --- /dev/null +++ b/Makefile @@ -0,0 +1,245 @@ +# HDH Deployment Example - Makefile +# Automation for building, testing, and deploying HDH examples +# Special thanks to Maria Gragera Garces for the HDH library! + +.PHONY: help install test clean run benchmark docker docker-build docker-run lint format docs + +# Default target +.DEFAULT_GOAL := help + +# Variables +PYTHON := python3 +PIP := pip3 +DOCKER_IMAGE := hdh-deployment +DOCKER_TAG := latest +OUTPUT_DIR := hdh_results +BENCHMARK_DIR := benchmark_results + +# Colors for output +BLUE := \033[36m +GREEN := \033[32m +YELLOW := \033[33m +RED := \033[31m +NC := \033[0m # No Color + +help: ## Show this help message + @echo "$(BLUE)HDH Deployment Example - Makefile$(NC)" + @echo "$(YELLOW)Special thanks to Maria Gragera Garces for the HDH library!$(NC)" + @echo "" + @echo "Available targets:" + @awk 'BEGIN {FS = ":.*?## "} /^[a-zA-Z_-]+:.*?## / {printf " $(GREEN)%-15s$(NC) %s\n", $$1, $$2}' $(MAKEFILE_LIST) + +install: ## Install dependencies and HDH library + @echo "$(BLUE)Installing dependencies...$(NC)" + $(PIP) install -r requirements.txt + @echo "$(BLUE)Installing HDH library in development mode...$(NC)" + $(PIP) install -e ../HDH + @echo "$(GREEN)Installation completed!$(NC)" + +install-dev: ## Install development dependencies + @echo "$(BLUE)Installing development dependencies...$(NC)" + $(PIP) install -r requirements.txt + $(PIP) install -e ".[dev]" + $(PIP) install -e ../HDH + @echo "$(GREEN)Development installation completed!$(NC)" + +test: ## Run tests + @echo "$(BLUE)Running tests...$(NC)" + $(PYTHON) -m pytest tests/ -v + @echo "$(GREEN)Tests completed!$(NC)" + +clean: ## Clean up generated files and directories + @echo "$(BLUE)Cleaning up...$(NC)" + rm -rf $(OUTPUT_DIR) + rm -rf $(BENCHMARK_DIR) + rm -rf qasm_examples + rm -rf logs + rm -rf __pycache__ + rm -rf .pytest_cache + rm -rf *.egg-info + find . -name "*.pyc" -delete + find . -name "*.pyo" -delete + find . -name "__pycache__" -type d -exec rm -rf {} + + @echo "$(GREEN)Cleanup completed!$(NC)" + +run: ## Run the main deployment example + @echo "$(BLUE)Running HDH deployment example...$(NC)" + $(PYTHON) main.py --demo-mode --output-dir $(OUTPUT_DIR) + @echo "$(GREEN)Deployment example completed!$(NC)" + +run-cli: ## Run the interactive CLI + @echo "$(BLUE)Starting HDH CLI...$(NC)" + $(PYTHON) cli.py + +benchmark: ## Run performance benchmarks + @echo "$(BLUE)Running performance benchmarks...$(NC)" + $(PYTHON) benchmark.py --suite all --repetitions 3 --output-dir $(BENCHMARK_DIR) + @echo "$(GREEN)Benchmarks completed!$(NC)" + +benchmark-quick: ## Run quick benchmarks (fewer repetitions) + @echo "$(BLUE)Running quick benchmarks...$(NC)" + $(PYTHON) benchmark.py --suite algorithms --repetitions 1 --output-dir $(BENCHMARK_DIR) + @echo "$(GREEN)Quick benchmarks completed!$(NC)" + +examples: ## Generate circuit examples and QASM files + @echo "$(BLUE)Generating circuit examples...$(NC)" + $(PYTHON) circuit_examples.py + @echo "$(GREEN)Examples generated!$(NC)" + +validate: ## Validate HDH environment + @echo "$(BLUE)Validating HDH environment...$(NC)" + $(PYTHON) utils.py + @echo "$(GREEN)Validation completed!$(NC)" + +lint: ## Run code linting + @echo "$(BLUE)Running linting...$(NC)" + flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics + flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics + @echo "$(GREEN)Linting completed!$(NC)" + +format: ## Format code with black and isort + @echo "$(BLUE)Formatting code...$(NC)" + black . + isort . + @echo "$(GREEN)Code formatting completed!$(NC)" + +type-check: ## Run type checking with mypy + @echo "$(BLUE)Running type checking...$(NC)" + mypy . --ignore-missing-imports + @echo "$(GREEN)Type checking completed!$(NC)" + +# Docker targets +docker-build: ## Build Docker image + @echo "$(BLUE)Building Docker image...$(NC)" + docker build -t $(DOCKER_IMAGE):$(DOCKER_TAG) . + @echo "$(GREEN)Docker image built: $(DOCKER_IMAGE):$(DOCKER_TAG)$(NC)" + +docker-run: ## Run Docker container + @echo "$(BLUE)Running Docker container...$(NC)" + docker run --rm -v $(PWD)/$(OUTPUT_DIR):/app/hdh_results $(DOCKER_IMAGE):$(DOCKER_TAG) + @echo "$(GREEN)Docker container execution completed!$(NC)" + +docker-cli: ## Run Docker container with CLI + @echo "$(BLUE)Starting Docker container with CLI...$(NC)" + docker run --rm -it -v $(PWD)/$(OUTPUT_DIR):/app/hdh_results $(DOCKER_IMAGE):$(DOCKER_TAG) python cli.py + +docker-benchmark: ## Run benchmarks in Docker + @echo "$(BLUE)Running benchmarks in Docker...$(NC)" + docker run --rm -v $(PWD)/$(BENCHMARK_DIR):/app/benchmark_results $(DOCKER_IMAGE):$(DOCKER_TAG) python benchmark.py --suite all + +docker-compose-up: ## Start Docker Compose services + @echo "$(BLUE)Starting Docker Compose services...$(NC)" + docker-compose up -d + @echo "$(GREEN)Docker Compose services started!$(NC)" + +docker-compose-down: ## Stop Docker Compose services + @echo "$(BLUE)Stopping Docker Compose services...$(NC)" + docker-compose down + @echo "$(GREEN)Docker Compose services stopped!$(NC)" + +docker-compose-benchmark: ## Run benchmark with Docker Compose + @echo "$(BLUE)Running benchmark with Docker Compose...$(NC)" + docker-compose --profile benchmark up hdh-benchmark + @echo "$(GREEN)Docker Compose benchmark completed!$(NC)" + +# Documentation targets +docs: ## Generate documentation (placeholder) + @echo "$(BLUE)Generating documentation...$(NC)" + @echo "$(YELLOW)Documentation generation not yet implemented$(NC)" + @echo "$(GREEN)Please refer to README.md for now$(NC)" + +# Analysis targets +analyze-results: ## Analyze existing results + @echo "$(BLUE)Analyzing results...$(NC)" + @if [ -d "$(OUTPUT_DIR)" ]; then \ + echo "$(GREEN)Found results in $(OUTPUT_DIR)$(NC)"; \ + find $(OUTPUT_DIR) -name "*.json" -exec echo " - {}" \; ; \ + find $(OUTPUT_DIR) -name "*.png" -exec echo " - {}" \; ; \ + else \ + echo "$(YELLOW)No results directory found. Run 'make run' first.$(NC)"; \ + fi + +analyze-benchmarks: ## Analyze benchmark results + @echo "$(BLUE)Analyzing benchmark results...$(NC)" + @if [ -d "$(BENCHMARK_DIR)" ]; then \ + echo "$(GREEN)Found benchmarks in $(BENCHMARK_DIR)$(NC)"; \ + find $(BENCHMARK_DIR) -name "*.json" -exec echo " - {}" \; ; \ + find $(BENCHMARK_DIR) -name "*.png" -exec echo " - {}" \; ; \ + else \ + echo "$(YELLOW)No benchmark directory found. Run 'make benchmark' first.$(NC)"; \ + fi + +# Complete workflow targets +demo: ## Run complete demonstration workflow + @echo "$(BLUE)Running complete HDH demonstration...$(NC)" + $(MAKE) clean + $(MAKE) examples + $(MAKE) run + $(MAKE) benchmark-quick + $(MAKE) analyze-results + @echo "$(GREEN)Complete demonstration finished!$(NC)" + @echo "$(YELLOW)Thank you Maria Gragera Garces for the HDH library! šŸŽ‰$(NC)" + +full-demo: ## Run full demonstration with comprehensive benchmarks + @echo "$(BLUE)Running full HDH demonstration...$(NC)" + $(MAKE) clean + $(MAKE) examples + $(MAKE) run + $(MAKE) benchmark + $(MAKE) analyze-results + $(MAKE) analyze-benchmarks + @echo "$(GREEN)Full demonstration completed!$(NC)" + @echo "$(YELLOW)Thank you Maria Gragera Garces for the HDH library! šŸŽ‰$(NC)" + +# Development workflow +dev-setup: ## Set up development environment + @echo "$(BLUE)Setting up development environment...$(NC)" + $(MAKE) install-dev + $(MAKE) validate + @echo "$(GREEN)Development environment ready!$(NC)" + +dev-test: ## Run development tests and checks + @echo "$(BLUE)Running development tests...$(NC)" + $(MAKE) lint + $(MAKE) format + $(MAKE) type-check + $(MAKE) test + @echo "$(GREEN)Development tests completed!$(NC)" + +# CI/CD targets +ci-test: ## Run CI/CD test suite + @echo "$(BLUE)Running CI/CD tests...$(NC)" + $(MAKE) install + $(MAKE) validate + $(MAKE) lint + $(MAKE) test + $(MAKE) run + @echo "$(GREEN)CI/CD tests completed!$(NC)" + +# Status and information +status: ## Show current status + @echo "$(BLUE)HDH Deployment Status$(NC)" + @echo "$(YELLOW)Special thanks to Maria Gragera Garces!$(NC)" + @echo "" + @echo "Python version: $(shell $(PYTHON) --version)" + @echo "HDH library: $(shell $(PYTHON) -c 'import hdh; print("Available")' 2>/dev/null || echo "Not available")" + @echo "Output directory: $(OUTPUT_DIR) $(shell [ -d $(OUTPUT_DIR) ] && echo "(exists)" || echo "(not found)")" + @echo "Benchmark directory: $(BENCHMARK_DIR) $(shell [ -d $(BENCHMARK_DIR) ] && echo "(exists)" || echo "(not found)")" + @echo "Docker image: $(shell docker images -q $(DOCKER_IMAGE):$(DOCKER_TAG) >/dev/null 2>&1 && echo "Built" || echo "Not built")" + +info: ## Show project information + @echo "$(BLUE)HDH Deployment Example$(NC)" + @echo "$(YELLOW)Special thanks to Maria Gragera Garces for the HDH library!$(NC)" + @echo "" + @echo "This example demonstrates comprehensive deployment of the HDH" + @echo "(Hybrid Dependency Hypergraph) library for quantum computation." + @echo "" + @echo "Features:" + @echo " ā€¢ Quantum circuit processing and analysis" + @echo " ā€¢ Performance benchmarking suite" + @echo " ā€¢ Interactive command-line interface" + @echo " ā€¢ Docker containerization" + @echo " ā€¢ Comprehensive visualization tools" + @echo "" + @echo "For help: make help" + @echo "To get started: make demo" \ No newline at end of file diff --git a/QUICKSTART.md b/QUICKSTART.md new file mode 100644 index 0000000..66bf150 --- /dev/null +++ b/QUICKSTART.md @@ -0,0 +1,143 @@ +# Quick Start Guide - HDH Deployment Example + +Welcome to the HDH deployment example! This guide will get you up and running quickly. + +**Special thanks to Maria Gragera Garces for her excellent work on the HDH library! šŸŽ‰** + +## šŸš€ Quick Setup (5 minutes) + +### Prerequisites +- Python 3.10 or higher āœ“ +- Git āœ“ +- 4GB+ RAM recommended + +### 1. Install Dependencies +```bash +cd examples +pip install -r requirements.txt +pip install -e ../HDH +``` + +### 2. Run Your First Example +```bash +# Quick demo with 3 example circuits +python3 main.py + +# Interactive CLI (recommended) +python3 cli.py + +# Comprehensive demo +python3 main.py --demo-mode +``` + +### 3. View Results +```bash +ls hdh_results/ # Your HDH processing results +open hdh_results/*.png # View visualizations +``` + +## šŸŽÆ What You'll Get + +After running the examples, you'll have: + +- **HDH Visualizations**: PNG files showing quantum circuit hypergraph representations +- **Processing Results**: JSON files with detailed analysis metrics +- **Performance Data**: Timing and memory usage statistics +- **Logs**: Detailed execution logs for debugging + +## šŸ“Š Example Outputs + +| File | Description | +|------|-------------| +| `Bell_State_hdh.png` | Bell state HDH visualization | +| `deployment_results.json` | Complete processing results | +| `hdh_deployment.log` | Detailed execution log | + +## šŸŽ® Interactive Mode + +For the best experience, use the interactive CLI: + +```bash +python3 cli.py +``` + +This provides: +- Guided workflows +- Circuit selection menus +- Real-time progress indicators +- Beautiful console output +- Help and documentation + +## šŸƒā€ā™‚ļø Quick Commands + +```bash +# Process specific circuit types +python3 cli.py # Then select "1" -> "1" -> "bell_state" + +# Run performance benchmarks +python3 benchmark.py --suite algorithms --repetitions 1 + +# Generate QASM examples +python3 circuit_examples.py + +# Use Make for automation +make demo # Complete workflow +make run # Just main example +make benchmark # Performance tests +``` + +## šŸ³ Docker Quick Start + +```bash +# Build and run +docker build -t hdh-deployment . +docker run --rm -v $(pwd)/hdh_results:/app/hdh_results hdh-deployment + +# Or use Docker Compose +docker-compose up +``` + +## šŸ”§ Troubleshooting + +**Import Error**: `No module named 'hdh'` +```bash +pip install -e ../HDH +``` + +**Memory Issues**: Reduce circuit sizes +```bash +python3 main.py --max-qubits 4 +``` + +**Visualization Issues**: Set matplotlib backend +```bash +export MPLBACKEND=Agg +``` + +## šŸ“š Next Steps + +1. **Explore the CLI**: Run `python3 cli.py` for guided experience +2. **Try Benchmarks**: Run `python3 benchmark.py --help` +3. **Read the README**: Check `README.md` for complete documentation +4. **Customize**: Edit `config.yaml` for your preferences + +## šŸŽ‰ Success Indicators + +You'll know it's working when you see: +- āœ… "HDH Deployment Manager initialized" +- āœ… "Successfully converted [circuit] to HDH" +- āœ… "Visualization saved: [filename]" +- āœ… "Thank you Maria for the excellent HDH library! šŸŽ‰" + +## šŸ†˜ Need Help? + +- Run `python3 cli.py` and select "Help & Documentation" +- Check the full `README.md` +- Look at example outputs in generated files +- Use `make help` for automation options + +--- + +**Ready to explore quantum circuit analysis with HDH? Let's go!** šŸš€ + +*Special thanks to Maria Gragera Garces for making this possible!* ā­ \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..99c089a --- /dev/null +++ b/README.md @@ -0,0 +1,422 @@ +# HDH Deployment Example šŸš€ + +A comprehensive deployment example showcasing the power of the **HDH (Hybrid Dependency Hypergraph)** library for quantum computation analysis and visualization. + +**Special thanks to [Maria Gragera Garces](https://github.com/grageragarces) for her excellent work on the HDH library! šŸŽ‰** + +## Overview + +This deployment example demonstrates real-world usage of the HDH library, providing: + +- **Comprehensive circuit processing** - Convert quantum circuits to HDH format +- **Performance benchmarking** - Analyze HDH performance across different circuit types +- **Visualization capabilities** - Generate HDH visualizations and analysis plots +- **Scalability testing** - Test HDH with circuits of varying complexity +- **Production-ready deployment** - Docker support, logging, error handling +- **CLI tools** - Command-line interface for easy interaction + +## Features + +### āœØ Core Capabilities + +- **Multi-framework support**: Qiskit, Braket, Cirq, PennyLane circuit conversion +- **QASM file processing**: Import and analyze OpenQASM 2.0 files +- **Advanced analysis**: Circuit partitioning, dependency analysis, metrics computation +- **Rich visualizations**: HDH graphs, performance plots, scalability analysis +- **Benchmarking suite**: Comprehensive performance evaluation tools + +### šŸ”§ Production Features + +- **Error handling**: Robust error handling and recovery +- **Logging**: Comprehensive logging with configurable levels +- **Configuration**: YAML-based configuration management +- **Docker support**: Containerized deployment options +- **CLI interface**: User-friendly command-line tools +- **Performance monitoring**: Memory usage and execution time tracking + +## Installation + +### Prerequisites + +- Python 3.10 or higher +- HDH library (from the parent HDH directory) +- Dependencies listed in `requirements.txt` + +### Quick Setup + +```bash +# Clone and navigate to the examples directory +cd examples + +# Create virtual environment (recommended) +python -m venv hdh-env +source hdh-env/bin/activate # On Windows: hdh-env\Scripts\activate + +# Install dependencies +pip install -r requirements.txt + +# Install HDH library in development mode +pip install -e ../HDH + +# Verify installation +python main.py --help +``` + +### Development Setup + +```bash +# Install with development dependencies +pip install -e ".[dev]" + +# Run tests +pytest + +# Format code +black . +isort . +``` + +## Usage + +### šŸš€ Quick Start + +Run the basic deployment example: + +```bash +python main.py +``` + +This will process several example quantum circuits and generate HDH visualizations. + +### šŸ“Š Comprehensive Demo + +Run the full demonstration suite: + +```bash +python main.py --demo-mode --output-dir results +``` + +### šŸ“ˆ Performance Benchmarking + +Run comprehensive performance benchmarks: + +```bash +python benchmark.py --suite all --repetitions 5 +``` + +Benchmark specific circuit types: + +```bash +# Test scalability +python benchmark.py --suite scalability --max-qubits 8 + +# Test algorithms +python benchmark.py --suite algorithms + +# Test random circuits +python benchmark.py --suite random +``` + +### šŸ” Process Specific Files + +Process a specific QASM file: + +```bash +python main.py --qasm-file path/to/circuit.qasm +``` + +### šŸŽ›ļø Command Line Interface + +Use the interactive CLI: + +```bash +python cli.py +``` + +## Examples + +### Basic Circuit Processing + +```python +from main import HDHDeploymentManager +from circuit_examples import HDHCircuitLibrary + +# Initialize deployment manager +manager = HDHDeploymentManager(output_dir="my_results") + +# Create example circuit +library = HDHCircuitLibrary() +bell_circuit = library.bell_state() + +# Process circuit +result = manager.process_circuit(bell_circuit, save_plots=True) +print(f"Processed {result['circuit_name']}: {result['hdh_stats']['nodes']} nodes") +``` + +### Benchmarking Suite + +```python +from benchmark import HDHBenchmarkSuite + +# Initialize benchmark suite +benchmark = HDHBenchmarkSuite(repetitions=3) + +# Run comprehensive benchmarks +report = benchmark.run_full_benchmark() +print(f"Benchmarked {report['benchmark_summary']['total_circuits']} circuits") +``` + +## Circuit Library + +The deployment includes a comprehensive quantum circuit library: + +### Basic Quantum States +- **Bell States**: All four Bell state variants +- **GHZ States**: Multi-qubit entangled states +- **W States**: Symmetric superposition states + +### Quantum Algorithms +- **Quantum Fourier Transform (QFT)**: Efficient Fourier transform implementation +- **Grover's Algorithm**: Quantum search algorithm +- **Deutsch-Jozsa Algorithm**: Quantum function evaluation +- **Shor's Algorithm**: Period finding (simplified) + +### Quantum Protocols +- **Quantum Teleportation**: State transfer protocol +- **Quantum Error Correction**: 3-qubit bit-flip code + +### Variational Algorithms +- **VQE (Variational Quantum Eigensolver)**: Quantum optimization +- **QAOA (Quantum Approximate Optimization Algorithm)**: Combinatorial optimization + +### Random Circuits +- **Parameterized random circuits**: For benchmarking and testing + +## Configuration + +Customize the deployment using `config.yaml`: + +```yaml +# Logging configuration +logging: + level: INFO + file: "hdh_deployment.log" + +# Output settings +output: + directory: "hdh_results" + save_plots: true + plot_dpi: 300 + +# Circuit processing +circuits: + max_qubits: 10 + default_partitions: 3 + enable_visualization: true + +# Performance settings +performance: + timeout_seconds: 300 + max_memory_gb: 8 +``` + +## Docker Deployment + +### Build and Run + +```bash +# Build Docker image +docker build -t hdh-deployment . + +# Run deployment +docker run -v $(pwd)/results:/app/results hdh-deployment + +# Run with custom configuration +docker run -v $(pwd)/config.yaml:/app/config.yaml hdh-deployment +``` + +### Docker Compose + +```bash +# Start complete deployment stack +docker-compose up + +# Run benchmarks +docker-compose run benchmark python benchmark.py --suite all +``` + +## API Reference + +### HDHDeploymentManager + +Main deployment management class: + +```python +manager = HDHDeploymentManager( + output_dir="results", # Output directory + log_level="INFO" # Logging level +) + +# Process quantum circuit +result = manager.process_circuit(quantum_circuit) + +# Process QASM file +result = manager.process_qasm_file("circuit.qasm") + +# Run comprehensive demo +summary = manager.run_comprehensive_demo() +``` + +### HDHBenchmarkSuite + +Performance benchmarking suite: + +```python +benchmark = HDHBenchmarkSuite( + output_dir="benchmark_results", + repetitions=3 +) + +# Run specific benchmarks +results = benchmark.run_scalability_benchmark() +results = benchmark.run_algorithm_benchmark() + +# Generate performance plots +benchmark.generate_performance_plots(results) +``` + +### HDHCircuitLibrary + +Quantum circuit examples library: + +```python +library = HDHCircuitLibrary() + +# Get individual circuits +bell = library.bell_state() +ghz = library.ghz_state(4) +qft = library.qft_circuit(3) + +# Get all examples +examples = library.get_all_examples() + +# Get benchmark suite +benchmark_circuits = library.get_benchmark_suite() +``` + +## Results and Output + +The deployment generates comprehensive results: + +### Directory Structure +``` +hdh_results/ +ā”œā”€ā”€ hdh_deployment.log # Detailed logging +ā”œā”€ā”€ deployment_results.json # Processing results +ā”œā”€ā”€ Bell_State_hdh.png # Circuit visualizations +ā”œā”€ā”€ GHZ-3_hdh.png +ā”œā”€ā”€ QFT-3_hdh.png +ā””ā”€ā”€ ... + +benchmark_results/ +ā”œā”€ā”€ benchmark.log # Benchmark logging +ā”œā”€ā”€ benchmark_report.json # Detailed benchmark data +ā”œā”€ā”€ scaling_performance.png # Performance scaling plots +ā”œā”€ā”€ algorithm_comparison.png # Algorithm comparison +ā”œā”€ā”€ memory_analysis.png # Memory usage analysis +ā””ā”€ā”€ performance_complexity.png # Complexity analysis +``` + +### Performance Metrics + +The deployment tracks comprehensive performance metrics: + +- **Conversion time**: Time to convert circuits to HDH +- **Memory usage**: Peak memory consumption +- **HDH statistics**: Nodes, edges, timesteps +- **Partitioning metrics**: Cut cost, parallelism analysis +- **Scalability data**: Performance vs circuit size + +## Contributing + +We welcome contributions! Here's how you can help: + +1. **Report Issues**: Found a bug? Report it! +2. **Add Examples**: Contribute new quantum circuit examples +3. **Improve Performance**: Optimize benchmarking and analysis +4. **Documentation**: Help improve documentation +5. **Testing**: Add more comprehensive tests + +### Development Guidelines + +```bash +# Run tests +pytest tests/ + +# Format code +black . +isort . +flake8 . + +# Type checking +mypy . +``` + +## Troubleshooting + +### Common Issues + +**Import Error**: "No module named 'hdh'" +```bash +# Ensure HDH is installed +pip install -e ../HDH +``` + +**Memory Issues**: Large circuits consuming too much memory +```bash +# Reduce circuit size or use configuration limits +python main.py --max-qubits 6 +``` + +**Visualization Errors**: Matplotlib backend issues +```bash +# Set backend explicitly +export MPLBACKEND=Agg +``` + +### Performance Tips + +1. **Limit circuit size**: Start with smaller circuits +2. **Use configuration**: Customize limits in config.yaml +3. **Monitor memory**: Use the memory profiling features +4. **Batch processing**: Process circuits in batches for large datasets + +## Acknowledgments + +This deployment example was created to showcase the capabilities of the HDH library. + +**Special recognition goes to [Maria Gragera Garces](https://github.com/grageragarces) for her outstanding work developing the HDH library. Her innovative approach to quantum computation analysis through hybrid dependency hypergraphs has made this comprehensive deployment example possible.** šŸ™ + +### References + +- **HDH Library**: [GitHub Repository](https://github.com/grageragarces/hdh) +- **Documentation**: [HDH Documentation](https://grageragarces.github.io/HDH/) +- **PyPI Package**: [hdh](https://pypi.org/project/hdh/) + +## License + +This deployment example is provided under the MIT License, consistent with the HDH library. + +## Support + +For questions and support: + +- **HDH Library Issues**: [GitHub Issues](https://github.com/grageragarces/hdh/issues) +- **Deployment Example**: Create an issue in this repository +- **General Questions**: Check the HDH documentation + +--- + +*Built with ā¤ļø for the quantum computing community* + +*Thank you Maria for making quantum computation analysis more accessible through HDH! šŸŒŸ* \ No newline at end of file diff --git a/benchmark.py b/benchmark.py new file mode 100644 index 0000000..839880a --- /dev/null +++ b/benchmark.py @@ -0,0 +1,729 @@ +#!/usr/bin/env python3 +""" +HDH Performance Benchmarking Suite +=================================== + +Comprehensive benchmarking and performance analysis for the HDH library. +This script evaluates HDH performance across different circuit types, sizes, +and complexity levels. + +Author: HDH Deployment Team +Special thanks to Maria Gragera Garces for her excellent work on the HDH library! + +Features: +- Circuit conversion performance benchmarking +- Memory usage analysis +- Scalability testing +- Partitioning algorithm evaluation +- Comparative analysis across circuit types +- Statistical analysis and reporting +""" + +import os +import sys +import time +import json +import logging +import argparse +import traceback +from pathlib import Path +from typing import Dict, List, Tuple, Any, Optional +from datetime import datetime +from dataclasses import dataclass, asdict +from statistics import mean, median, stdev +import matplotlib.pyplot as plt +import numpy as np + +# Memory and performance monitoring +import psutil +import gc +from memory_profiler import profile + +# Add HDH to path +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'HDH'))) + +# HDH imports +from hdh import HDH, plot_hdh +from hdh.converters.qiskit import from_qiskit +from hdh.converters.qasm import from_qasm +from hdh.passes.cut import compute_cut, cost, partition_sizes, compute_parallelism_by_time + +# Circuit examples +from circuit_examples import HDHCircuitLibrary + + +@dataclass +class BenchmarkResult: + """Data structure for storing benchmark results.""" + circuit_name: str + num_qubits: int + circuit_depth: int + circuit_size: int + conversion_time: float + memory_peak_mb: float + hdh_nodes: int + hdh_edges: int + hdh_timesteps: int + partitioning_time: float + partition_cost: float + visualization_time: Optional[float] = None + error_message: Optional[str] = None + success: bool = True + + +class HDHBenchmarkSuite: + """ + Comprehensive benchmarking suite for HDH performance evaluation. + """ + + def __init__(self, output_dir: str = "benchmark_results", repetitions: int = 3): + """Initialize the benchmark suite.""" + self.output_dir = Path(output_dir) + self.output_dir.mkdir(exist_ok=True) + self.repetitions = repetitions + + # Setup logging + self.setup_logging() + self.logger = logging.getLogger(__name__) + + # Results storage + self.results: List[BenchmarkResult] = [] + + # Circuit library + self.circuit_library = HDHCircuitLibrary() + + self.logger.info("HDH Benchmark Suite initialized") + + def setup_logging(self): + """Configure logging for benchmarking.""" + log_file = self.output_dir / "benchmark.log" + + logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler(log_file), + logging.StreamHandler(sys.stdout) + ] + ) + + def get_memory_usage(self) -> float: + """Get current memory usage in MB.""" + process = psutil.Process() + return process.memory_info().rss / 1024 / 1024 + + def benchmark_circuit_conversion(self, circuit, circuit_name: str) -> BenchmarkResult: + """Benchmark a single circuit conversion to HDH.""" + self.logger.info(f"Benchmarking circuit: {circuit_name}") + + # Initial memory measurement + gc.collect() # Force garbage collection + initial_memory = self.get_memory_usage() + + try: + # Time the conversion + start_time = time.perf_counter() + hdh = from_qiskit(circuit) + conversion_time = time.perf_counter() - start_time + + # Memory peak measurement + peak_memory = self.get_memory_usage() + memory_used = peak_memory - initial_memory + + # HDH statistics + hdh_nodes = len(hdh.S) + hdh_edges = len(hdh.C) + hdh_timesteps = len(hdh.T) + + # Partitioning benchmark (if applicable) + partitioning_time = 0 + partition_cost = 0 + + if hdh_nodes > 1: + try: + num_parts = min(3, max(2, hdh_nodes // 2)) + start_partition = time.perf_counter() + partitions = compute_cut(hdh, num_parts) + partitioning_time = time.perf_counter() - start_partition + partition_cost = cost(hdh, partitions) + except Exception as e: + self.logger.warning(f"Partitioning failed for {circuit_name}: {str(e)}") + + # Visualization benchmark (optional) + visualization_time = None + try: + start_vis = time.perf_counter() + # Don't actually save, just measure rendering time + plot_hdh(hdh, save_path=None) + plt.close('all') # Clean up + visualization_time = time.perf_counter() - start_vis + except Exception as e: + self.logger.warning(f"Visualization failed for {circuit_name}: {str(e)}") + + return BenchmarkResult( + circuit_name=circuit_name, + num_qubits=circuit.num_qubits, + circuit_depth=circuit.depth(), + circuit_size=circuit.size(), + conversion_time=conversion_time, + memory_peak_mb=memory_used, + hdh_nodes=hdh_nodes, + hdh_edges=hdh_edges, + hdh_timesteps=hdh_timesteps, + partitioning_time=partitioning_time, + partition_cost=partition_cost, + visualization_time=visualization_time, + success=True + ) + + except Exception as e: + self.logger.error(f"Benchmark failed for {circuit_name}: {str(e)}") + self.logger.debug(traceback.format_exc()) + + return BenchmarkResult( + circuit_name=circuit_name, + num_qubits=circuit.num_qubits, + circuit_depth=circuit.depth(), + circuit_size=circuit.size(), + conversion_time=0, + memory_peak_mb=0, + hdh_nodes=0, + hdh_edges=0, + hdh_timesteps=0, + partitioning_time=0, + partition_cost=0, + error_message=str(e), + success=False + ) + + def run_scalability_benchmark(self) -> List[BenchmarkResult]: + """Run scalability benchmarks with varying circuit sizes.""" + self.logger.info("Running scalability benchmark") + + results = [] + + # Test different qubit counts + qubit_counts = [2, 3, 4, 5, 6, 7, 8] + + for n_qubits in qubit_counts: + # Test different circuit types + test_circuits = [ + (self.circuit_library.ghz_state(n_qubits), f"GHZ-{n_qubits}"), + (self.circuit_library.qft_circuit(min(n_qubits, 6)), f"QFT-{min(n_qubits, 6)}"), # Limit QFT size + (self.circuit_library.random_circuit(n_qubits, n_qubits * 2, seed=42), f"Random-{n_qubits}") + ] + + for circuit, name in test_circuits: + if circuit.num_qubits <= 8: # Safety limit + # Run multiple repetitions + repetition_results = [] + for rep in range(self.repetitions): + result = self.benchmark_circuit_conversion(circuit, f"{name}-rep{rep}") + repetition_results.append(result) + + # Average the repetitions + if repetition_results and any(r.success for r in repetition_results): + successful_results = [r for r in repetition_results if r.success] + if successful_results: + avg_result = self.average_results(successful_results, name) + results.append(avg_result) + + return results + + def run_algorithm_benchmark(self) -> List[BenchmarkResult]: + """Benchmark specific quantum algorithms.""" + self.logger.info("Running algorithm benchmark") + + results = [] + + # Algorithm test suite + algorithms = [ + (self.circuit_library.bell_state(), "Bell State"), + (self.circuit_library.ghz_state(4), "GHZ-4"), + (self.circuit_library.w_state(4), "W-4"), + (self.circuit_library.qft_circuit(4), "QFT-4"), + (self.circuit_library.grover_search(3), "Grover-3"), + (self.circuit_library.deutsch_jozsa(4), "Deutsch-Jozsa"), + (self.circuit_library.quantum_teleportation(), "Teleportation"), + (self.circuit_library.vqe_ansatz(4, 2), "VQE"), + (self.circuit_library.quantum_error_correction_3bit(), "QEC-3bit") + ] + + for circuit, name in algorithms: + repetition_results = [] + for rep in range(self.repetitions): + result = self.benchmark_circuit_conversion(circuit, f"{name}-rep{rep}") + repetition_results.append(result) + + # Average the repetitions + if repetition_results and any(r.success for r in repetition_results): + successful_results = [r for r in repetition_results if r.success] + if successful_results: + avg_result = self.average_results(successful_results, name) + results.append(avg_result) + + return results + + def run_random_circuit_benchmark(self, max_qubits: int = 6, max_depth: int = 20) -> List[BenchmarkResult]: + """Benchmark random circuits of varying complexity.""" + self.logger.info("Running random circuit benchmark") + + results = [] + + # Generate random circuits with different parameters + test_configs = [ + (3, 5), (3, 10), (4, 5), (4, 10), (5, 8), (6, 6) + ] + + seeds = [42, 123, 456] # Multiple seeds for variety + + for n_qubits, depth in test_configs: + if n_qubits <= max_qubits and depth <= max_depth: + for seed in seeds: + circuit = self.circuit_library.random_circuit(n_qubits, depth, seed) + name = f"Random-{n_qubits}q-{depth}d-s{seed}" + + result = self.benchmark_circuit_conversion(circuit, name) + if result.success: + results.append(result) + + return results + + def average_results(self, results: List[BenchmarkResult], name: str) -> BenchmarkResult: + """Average multiple benchmark results.""" + if not results: + raise ValueError("No results to average") + + # Take the first result as template + template = results[0] + + # Average numerical fields + avg_conversion_time = mean([r.conversion_time for r in results]) + avg_memory = mean([r.memory_peak_mb for r in results]) + avg_partitioning_time = mean([r.partitioning_time for r in results]) + avg_partition_cost = mean([r.partition_cost for r in results]) + + avg_vis_time = None + vis_times = [r.visualization_time for r in results if r.visualization_time is not None] + if vis_times: + avg_vis_time = mean(vis_times) + + return BenchmarkResult( + circuit_name=name, + num_qubits=template.num_qubits, + circuit_depth=template.circuit_depth, + circuit_size=template.circuit_size, + conversion_time=avg_conversion_time, + memory_peak_mb=avg_memory, + hdh_nodes=template.hdh_nodes, + hdh_edges=template.hdh_edges, + hdh_timesteps=template.hdh_timesteps, + partitioning_time=avg_partitioning_time, + partition_cost=avg_partition_cost, + visualization_time=avg_vis_time, + success=True + ) + + def run_comprehensive_benchmark(self) -> Dict[str, List[BenchmarkResult]]: + """Run all benchmark suites.""" + self.logger.info("Starting comprehensive HDH benchmark") + + benchmark_results = { + 'scalability': [], + 'algorithms': [], + 'random_circuits': [] + } + + # Run individual benchmark suites + try: + benchmark_results['scalability'] = self.run_scalability_benchmark() + except Exception as e: + self.logger.error(f"Scalability benchmark failed: {str(e)}") + + try: + benchmark_results['algorithms'] = self.run_algorithm_benchmark() + except Exception as e: + self.logger.error(f"Algorithm benchmark failed: {str(e)}") + + try: + benchmark_results['random_circuits'] = self.run_random_circuit_benchmark() + except Exception as e: + self.logger.error(f"Random circuit benchmark failed: {str(e)}") + + # Store all results + for suite_results in benchmark_results.values(): + self.results.extend(suite_results) + + return benchmark_results + + def generate_performance_plots(self, results: Dict[str, List[BenchmarkResult]]): + """Generate performance analysis plots.""" + self.logger.info("Generating performance plots") + + # Scaling plot + self.plot_scaling_performance(results['scalability']) + + # Algorithm comparison + self.plot_algorithm_comparison(results['algorithms']) + + # Memory usage analysis + all_results = [] + for suite_results in results.values(): + all_results.extend(suite_results) + self.plot_memory_analysis(all_results) + + # Performance vs complexity + self.plot_performance_vs_complexity(all_results) + + def plot_scaling_performance(self, results: List[BenchmarkResult]): + """Plot performance scaling with circuit size.""" + if not results: + return + + fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 12)) + + # Group by circuit type + circuit_types = {} + for result in results: + circuit_type = result.circuit_name.split('-')[0] + if circuit_type not in circuit_types: + circuit_types[circuit_type] = [] + circuit_types[circuit_type].append(result) + + # Plot 1: Conversion time vs qubits + for circuit_type, type_results in circuit_types.items(): + qubits = [r.num_qubits for r in type_results] + times = [r.conversion_time for r in type_results] + ax1.plot(qubits, times, 'o-', label=circuit_type, alpha=0.7) + + ax1.set_xlabel('Number of Qubits') + ax1.set_ylabel('Conversion Time (s)') + ax1.set_title('HDH Conversion Time Scaling') + ax1.legend() + ax1.grid(True, alpha=0.3) + ax1.set_yscale('log') + + # Plot 2: Memory usage vs qubits + for circuit_type, type_results in circuit_types.items(): + qubits = [r.num_qubits for r in type_results] + memory = [r.memory_peak_mb for r in type_results] + ax2.plot(qubits, memory, 's-', label=circuit_type, alpha=0.7) + + ax2.set_xlabel('Number of Qubits') + ax2.set_ylabel('Peak Memory Usage (MB)') + ax2.set_title('Memory Usage Scaling') + ax2.legend() + ax2.grid(True, alpha=0.3) + + # Plot 3: HDH nodes vs circuit size + all_sizes = [r.circuit_size for r in results] + all_nodes = [r.hdh_nodes for r in results] + ax3.scatter(all_sizes, all_nodes, alpha=0.6) + ax3.set_xlabel('Circuit Size (gates)') + ax3.set_ylabel('HDH Nodes') + ax3.set_title('HDH Representation Size') + ax3.grid(True, alpha=0.3) + + # Plot 4: Partitioning cost distribution + costs = [r.partition_cost for r in results if r.partition_cost > 0] + if costs: + ax4.hist(costs, bins=20, alpha=0.7, color='skyblue', edgecolor='black') + ax4.set_xlabel('Partition Cost') + ax4.set_ylabel('Frequency') + ax4.set_title('Partitioning Cost Distribution') + ax4.grid(True, alpha=0.3) + + plt.tight_layout() + plt.savefig(self.output_dir / 'scaling_performance.png', dpi=300, bbox_inches='tight') + plt.close() + + def plot_algorithm_comparison(self, results: List[BenchmarkResult]): + """Plot algorithm-specific performance comparison.""" + if not results: + return + + fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6)) + + names = [r.circuit_name for r in results] + conversion_times = [r.conversion_time for r in results] + memory_usage = [r.memory_peak_mb for r in results] + + # Conversion time comparison + bars1 = ax1.bar(range(len(names)), conversion_times, alpha=0.7, color='lightcoral') + ax1.set_xlabel('Algorithm') + ax1.set_ylabel('Conversion Time (s)') + ax1.set_title('HDH Conversion Time by Algorithm') + ax1.set_xticks(range(len(names))) + ax1.set_xticklabels(names, rotation=45, ha='right') + ax1.grid(True, alpha=0.3, axis='y') + + # Add value labels on bars + for bar, time in zip(bars1, conversion_times): + height = bar.get_height() + ax1.text(bar.get_x() + bar.get_width()/2., height, + f'{time:.3f}s', ha='center', va='bottom', fontsize=8) + + # Memory usage comparison + bars2 = ax2.bar(range(len(names)), memory_usage, alpha=0.7, color='lightblue') + ax2.set_xlabel('Algorithm') + ax2.set_ylabel('Peak Memory (MB)') + ax2.set_title('Memory Usage by Algorithm') + ax2.set_xticks(range(len(names))) + ax2.set_xticklabels(names, rotation=45, ha='right') + ax2.grid(True, alpha=0.3, axis='y') + + # Add value labels on bars + for bar, mem in zip(bars2, memory_usage): + height = bar.get_height() + ax2.text(bar.get_x() + bar.get_width()/2., height, + f'{mem:.1f}MB', ha='center', va='bottom', fontsize=8) + + plt.tight_layout() + plt.savefig(self.output_dir / 'algorithm_comparison.png', dpi=300, bbox_inches='tight') + plt.close() + + def plot_memory_analysis(self, results: List[BenchmarkResult]): + """Plot memory usage analysis.""" + if not results: + return + + fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6)) + + # Memory vs HDH size + hdh_sizes = [r.hdh_nodes + r.hdh_edges for r in results] + memory_usage = [r.memory_peak_mb for r in results] + + ax1.scatter(hdh_sizes, memory_usage, alpha=0.6, color='green') + ax1.set_xlabel('HDH Size (nodes + edges)') + ax1.set_ylabel('Peak Memory (MB)') + ax1.set_title('Memory Usage vs HDH Size') + ax1.grid(True, alpha=0.3) + + # Memory efficiency (memory per HDH element) + efficiency = [mem / max(size, 1) for mem, size in zip(memory_usage, hdh_sizes)] + + ax2.hist(efficiency, bins=20, alpha=0.7, color='orange', edgecolor='black') + ax2.set_xlabel('Memory per HDH Element (MB)') + ax2.set_ylabel('Frequency') + ax2.set_title('Memory Efficiency Distribution') + ax2.grid(True, alpha=0.3) + + plt.tight_layout() + plt.savefig(self.output_dir / 'memory_analysis.png', dpi=300, bbox_inches='tight') + plt.close() + + def plot_performance_vs_complexity(self, results: List[BenchmarkResult]): + """Plot performance vs circuit complexity.""" + if not results: + return + + fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 12)) + + # Performance vs circuit depth + depths = [r.circuit_depth for r in results] + times = [r.conversion_time for r in results] + + ax1.scatter(depths, times, alpha=0.6) + ax1.set_xlabel('Circuit Depth') + ax1.set_ylabel('Conversion Time (s)') + ax1.set_title('Performance vs Circuit Depth') + ax1.grid(True, alpha=0.3) + ax1.set_yscale('log') + + # Performance vs circuit size + sizes = [r.circuit_size for r in results] + + ax2.scatter(sizes, times, alpha=0.6, color='red') + ax2.set_xlabel('Circuit Size (gates)') + ax2.set_ylabel('Conversion Time (s)') + ax2.set_title('Performance vs Circuit Size') + ax2.grid(True, alpha=0.3) + ax2.set_yscale('log') + + # HDH efficiency + hdh_efficiency = [r.hdh_nodes / max(r.circuit_size, 1) for r in results] + + ax3.scatter(sizes, hdh_efficiency, alpha=0.6, color='purple') + ax3.set_xlabel('Circuit Size (gates)') + ax3.set_ylabel('HDH Nodes / Circuit Gates') + ax3.set_title('HDH Representation Efficiency') + ax3.grid(True, alpha=0.3) + + # Partitioning efficiency + partition_efficiency = [r.partitioning_time / max(r.conversion_time, 0.001) for r in results if r.partitioning_time > 0] + hdh_sizes_with_partition = [r.hdh_nodes for r in results if r.partitioning_time > 0] + + if partition_efficiency: + ax4.scatter(hdh_sizes_with_partition, partition_efficiency, alpha=0.6, color='brown') + ax4.set_xlabel('HDH Nodes') + ax4.set_ylabel('Partitioning Time / Conversion Time') + ax4.set_title('Partitioning Overhead') + ax4.grid(True, alpha=0.3) + ax4.set_yscale('log') + + plt.tight_layout() + plt.savefig(self.output_dir / 'performance_complexity.png', dpi=300, bbox_inches='tight') + plt.close() + + def generate_report(self, results: Dict[str, List[BenchmarkResult]]) -> Dict[str, Any]: + """Generate comprehensive benchmark report.""" + self.logger.info("Generating benchmark report") + + all_results = [] + for suite_results in results.values(): + all_results.extend([r for r in suite_results if r.success]) + + if not all_results: + return {"error": "No successful benchmark results"} + + # Basic statistics + conversion_times = [r.conversion_time for r in all_results] + memory_usage = [r.memory_peak_mb for r in all_results] + hdh_nodes = [r.hdh_nodes for r in all_results] + + report = { + "benchmark_summary": { + "total_circuits": len(all_results), + "successful_circuits": len(all_results), + "benchmark_date": datetime.now().isoformat(), + "repetitions": self.repetitions + }, + "performance_statistics": { + "conversion_time": { + "mean": mean(conversion_times), + "median": median(conversion_times), + "min": min(conversion_times), + "max": max(conversion_times), + "std": stdev(conversion_times) if len(conversion_times) > 1 else 0 + }, + "memory_usage": { + "mean": mean(memory_usage), + "median": median(memory_usage), + "min": min(memory_usage), + "max": max(memory_usage), + "std": stdev(memory_usage) if len(memory_usage) > 1 else 0 + }, + "hdh_size": { + "mean_nodes": mean(hdh_nodes), + "median_nodes": median(hdh_nodes), + "min_nodes": min(hdh_nodes), + "max_nodes": max(hdh_nodes) + } + }, + "scalability_analysis": { + "largest_circuit_qubits": max([r.num_qubits for r in all_results]), + "largest_circuit_size": max([r.circuit_size for r in all_results]), + "largest_hdh_nodes": max(hdh_nodes) + }, + "suite_results": { + suite_name: { + "count": len(suite_results), + "avg_conversion_time": mean([r.conversion_time for r in suite_results]) if suite_results else 0, + "avg_memory": mean([r.memory_peak_mb for r in suite_results]) if suite_results else 0 + } + for suite_name, suite_results in results.items() if suite_results + } + } + + # Save detailed results + detailed_results = { + "report": report, + "detailed_results": [asdict(r) for r in all_results] + } + + report_file = self.output_dir / "benchmark_report.json" + with open(report_file, 'w') as f: + json.dump(detailed_results, f, indent=2, default=str) + + return report + + def run_full_benchmark(self) -> Dict[str, Any]: + """Run complete benchmark suite and generate report.""" + start_time = time.time() + + # Run benchmarks + results = self.run_comprehensive_benchmark() + + # Generate visualizations + self.generate_performance_plots(results) + + # Generate report + report = self.generate_report(results) + + total_time = time.time() - start_time + report["benchmark_summary"]["total_benchmark_time"] = total_time + + self.logger.info(f"Benchmark completed in {total_time:.2f} seconds") + return report + + +def main(): + """Main benchmarking function.""" + parser = argparse.ArgumentParser(description="HDH Performance Benchmark Suite") + parser.add_argument("--output-dir", default="benchmark_results", help="Output directory") + parser.add_argument("--repetitions", type=int, default=3, help="Number of repetitions per test") + parser.add_argument("--suite", choices=["scalability", "algorithms", "random", "all"], + default="all", help="Benchmark suite to run") + parser.add_argument("--max-qubits", type=int, default=8, help="Maximum qubits for scaling tests") + parser.add_argument("--verbose", action="store_true", help="Verbose logging") + + args = parser.parse_args() + + # Initialize benchmark suite + benchmark = HDHBenchmarkSuite( + output_dir=args.output_dir, + repetitions=args.repetitions + ) + + if args.verbose: + logging.getLogger().setLevel(logging.DEBUG) + + try: + print("šŸš€ HDH Performance Benchmark Suite") + print("=" * 50) + print("Special thanks to Maria Gragera Garces for the HDH library!") + print() + + if args.suite == "all": + report = benchmark.run_full_benchmark() + else: + # Run specific benchmark suite + if args.suite == "scalability": + results = {"scalability": benchmark.run_scalability_benchmark()} + elif args.suite == "algorithms": + results = {"algorithms": benchmark.run_algorithm_benchmark()} + elif args.suite == "random": + results = {"random": benchmark.run_random_circuit_benchmark(args.max_qubits)} + + benchmark.generate_performance_plots(results) + report = benchmark.generate_report(results) + + # Print summary + if "error" not in report: + summary = report["benchmark_summary"] + perf_stats = report["performance_statistics"] + + print(f"\nšŸ“Š Benchmark Results Summary:") + print(f"Circuits tested: {summary['total_circuits']}") + print(f"Success rate: 100%") + print(f"Average conversion time: {perf_stats['conversion_time']['mean']:.4f}s") + print(f"Average memory usage: {perf_stats['memory_usage']['mean']:.2f}MB") + print(f"Largest circuit: {report['scalability_analysis']['largest_circuit_qubits']} qubits") + print(f"Total benchmark time: {summary.get('total_benchmark_time', 0):.2f}s") + + print(f"\nšŸ“ Results saved in: {benchmark.output_dir}") + print("šŸ“ˆ Performance plots generated") + print("šŸ“‹ Detailed report: benchmark_report.json") + else: + print(f"\nāŒ Benchmark failed: {report['error']}") + + except KeyboardInterrupt: + print("\nā¹ļø Benchmark interrupted by user") + except Exception as e: + print(f"\nšŸ’„ Benchmark failed: {str(e)}") + logging.error(traceback.format_exc()) + raise + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/circuit_examples.py b/circuit_examples.py new file mode 100644 index 0000000..833a0f8 --- /dev/null +++ b/circuit_examples.py @@ -0,0 +1,455 @@ +#!/usr/bin/env python3 +""" +HDH Circuit Examples Library +============================ + +This module provides a comprehensive collection of quantum circuits for testing +and demonstrating the HDH (Hybrid Dependency Hypergraph) library capabilities. + +Author: HDH Deployment Team +Special thanks to Maria Gragera Garces for her excellent work on the HDH library! + +The examples include: +- Basic quantum circuits (Bell states, GHZ states) +- Quantum algorithms (QFT, Grover, Deutsch-Jozsa) +- Quantum error correction codes +- Random circuits for benchmarking +- Real-world quantum applications +""" + +import os +import sys +from typing import List, Dict, Tuple, Optional +from pathlib import Path +import numpy as np + +# Add HDH to path +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'HDH'))) + +from qiskit import QuantumCircuit, QuantumRegister, ClassicalRegister +from qiskit.circuit.library import ( + QFT, GroverOperator, DeutschJozsaOracle, + TwoLocal, RealAmplitudes, EfficientSU2 +) +from qiskit.circuit import Parameter +import qiskit.circuit.library as qlib + + +class HDHCircuitLibrary: + """ + A comprehensive library of quantum circuits for HDH testing and demonstration. + """ + + @staticmethod + def bell_state() -> QuantumCircuit: + """Create a Bell state (|00āŸ© + |11āŸ©)/āˆš2.""" + qc = QuantumCircuit(2, 2, name="Bell State") + qc.h(0) + qc.cx(0, 1) + qc.measure_all() + return qc + + @staticmethod + def bell_state_variants() -> List[QuantumCircuit]: + """Create all four Bell states.""" + circuits = [] + + # |Ī¦+āŸ© = (|00āŸ© + |11āŸ©)/āˆš2 + qc1 = QuantumCircuit(2, 2, name="Bell Phi+") + qc1.h(0) + qc1.cx(0, 1) + qc1.measure_all() + circuits.append(qc1) + + # |Ī¦-āŸ© = (|00āŸ© - |11āŸ©)/āˆš2 + qc2 = QuantumCircuit(2, 2, name="Bell Phi-") + qc2.h(0) + qc2.z(0) + qc2.cx(0, 1) + qc2.measure_all() + circuits.append(qc2) + + # |ĪØ+āŸ© = (|01āŸ© + |10āŸ©)/āˆš2 + qc3 = QuantumCircuit(2, 2, name="Bell Psi+") + qc3.h(0) + qc3.cx(0, 1) + qc3.x(1) + qc3.measure_all() + circuits.append(qc3) + + # |ĪØ-āŸ© = (|01āŸ© - |10āŸ©)/āˆš2 + qc4 = QuantumCircuit(2, 2, name="Bell Psi-") + qc4.h(0) + qc4.z(0) + qc4.cx(0, 1) + qc4.x(1) + qc4.measure_all() + circuits.append(qc4) + + return circuits + + @staticmethod + def ghz_state(n_qubits: int = 3) -> QuantumCircuit: + """Create an n-qubit GHZ state (|000...āŸ© + |111...āŸ©)/āˆš2.""" + qc = QuantumCircuit(n_qubits, n_qubits, name=f"GHZ-{n_qubits}") + qc.h(0) + for i in range(1, n_qubits): + qc.cx(0, i) + qc.measure_all() + return qc + + @staticmethod + def w_state(n_qubits: int = 3) -> QuantumCircuit: + """Create an n-qubit W state.""" + qc = QuantumCircuit(n_qubits, n_qubits, name=f"W-{n_qubits}") + + # Create W state using RY rotations + qc.ry(2 * np.arccos(np.sqrt((n_qubits - 1) / n_qubits)), 0) + + for i in range(1, n_qubits): + angle = 2 * np.arccos(np.sqrt((n_qubits - i - 1) / (n_qubits - i))) + qc.cry(angle, i-1, i) + + qc.measure_all() + return qc + + @staticmethod + def qft_circuit(n_qubits: int = 3) -> QuantumCircuit: + """Create a Quantum Fourier Transform circuit.""" + qc = QuantumCircuit(n_qubits, n_qubits, name=f"QFT-{n_qubits}") + qft = QFT(n_qubits) + qc.compose(qft, inplace=True) + qc.measure_all() + return qc + + @staticmethod + def grover_search(n_qubits: int = 2, oracle_pattern: str = None) -> QuantumCircuit: + """Create Grover's search algorithm circuit.""" + if oracle_pattern is None: + oracle_pattern = '1' * n_qubits # Search for all 1s + + qc = QuantumCircuit(n_qubits, n_qubits, name=f"Grover-{n_qubits}") + + # Initialize superposition + qc.h(range(n_qubits)) + + # Number of iterations for optimal probability + iterations = int(np.pi / 4 * np.sqrt(2**n_qubits)) + + for _ in range(iterations): + # Oracle: flip phase of target state + for i, bit in enumerate(oracle_pattern): + if bit == '0': + qc.x(i) + + if n_qubits > 1: + qc.mcrz(np.pi, list(range(n_qubits-1)), n_qubits-1) + else: + qc.rz(np.pi, 0) + + for i, bit in enumerate(oracle_pattern): + if bit == '0': + qc.x(i) + + # Diffusion operator + qc.h(range(n_qubits)) + qc.x(range(n_qubits)) + + if n_qubits > 1: + qc.mcrz(np.pi, list(range(n_qubits-1)), n_qubits-1) + else: + qc.rz(np.pi, 0) + + qc.x(range(n_qubits)) + qc.h(range(n_qubits)) + + qc.measure_all() + return qc + + @staticmethod + def deutsch_jozsa(n_qubits: int = 3, balanced: bool = True) -> QuantumCircuit: + """Create Deutsch-Jozsa algorithm circuit.""" + qc = QuantumCircuit(n_qubits + 1, n_qubits, name=f"DJ-{n_qubits}-{'Balanced' if balanced else 'Constant'}") + + # Initialize ancilla qubit in |1āŸ© + qc.x(n_qubits) + + # Create superposition + qc.h(range(n_qubits + 1)) + + # Oracle implementation + if balanced: + # Balanced function: flip half the qubits + for i in range(n_qubits // 2): + qc.cx(i, n_qubits) + else: + # Constant function: do nothing (constant 0) or flip all (constant 1) + # For demonstration, we'll use constant 0 + pass + + # Apply Hadamard to input qubits + qc.h(range(n_qubits)) + + # Measure input qubits + qc.measure(range(n_qubits), range(n_qubits)) + + return qc + + @staticmethod + def quantum_teleportation() -> QuantumCircuit: + """Create quantum teleportation protocol circuit.""" + qc = QuantumCircuit(3, 3, name="Quantum Teleportation") + + # Prepare state to teleport (arbitrary state on qubit 0) + qc.ry(np.pi/4, 0) # Example state + + # Create Bell pair between qubits 1 and 2 + qc.h(1) + qc.cx(1, 2) + + # Bell measurement on qubits 0 and 1 + qc.cx(0, 1) + qc.h(0) + qc.measure(0, 0) + qc.measure(1, 1) + + # Correction operations + qc.cx(1, 2) + qc.cz(0, 2) + + # Measure final state + qc.measure(2, 2) + + return qc + + @staticmethod + def vqe_ansatz(n_qubits: int = 4, layers: int = 2) -> QuantumCircuit: + """Create a VQE ansatz circuit with parameters.""" + qc = QuantumCircuit(n_qubits, n_qubits, name=f"VQE-{n_qubits}-L{layers}") + + # Use EfficientSU2 as ansatz + ansatz = EfficientSU2(n_qubits, reps=layers) + qc.compose(ansatz, inplace=True) + qc.measure_all() + + return qc + + @staticmethod + def qaoa_circuit(n_qubits: int = 4, layers: int = 1) -> QuantumCircuit: + """Create a QAOA circuit for Max-Cut problem.""" + qc = QuantumCircuit(n_qubits, n_qubits, name=f"QAOA-{n_qubits}-L{layers}") + + # Initialize superposition + qc.h(range(n_qubits)) + + # Parameters + gamma = Parameter('Ī³') + beta = Parameter('Ī²') + + for layer in range(layers): + # Problem Hamiltonian (Max-Cut on all edges) + for i in range(n_qubits): + for j in range(i + 1, n_qubits): + qc.rzz(gamma, i, j) + + # Mixer Hamiltonian + for i in range(n_qubits): + qc.rx(beta, i) + + qc.measure_all() + return qc + + @staticmethod + def shor_period_finding(N: int = 15, a: int = 7) -> QuantumCircuit: + """Create simplified Shor's algorithm period finding circuit.""" + # Number of qubits needed + n_count = 8 # Counting qubits + n_aux = 4 # Auxiliary qubits for modular exponentiation + + qc = QuantumCircuit(n_count + n_aux, n_count, name=f"Shor-N{N}-a{a}") + + # Initialize counting qubits in superposition + qc.h(range(n_count)) + + # Initialize auxiliary register to |1āŸ© + qc.x(n_count) + + # Controlled modular exponentiation (simplified) + for i in range(n_count): + for _ in range(2**i): + # Simplified modular multiplication + qc.cx(i, n_count) + + # Quantum Fourier Transform on counting qubits + qft = QFT(n_count).inverse() + qc.compose(qft, range(n_count), inplace=True) + + # Measure counting qubits + qc.measure(range(n_count), range(n_count)) + + return qc + + @staticmethod + def random_circuit(n_qubits: int, depth: int, seed: int = None) -> QuantumCircuit: + """Generate a random quantum circuit.""" + if seed is not None: + np.random.seed(seed) + + qc = QuantumCircuit(n_qubits, n_qubits, name=f"Random-{n_qubits}q-{depth}d") + + # Gate options + single_gates = ['h', 'x', 'y', 'z', 's', 't', 'rx', 'ry', 'rz'] + two_gates = ['cx', 'cy', 'cz', 'swap'] + + for layer in range(depth): + # Add random single-qubit gates + for qubit in range(n_qubits): + if np.random.random() < 0.7: # 70% chance of gate + gate = np.random.choice(single_gates) + if gate in ['rx', 'ry', 'rz']: + angle = np.random.uniform(0, 2*np.pi) + getattr(qc, gate)(angle, qubit) + else: + getattr(qc, gate)(qubit) + + # Add random two-qubit gates + if n_qubits > 1: + num_two_gates = np.random.randint(0, n_qubits // 2 + 1) + for _ in range(num_two_gates): + gate = np.random.choice(two_gates) + qubits = np.random.choice(n_qubits, 2, replace=False) + getattr(qc, gate)(qubits[0], qubits[1]) + + qc.measure_all() + return qc + + @staticmethod + def quantum_error_correction_3bit() -> QuantumCircuit: + """Create a 3-qubit bit-flip error correction circuit.""" + qc = QuantumCircuit(9, 3, name="QEC-3bit-BitFlip") + + # Encode logical |0āŸ© or |1āŸ© (start with |+āŸ© state) + qc.h(0) + + # Encoding + qc.cx(0, 3) + qc.cx(0, 6) + + # Simulate error (bit flip on qubit 3) + qc.x(3) + + # Syndrome measurement + qc.cx(0, 1) + qc.cx(3, 1) + qc.cx(3, 2) + qc.cx(6, 2) + + # Error correction (simplified) + qc.ccx(1, 2, 3) # Correct bit flip if detected + + # Decoding and measurement + qc.cx(0, 3) + qc.cx(0, 6) + qc.measure([0, 1, 2], [0, 1, 2]) + + return qc + + @classmethod + def get_all_examples(cls) -> Dict[str, QuantumCircuit]: + """Get all example circuits as a dictionary.""" + examples = {} + + # Basic states + examples['bell_state'] = cls.bell_state() + examples['ghz_3'] = cls.ghz_state(3) + examples['ghz_4'] = cls.ghz_state(4) + examples['w_state'] = cls.w_state(3) + + # Algorithms + examples['qft_3'] = cls.qft_circuit(3) + examples['qft_4'] = cls.qft_circuit(4) + examples['grover_2'] = cls.grover_search(2) + examples['deutsch_jozsa'] = cls.deutsch_jozsa(3, balanced=True) + + # Protocols + examples['teleportation'] = cls.quantum_teleportation() + examples['error_correction'] = cls.quantum_error_correction_3bit() + + # Variational algorithms + examples['vqe_ansatz'] = cls.vqe_ansatz(4, 2) + examples['qaoa'] = cls.qaoa_circuit(4, 1) + + # Random circuits + examples['random_small'] = cls.random_circuit(3, 5, seed=42) + examples['random_medium'] = cls.random_circuit(5, 8, seed=42) + + return examples + + @classmethod + def get_benchmark_suite(cls) -> List[QuantumCircuit]: + """Get a comprehensive benchmark suite.""" + circuits = [] + + # Scalability test circuits + for n in [2, 3, 4, 5]: + circuits.append(cls.ghz_state(n)) + circuits.append(cls.qft_circuit(n)) + + for n in [2, 3]: + circuits.append(cls.grover_search(n)) + + # Algorithm demonstrations + circuits.append(cls.deutsch_jozsa(4, balanced=True)) + circuits.append(cls.quantum_teleportation()) + circuits.append(cls.vqe_ansatz(4, 2)) + circuits.append(cls.quantum_error_correction_3bit()) + + # Random circuits for stress testing + for n_qubits in [3, 5, 7]: + for depth in [5, 10]: + circuits.append(cls.random_circuit(n_qubits, depth, seed=42)) + + return circuits + + +def save_example_qasm_files(output_dir: str = "qasm_examples"): + """Save example circuits as QASM files.""" + output_path = Path(output_dir) + output_path.mkdir(exist_ok=True) + + library = HDHCircuitLibrary() + examples = library.get_all_examples() + + for name, circuit in examples.items(): + # Remove measurements for QASM export (QASM 2.0 limitation) + qasm_circuit = circuit.copy() + qasm_circuit.remove_final_measurements(inplace=True) + + qasm_file = output_path / f"{name}.qasm" + with open(qasm_file, 'w') as f: + f.write(qasm_circuit.qasm()) + + print(f"Saved {name} to {qasm_file}") + + +if __name__ == "__main__": + """Demonstration of the circuit library.""" + print("HDH Circuit Examples Library") + print("=" * 50) + print("Special thanks to Maria Gragera Garces for the HDH library!") + print() + + library = HDHCircuitLibrary() + + # Show all available examples + examples = library.get_all_examples() + print(f"Available circuits ({len(examples)}):") + for name, qc in examples.items(): + print(f" {name:20} - {qc.num_qubits} qubits, depth {qc.depth()}") + + print(f"\nBenchmark suite: {len(library.get_benchmark_suite())} circuits") + + # Save QASM examples + print("\nSaving QASM examples...") + save_example_qasm_files() + print("Done!") \ No newline at end of file diff --git a/cli.py b/cli.py new file mode 100644 index 0000000..f59d867 --- /dev/null +++ b/cli.py @@ -0,0 +1,843 @@ +#!/usr/bin/env python3 +""" +HDH Command Line Interface +========================== + +Interactive command-line interface for the HDH deployment example. +Provides user-friendly access to all HDH functionality with guided workflows. + +Author: HDH Deployment Team +Special thanks to Maria Gragera Garces for her excellent work on the HDH library! + +Features: +- Interactive menu system +- Circuit processing workflows +- Benchmarking tools +- Configuration management +- Results visualization +- Help and documentation +""" + +import os +import sys +import json +import time +import click +from pathlib import Path +from typing import Dict, List, Optional, Any +from datetime import datetime + +# Rich console for beautiful CLI output +from rich.console import Console +from rich.panel import Panel +from rich.table import Table +from rich.progress import Progress, SpinnerColumn, TextColumn +from rich.prompt import Prompt, Confirm +from rich.syntax import Syntax +from rich.tree import Tree +from rich.text import Text + +# Add HDH to path +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'HDH'))) + +# Import our modules +from main import HDHDeploymentManager +from benchmark import HDHBenchmarkSuite +from circuit_examples import HDHCircuitLibrary + +# Console for rich output +console = Console() + + +class HDHCommandLineInterface: + """ + Interactive command-line interface for HDH deployment. + """ + + def __init__(self): + """Initialize the CLI.""" + self.deployment_manager = None + self.benchmark_suite = None + self.circuit_library = HDHCircuitLibrary() + self.output_dir = "hdh_results" + + # Display welcome message + self.show_welcome() + + def show_welcome(self): + """Display welcome message.""" + welcome_text = """ + [bold blue]HDH (Hybrid Dependency Hypergraph) CLI[/bold blue] + + Interactive interface for quantum circuit analysis and HDH deployment. + + [italic]Special thanks to Maria Gragera Garces for the HDH library![/italic] šŸŽ‰ + """ + + console.print(Panel(welcome_text.strip(), expand=False, border_style="blue")) + + def show_main_menu(self) -> str: + """Display main menu and get user choice.""" + console.print("\n[bold green]Main Menu[/bold green]") + console.print("ā”€" * 50) + + choices = { + "1": "Process Quantum Circuits", + "2": "Run Performance Benchmarks", + "3": "Circuit Library & Examples", + "4": "Configuration & Settings", + "5": "View Results & Reports", + "6": "Help & Documentation", + "q": "Quit" + } + + for key, description in choices.items(): + if key == "q": + console.print(f"[red]{key}[/red]. {description}") + else: + console.print(f"[cyan]{key}[/cyan]. {description}") + + return Prompt.ask("\nSelect an option", choices=list(choices.keys()), default="1") + + def process_circuits_menu(self): + """Handle circuit processing menu.""" + console.print("\n[bold yellow]Circuit Processing[/bold yellow]") + + # Initialize deployment manager if needed + if not self.deployment_manager: + output_dir = Prompt.ask("Output directory", default=self.output_dir) + log_level = Prompt.ask("Log level", choices=["DEBUG", "INFO", "WARNING", "ERROR"], default="INFO") + + with console.status("[bold green]Initializing HDH deployment manager..."): + self.deployment_manager = HDHDeploymentManager(output_dir, log_level) + self.output_dir = output_dir + + while True: + console.print("\n[bold]Circuit Processing Options:[/bold]") + options = { + "1": "Process Example Circuits", + "2": "Process QASM File", + "3": "Run Comprehensive Demo", + "4": "Custom Circuit Processing", + "b": "Back to Main Menu" + } + + for key, desc in options.items(): + console.print(f" [cyan]{key}[/cyan]. {desc}") + + choice = Prompt.ask("Select option", choices=list(options.keys()), default="1") + + if choice == "b": + break + elif choice == "1": + self.process_example_circuits() + elif choice == "2": + self.process_qasm_file() + elif choice == "3": + self.run_comprehensive_demo() + elif choice == "4": + self.custom_circuit_processing() + + def process_example_circuits(self): + """Process example circuits.""" + console.print("\n[bold]Example Circuits[/bold]") + + # Get available examples + examples = self.circuit_library.get_all_examples() + + # Show available circuits + table = Table(title="Available Example Circuits") + table.add_column("Name", style="cyan") + table.add_column("Qubits", justify="right") + table.add_column("Depth", justify="right") + table.add_column("Description", style="green") + + circuit_descriptions = { + "bell_state": "Entangled Bell state", + "ghz_3": "3-qubit GHZ state", + "ghz_4": "4-qubit GHZ state", + "w_state": "W state superposition", + "qft_3": "3-qubit Quantum Fourier Transform", + "qft_4": "4-qubit Quantum Fourier Transform", + "grover_2": "2-qubit Grover search", + "deutsch_jozsa": "Deutsch-Jozsa algorithm", + "teleportation": "Quantum teleportation protocol", + "error_correction": "3-bit error correction", + "vqe_ansatz": "VQE variational ansatz", + "qaoa": "QAOA optimization circuit" + } + + for name, circuit in examples.items(): + desc = circuit_descriptions.get(name, "Quantum circuit example") + table.add_row(name, str(circuit.num_qubits), str(circuit.depth()), desc) + + console.print(table) + + # Let user select circuits + selected = Prompt.ask( + "\nSelect circuits to process (comma-separated names or 'all')", + default="bell_state,ghz_3,qft_3" + ) + + if selected.lower() == "all": + circuits_to_process = list(examples.items()) + else: + circuit_names = [name.strip() for name in selected.split(",")] + circuits_to_process = [(name, examples[name]) for name in circuit_names if name in examples] + + save_plots = Confirm.ask("Save visualization plots?", default=True) + + # Process circuits + results = [] + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + console=console + ) as progress: + task = progress.add_task("Processing circuits...", total=len(circuits_to_process)) + + for name, circuit in circuits_to_process: + progress.update(task, description=f"Processing {name}...") + result = self.deployment_manager.process_circuit(circuit, save_plots) + results.append(result) + progress.advance(task) + + # Show results + self.show_processing_results(results) + + def process_qasm_file(self): + """Process a QASM file.""" + console.print("\n[bold]QASM File Processing[/bold]") + + qasm_file = Prompt.ask("Enter QASM file path") + + if not Path(qasm_file).exists(): + console.print(f"[red]Error: File '{qasm_file}' not found[/red]") + return + + save_plots = Confirm.ask("Save visualization plots?", default=True) + + with console.status(f"[bold green]Processing {qasm_file}..."): + result = self.deployment_manager.process_qasm_file(qasm_file, save_plots) + + self.show_processing_results([result]) + + def run_comprehensive_demo(self): + """Run comprehensive demo.""" + console.print("\n[bold]Comprehensive Demo[/bold]") + console.print("This will process multiple example circuits and generate comprehensive analysis.") + + if not Confirm.ask("Continue with comprehensive demo?", default=True): + return + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + console=console + ) as progress: + task = progress.add_task("Running comprehensive demo...", total=None) + summary = self.deployment_manager.run_comprehensive_demo() + progress.update(task, description="Demo completed!") + + # Show summary + console.print("\n[bold green]Demo Summary[/bold green]") + console.print(f"Total processed: {summary['total_processed']}") + console.print(f"Success rate: {summary['success_rate']:.2%}") + console.print(f"Average processing time: {summary['average_processing_time']:.3f}s") + console.print(f"Total nodes processed: {summary['total_nodes_processed']}") + console.print(f"Duration: {summary['deployment_duration']:.2f}s") + + def custom_circuit_processing(self): + """Custom circuit processing workflow.""" + console.print("\n[bold]Custom Circuit Processing[/bold]") + console.print("Create custom quantum circuits for HDH analysis.") + + circuit_types = { + "1": "Random Circuit", + "2": "GHZ State (custom size)", + "3": "QFT (custom size)", + "4": "Bell State Variants" + } + + for key, desc in circuit_types.items(): + console.print(f" [cyan]{key}[/cyan]. {desc}") + + choice = Prompt.ask("Select circuit type", choices=list(circuit_types.keys())) + + if choice == "1": + n_qubits = int(Prompt.ask("Number of qubits", default="4")) + depth = int(Prompt.ask("Circuit depth", default="8")) + seed = int(Prompt.ask("Random seed", default="42")) + + circuit = self.circuit_library.random_circuit(n_qubits, depth, seed) + circuit.name = f"Custom Random {n_qubits}q {depth}d" + + elif choice == "2": + n_qubits = int(Prompt.ask("Number of qubits", default="5")) + circuit = self.circuit_library.ghz_state(n_qubits) + + elif choice == "3": + n_qubits = int(Prompt.ask("Number of qubits", default="4")) + circuit = self.circuit_library.qft_circuit(n_qubits) + + elif choice == "4": + variants = self.circuit_library.bell_state_variants() + console.print(f"Processing all {len(variants)} Bell state variants...") + + results = [] + for variant in variants: + result = self.deployment_manager.process_circuit(variant, True) + results.append(result) + + self.show_processing_results(results) + return + + save_plots = Confirm.ask("Save visualization plots?", default=True) + + with console.status(f"Processing {circuit.name}..."): + result = self.deployment_manager.process_circuit(circuit, save_plots) + + self.show_processing_results([result]) + + def benchmarking_menu(self): + """Handle benchmarking menu.""" + console.print("\n[bold yellow]Performance Benchmarking[/bold yellow]") + + if not self.benchmark_suite: + output_dir = Prompt.ask("Benchmark output directory", default="benchmark_results") + repetitions = int(Prompt.ask("Number of repetitions per test", default="3")) + + with console.status("[bold green]Initializing benchmark suite..."): + self.benchmark_suite = HDHBenchmarkSuite(output_dir, repetitions) + + while True: + console.print("\n[bold]Benchmarking Options:[/bold]") + options = { + "1": "Scalability Benchmark", + "2": "Algorithm Benchmark", + "3": "Random Circuit Benchmark", + "4": "Comprehensive Benchmark", + "5": "View Benchmark Results", + "b": "Back to Main Menu" + } + + for key, desc in options.items(): + console.print(f" [cyan]{key}[/cyan]. {desc}") + + choice = Prompt.ask("Select option", choices=list(options.keys()), default="4") + + if choice == "b": + break + elif choice == "1": + self.run_scalability_benchmark() + elif choice == "2": + self.run_algorithm_benchmark() + elif choice == "3": + self.run_random_benchmark() + elif choice == "4": + self.run_comprehensive_benchmark() + elif choice == "5": + self.view_benchmark_results() + + def run_scalability_benchmark(self): + """Run scalability benchmark.""" + console.print("\n[bold]Scalability Benchmark[/bold]") + console.print("Testing HDH performance across different circuit sizes...") + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + console=console + ) as progress: + task = progress.add_task("Running scalability benchmark...", total=None) + results = self.benchmark_suite.run_scalability_benchmark() + progress.update(task, description="Scalability benchmark completed!") + + self.show_benchmark_results(results, "Scalability") + + def run_algorithm_benchmark(self): + """Run algorithm benchmark.""" + console.print("\n[bold]Algorithm Benchmark[/bold]") + console.print("Testing HDH performance on quantum algorithms...") + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + console=console + ) as progress: + task = progress.add_task("Running algorithm benchmark...", total=None) + results = self.benchmark_suite.run_algorithm_benchmark() + progress.update(task, description="Algorithm benchmark completed!") + + self.show_benchmark_results(results, "Algorithm") + + def run_random_benchmark(self): + """Run random circuit benchmark.""" + console.print("\n[bold]Random Circuit Benchmark[/bold]") + + max_qubits = int(Prompt.ask("Maximum qubits", default="6")) + max_depth = int(Prompt.ask("Maximum depth", default="20")) + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + console=console + ) as progress: + task = progress.add_task("Running random circuit benchmark...", total=None) + results = self.benchmark_suite.run_random_circuit_benchmark(max_qubits, max_depth) + progress.update(task, description="Random circuit benchmark completed!") + + self.show_benchmark_results(results, "Random Circuit") + + def run_comprehensive_benchmark(self): + """Run comprehensive benchmark.""" + console.print("\n[bold]Comprehensive Benchmark[/bold]") + console.print("Running all benchmark suites and generating analysis...") + + if not Confirm.ask("This may take several minutes. Continue?", default=True): + return + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + console=console + ) as progress: + task = progress.add_task("Running comprehensive benchmark...", total=None) + report = self.benchmark_suite.run_full_benchmark() + progress.update(task, description="Comprehensive benchmark completed!") + + # Show comprehensive report + console.print("\n[bold green]Comprehensive Benchmark Report[/bold green]") + + if "error" not in report: + summary = report["benchmark_summary"] + perf_stats = report["performance_statistics"] + + table = Table(title="Benchmark Summary") + table.add_column("Metric", style="cyan") + table.add_column("Value", style="green") + + table.add_row("Total Circuits", str(summary['total_circuits'])) + table.add_row("Success Rate", "100%") + table.add_row("Avg Conversion Time", f"{perf_stats['conversion_time']['mean']:.4f}s") + table.add_row("Avg Memory Usage", f"{perf_stats['memory_usage']['mean']:.2f}MB") + table.add_row("Largest Circuit", f"{report['scalability_analysis']['largest_circuit_qubits']} qubits") + table.add_row("Total Time", f"{summary.get('total_benchmark_time', 0):.2f}s") + + console.print(table) + console.print(f"\nšŸ“ Results saved in: {self.benchmark_suite.output_dir}") + else: + console.print(f"[red]Benchmark failed: {report['error']}[/red]") + + def circuit_library_menu(self): + """Handle circuit library menu.""" + console.print("\n[bold yellow]Circuit Library & Examples[/bold yellow]") + + while True: + console.print("\n[bold]Circuit Library Options:[/bold]") + options = { + "1": "View Available Circuits", + "2": "Generate QASM Examples", + "3": "Create Custom Circuit", + "4": "Circuit Information", + "b": "Back to Main Menu" + } + + for key, desc in options.items(): + console.print(f" [cyan]{key}[/cyan]. {desc}") + + choice = Prompt.ask("Select option", choices=list(options.keys()), default="1") + + if choice == "b": + break + elif choice == "1": + self.view_available_circuits() + elif choice == "2": + self.generate_qasm_examples() + elif choice == "3": + self.create_custom_circuit() + elif choice == "4": + self.show_circuit_information() + + def view_available_circuits(self): + """View available circuits in the library.""" + console.print("\n[bold]Available Quantum Circuits[/bold]") + + examples = self.circuit_library.get_all_examples() + + # Create tree structure + tree = Tree("HDH Circuit Library") + + # Group circuits by category + categories = { + "Basic States": ["bell_state", "ghz_3", "ghz_4", "w_state"], + "Algorithms": ["qft_3", "qft_4", "grover_2", "deutsch_jozsa"], + "Protocols": ["teleportation", "error_correction"], + "Variational": ["vqe_ansatz", "qaoa"], + "Random": ["random_small", "random_medium"] + } + + for category, circuit_names in categories.items(): + category_branch = tree.add(f"[bold blue]{category}[/bold blue]") + for name in circuit_names: + if name in examples: + circuit = examples[name] + info = f"{name} - {circuit.num_qubits}q, depth {circuit.depth()}" + category_branch.add(f"[green]{info}[/green]") + + console.print(tree) + + # Show benchmark suite info + benchmark_circuits = self.circuit_library.get_benchmark_suite() + console.print(f"\n[bold]Benchmark Suite:[/bold] {len(benchmark_circuits)} circuits") + + def generate_qasm_examples(self): + """Generate QASM example files.""" + console.print("\n[bold]Generate QASM Examples[/bold]") + + output_dir = Prompt.ask("Output directory for QASM files", default="qasm_examples") + + with console.status(f"Generating QASM examples in {output_dir}..."): + from circuit_examples import save_example_qasm_files + save_example_qasm_files(output_dir) + + console.print(f"[green]QASM examples generated in {output_dir}/[/green]") + + def show_processing_results(self, results: List[Dict[str, Any]]): + """Display circuit processing results.""" + console.print("\n[bold green]Processing Results[/bold green]") + + table = Table(title="Circuit Processing Summary") + table.add_column("Circuit", style="cyan") + table.add_column("Status", style="green") + table.add_column("Nodes", justify="right") + table.add_column("Edges", justify="right") + table.add_column("Time (s)", justify="right") + + for result in results: + status = "āœ… Success" if result['success'] else "āŒ Failed" + if result['success']: + table.add_row( + result['circuit_name'], + status, + str(result['hdh_stats']['nodes']), + str(result['hdh_stats']['edges']), + f"{result['processing_time']:.3f}" + ) + else: + table.add_row( + result['circuit_name'], + status, + "-", "-", + f"{result['processing_time']:.3f}" + ) + + console.print(table) + + # Show output directory + console.print(f"\nšŸ“ Results saved in: {self.output_dir}") + + def show_benchmark_results(self, results: List, benchmark_type: str): + """Display benchmark results.""" + console.print(f"\n[bold green]{benchmark_type} Benchmark Results[/bold green]") + + if not results: + console.print("[red]No results to display[/red]") + return + + table = Table(title=f"{benchmark_type} Performance") + table.add_column("Circuit", style="cyan") + table.add_column("Qubits", justify="right") + table.add_column("Conversion (s)", justify="right") + table.add_column("Memory (MB)", justify="right") + table.add_column("HDH Nodes", justify="right") + + for result in results: + if result.success: + table.add_row( + result.circuit_name, + str(result.num_qubits), + f"{result.conversion_time:.4f}", + f"{result.memory_peak_mb:.2f}", + str(result.hdh_nodes) + ) + + console.print(table) + + def configuration_menu(self): + """Handle configuration menu.""" + console.print("\n[bold yellow]Configuration & Settings[/bold yellow]") + + console.print("Current settings:") + console.print(f" Output directory: [cyan]{self.output_dir}[/cyan]") + + if self.deployment_manager: + console.print(" Deployment manager: [green]Initialized[/green]") + else: + console.print(" Deployment manager: [red]Not initialized[/red]") + + if self.benchmark_suite: + console.print(" Benchmark suite: [green]Initialized[/green]") + else: + console.print(" Benchmark suite: [red]Not initialized[/red]") + + # Configuration options + if Confirm.ask("\nUpdate output directory?", default=False): + self.output_dir = Prompt.ask("New output directory", default=self.output_dir) + # Reset managers to pick up new directory + self.deployment_manager = None + self.benchmark_suite = None + + def view_results_menu(self): + """Handle results viewing menu.""" + console.print("\n[bold yellow]View Results & Reports[/bold yellow]") + + # Look for result files + result_dirs = [ + Path(self.output_dir), + Path("hdh_results"), + Path("benchmark_results") + ] + + found_results = [] + for result_dir in result_dirs: + if result_dir.exists(): + json_files = list(result_dir.glob("*.json")) + png_files = list(result_dir.glob("*.png")) + found_results.append((result_dir, json_files, png_files)) + + if not found_results: + console.print("[red]No result files found[/red]") + return + + # Display found results + for result_dir, json_files, png_files in found_results: + console.print(f"\n[bold]Results in {result_dir}:[/bold]") + + if json_files: + console.print(" [cyan]JSON Reports:[/cyan]") + for json_file in json_files: + console.print(f" ā€¢ {json_file.name}") + + if png_files: + console.print(" [cyan]Visualizations:[/cyan]") + for png_file in png_files: + console.print(f" ā€¢ {png_file.name}") + + # Option to view specific files + if Confirm.ask("\nView a specific JSON report?", default=False): + all_json = [] + for _, json_files, _ in found_results: + all_json.extend(json_files) + + if all_json: + file_choices = {str(i+1): f.name for i, f in enumerate(all_json)} + file_choices["b"] = "Back" + + console.print("\nAvailable JSON reports:") + for key, filename in file_choices.items(): + console.print(f" [cyan]{key}[/cyan]. {filename}") + + choice = Prompt.ask("Select file", choices=list(file_choices.keys())) + + if choice != "b": + selected_file = all_json[int(choice) - 1] + self.view_json_report(selected_file) + + def view_json_report(self, json_file: Path): + """View a JSON report file.""" + try: + with open(json_file, 'r') as f: + data = json.load(f) + + # Pretty print JSON with syntax highlighting + json_str = json.dumps(data, indent=2) + syntax = Syntax(json_str, "json", theme="monokai", line_numbers=True) + + console.print(f"\n[bold]Contents of {json_file.name}:[/bold]") + console.print(syntax) + + except Exception as e: + console.print(f"[red]Error reading {json_file}: {str(e)}[/red]") + + def help_menu(self): + """Display help and documentation.""" + console.print("\n[bold yellow]Help & Documentation[/bold yellow]") + + help_text = """ +[bold blue]HDH CLI Help[/bold blue] + +[bold]About HDH:[/bold] +HDH (Hybrid Dependency Hypergraph) is a library for representing and analyzing +quantum circuits using hypergraph structures. It enables: +ā€¢ Circuit conversion from multiple quantum frameworks +ā€¢ Dependency analysis and visualization +ā€¢ Circuit partitioning and optimization +ā€¢ Performance benchmarking + +[bold]Main Features:[/bold] +1. [cyan]Circuit Processing[/cyan] - Convert and analyze quantum circuits +2. [cyan]Benchmarking[/cyan] - Measure HDH performance across different circuits +3. [cyan]Circuit Library[/cyan] - Access pre-built quantum circuit examples +4. [cyan]Configuration[/cyan] - Customize settings and output directories +5. [cyan]Results Viewing[/cyan] - Examine processing results and reports + +[bold]Supported Circuit Types:[/bold] +ā€¢ Qiskit QuantumCircuit objects +ā€¢ OpenQASM 2.0 files +ā€¢ Pre-built examples (Bell states, GHZ, QFT, Grover, etc.) + +[bold]Output Files:[/bold] +ā€¢ JSON reports with detailed metrics +ā€¢ PNG visualizations of HDH structures +ā€¢ Performance benchmark plots +ā€¢ Processing logs + +[bold]Tips:[/bold] +ā€¢ Start with small circuits (< 8 qubits) for faster processing +ā€¢ Use the comprehensive demo for a full overview +ā€¢ Enable plot saving to visualize HDH structures +ā€¢ Run benchmarks to understand performance characteristics + +[italic]Special thanks to Maria Gragera Garces for the HDH library![/italic] + """ + + console.print(Panel(help_text.strip(), expand=False, border_style="blue")) + + if Confirm.ask("\nWould you like to see example usage?", default=False): + self.show_example_usage() + + def show_example_usage(self): + """Show example usage scenarios.""" + console.print("\n[bold]Example Usage Scenarios[/bold]") + + examples = [ + { + "title": "Quick Start - Process Bell State", + "steps": [ + "1. Select 'Process Quantum Circuits'", + "2. Choose 'Process Example Circuits'", + "3. Enter 'bell_state' when prompted", + "4. Confirm to save plots", + "5. View results in output directory" + ] + }, + { + "title": "Benchmark Suite", + "steps": [ + "1. Select 'Run Performance Benchmarks'", + "2. Choose 'Comprehensive Benchmark'", + "3. Confirm to run (may take several minutes)", + "4. View generated performance plots", + "5. Check benchmark_report.json for detailed metrics" + ] + }, + { + "title": "Custom QASM Processing", + "steps": [ + "1. Select 'Process Quantum Circuits'", + "2. Choose 'Process QASM File'", + "3. Enter path to your .qasm file", + "4. Confirm visualization settings", + "5. Review HDH conversion results" + ] + } + ] + + for example in examples: + console.print(f"\n[bold green]{example['title']}:[/bold green]") + for step in example['steps']: + console.print(f" {step}") + + def create_custom_circuit(self): + """Guide for creating custom circuits.""" + console.print("\n[bold]Custom Circuit Creation[/bold]") + console.print("This feature allows you to create custom quantum circuits programmatically.") + console.print("For now, use the 'Custom Circuit Processing' option in the main processing menu.") + + def show_circuit_information(self): + """Show detailed information about circuits.""" + console.print("\n[bold]Circuit Information[/bold]") + + examples = self.circuit_library.get_all_examples() + circuit_name = Prompt.ask( + "Enter circuit name for details", + choices=list(examples.keys()), + default="bell_state" + ) + + if circuit_name in examples: + circuit = examples[circuit_name] + + info_table = Table(title=f"Circuit Information: {circuit_name}") + info_table.add_column("Property", style="cyan") + info_table.add_column("Value", style="green") + + info_table.add_row("Name", getattr(circuit, 'name', circuit_name)) + info_table.add_row("Qubits", str(circuit.num_qubits)) + info_table.add_row("Classical Bits", str(circuit.num_clbits)) + info_table.add_row("Depth", str(circuit.depth())) + info_table.add_row("Size (gates)", str(circuit.size())) + info_table.add_row("Operations", str(len(circuit.data))) + + console.print(info_table) + + # Show QASM representation + if Confirm.ask("Show QASM representation?", default=False): + try: + qasm_str = circuit.qasm() + syntax = Syntax(qasm_str, "qasm", theme="monokai", line_numbers=True) + console.print(f"\n[bold]QASM for {circuit_name}:[/bold]") + console.print(syntax) + except Exception as e: + console.print(f"[red]Error generating QASM: {str(e)}[/red]") + + def run(self): + """Main CLI loop.""" + try: + while True: + choice = self.show_main_menu() + + if choice == "q": + console.print("\n[bold blue]Thank you for using HDH CLI![/bold blue]") + console.print("[italic]Special thanks to Maria Gragera Garces! šŸŽ‰[/italic]") + break + elif choice == "1": + self.process_circuits_menu() + elif choice == "2": + self.benchmarking_menu() + elif choice == "3": + self.circuit_library_menu() + elif choice == "4": + self.configuration_menu() + elif choice == "5": + self.view_results_menu() + elif choice == "6": + self.help_menu() + + except KeyboardInterrupt: + console.print("\n[yellow]Goodbye![/yellow]") + except Exception as e: + console.print(f"\n[red]An error occurred: {str(e)}[/red]") + + +@click.command() +@click.option('--output-dir', default='hdh_results', help='Default output directory') +@click.option('--theme', default='default', help='Console theme') +def main(output_dir, theme): + """ + HDH Command Line Interface + + Interactive CLI for HDH (Hybrid Dependency Hypergraph) deployment and analysis. + """ + try: + cli = HDHCommandLineInterface() + cli.output_dir = output_dir + cli.run() + except Exception as e: + console.print(f"[red]Failed to start CLI: {str(e)}[/red]") + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..6090285 --- /dev/null +++ b/config.yaml @@ -0,0 +1,66 @@ +# HDH Deployment Configuration +# Configuration file for the HDH deployment example + +# Logging configuration +logging: + level: INFO + format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + file: "hdh_deployment.log" + +# Output settings +output: + directory: "hdh_results" + save_plots: true + plot_format: "png" + plot_dpi: 300 + +# Circuit processing settings +circuits: + max_qubits: 10 + default_partitions: 3 + enable_visualization: true + save_intermediate: false + +# Performance settings +performance: + timeout_seconds: 300 + max_memory_gb: 8 + parallel_processing: false + +# QASM processing +qasm: + supported_versions: ["2.0"] + max_file_size_mb: 10 + validate_syntax: true + +# Benchmarking settings +benchmark: + repetitions: 3 + warmup_runs: 1 + measure_memory: true + save_detailed_metrics: true + +# Visualization settings +visualization: + figsize: [12, 8] + node_colors: + quantum: "#FF6B6B" + classical: "#4DABF7" + edge_colors: + quantum: "#FF8E53" + classical: "#69DB7C" + layout: "spring" + node_size: 100 + edge_width: 1.5 + +# Docker settings (for containerized deployment) +docker: + base_image: "python:3.11-slim" + working_dir: "/app" + expose_port: 8080 + +# Development settings +development: + debug_mode: false + verbose_logging: false + save_debug_info: false \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..e95900f --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,192 @@ +version: '3.8' + +# HDH Deployment Docker Compose +# Comprehensive deployment setup for HDH library examples +# Special thanks to Maria Gragera Garces for the HDH library! + +services: + # Main HDH deployment service + hdh-deployment: + build: + context: . + dockerfile: Dockerfile + args: + BUILD_DATE: ${BUILD_DATE:-$(date -u +'%Y-%m-%dT%H:%M:%SZ')} + VERSION: ${VERSION:-1.0.0} + container_name: hdh-deployment + restart: unless-stopped + environment: + - HDH_OUTPUT_DIR=/app/hdh_results + - HDH_LOG_LEVEL=${HDH_LOG_LEVEL:-INFO} + - PYTHONPATH=/app + - MPLBACKEND=Agg + volumes: + - hdh_results:/app/hdh_results + - benchmark_results:/app/benchmark_results + - logs:/app/logs + - ./config.yaml:/app/config.yaml:ro + # Mount additional QASM files if available + - ${QASM_DIR:-./qasm_examples}:/app/qasm_examples:ro + command: ["python", "main.py", "--demo-mode", "--output-dir", "/app/hdh_results"] + networks: + - hdh-network + labels: + - "hdh.service=deployment" + - "hdh.description=Main HDH deployment service" + - "hdh.credits=Thanks to Maria Gragera Garces" + + # Benchmarking service + hdh-benchmark: + build: + context: . + dockerfile: Dockerfile + container_name: hdh-benchmark + restart: "no" # Run once + environment: + - HDH_OUTPUT_DIR=/app/benchmark_results + - HDH_LOG_LEVEL=${HDH_LOG_LEVEL:-INFO} + - PYTHONPATH=/app + - MPLBACKEND=Agg + volumes: + - benchmark_results:/app/benchmark_results + - logs:/app/logs + - ./config.yaml:/app/config.yaml:ro + command: [ + "python", "benchmark.py", + "--suite", "all", + "--repetitions", "${BENCHMARK_REPETITIONS:-3}", + "--output-dir", "/app/benchmark_results" + ] + networks: + - hdh-network + labels: + - "hdh.service=benchmark" + - "hdh.description=HDH performance benchmarking" + profiles: + - benchmark + + # Circuit examples generator + hdh-examples: + build: + context: . + dockerfile: Dockerfile + container_name: hdh-examples + restart: "no" # Run once + environment: + - PYTHONPATH=/app + volumes: + - qasm_examples:/app/qasm_examples + - logs:/app/logs + command: ["python", "circuit_examples.py"] + networks: + - hdh-network + labels: + - "hdh.service=examples" + - "hdh.description=Generate quantum circuit examples" + profiles: + - examples + + # Jupyter notebook service (optional) + hdh-jupyter: + build: + context: . + dockerfile: Dockerfile + container_name: hdh-jupyter + restart: unless-stopped + environment: + - JUPYTER_ENABLE_LAB=yes + - JUPYTER_TOKEN=${JUPYTER_TOKEN:-hdh-token} + - PYTHONPATH=/app + volumes: + - hdh_results:/app/hdh_results + - benchmark_results:/app/benchmark_results + - ./notebooks:/app/notebooks + - ./config.yaml:/app/config.yaml:ro + ports: + - "${JUPYTER_PORT:-8888}:8888" + command: [ + "jupyter", "lab", + "--ip=0.0.0.0", + "--port=8888", + "--no-browser", + "--allow-root", + "--notebook-dir=/app" + ] + networks: + - hdh-network + labels: + - "hdh.service=jupyter" + - "hdh.description=Jupyter Lab for interactive HDH analysis" + profiles: + - jupyter + + # Web dashboard (placeholder for future development) + hdh-dashboard: + build: + context: . + dockerfile: Dockerfile + container_name: hdh-dashboard + restart: unless-stopped + environment: + - FLASK_APP=dashboard.py + - FLASK_ENV=${FLASK_ENV:-production} + - HDH_RESULTS_DIR=/app/hdh_results + volumes: + - hdh_results:/app/hdh_results:ro + - benchmark_results:/app/benchmark_results:ro + ports: + - "${DASHBOARD_PORT:-8080}:8080" + command: ["python", "-c", "print('Dashboard service placeholder - to be implemented')"] + networks: + - hdh-network + labels: + - "hdh.service=dashboard" + - "hdh.description=Web dashboard for HDH results" + profiles: + - dashboard + +# Named volumes for persistent storage +volumes: + hdh_results: + driver: local + labels: + - "hdh.volume=results" + - "hdh.description=HDH processing results" + + benchmark_results: + driver: local + labels: + - "hdh.volume=benchmarks" + - "hdh.description=Performance benchmark results" + + qasm_examples: + driver: local + labels: + - "hdh.volume=examples" + - "hdh.description=Generated QASM examples" + + logs: + driver: local + labels: + - "hdh.volume=logs" + - "hdh.description=Application logs" + +# Custom network +networks: + hdh-network: + driver: bridge + labels: + - "hdh.network=main" + - "hdh.description=HDH deployment network" + +# Environment variables with defaults +# Create a .env file to override these values: +# +# HDH_LOG_LEVEL=DEBUG +# BENCHMARK_REPETITIONS=5 +# JUPYTER_TOKEN=your-secure-token +# JUPYTER_PORT=8888 +# DASHBOARD_PORT=8080 +# QASM_DIR=./qasm_files +# BUILD_DATE=2024-01-01T00:00:00Z +# VERSION=1.0.0 \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..4349be6 --- /dev/null +++ b/main.py @@ -0,0 +1,459 @@ +#!/usr/bin/env python3 +""" +HDH Deployment Example - Comprehensive Demo +========================================== + +This example demonstrates real-world deployment of the HDH (Hybrid Dependency Hypergraph) +library for quantum computation analysis and visualization. + +Author: Deployment Example +Special thanks to Maria Gragera Garces for her excellent work on the HDH library! + +Features demonstrated: +- Quantum circuit conversion to HDH format +- QASM file processing +- Circuit analysis and metrics +- Visualization generation +- Partitioning and optimization +- Error handling and logging +- Performance monitoring +""" + +import os +import sys +import time +import logging +import argparse +import json +from pathlib import Path +from typing import Dict, List, Tuple, Optional, Any +from datetime import datetime + +# Add HDH to path +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'HDH'))) + +# HDH imports +from hdh import HDH, plot_hdh +from hdh.converters.qiskit import from_qiskit +from hdh.converters.qasm import from_qasm +from hdh.passes.cut import compute_cut, cost, partition_sizes, compute_parallelism_by_time + +# Quantum computing imports +from qiskit import QuantumCircuit, QuantumRegister, ClassicalRegister +from qiskit.circuit.library import QFT, GroverOperator +import matplotlib.pyplot as plt + + +class HDHDeploymentManager: + """ + Main deployment manager for HDH operations. + Handles circuit processing, analysis, and reporting. + """ + + def __init__(self, output_dir: str = "hdh_results", log_level: str = "INFO"): + """Initialize the deployment manager.""" + self.output_dir = Path(output_dir) + self.output_dir.mkdir(exist_ok=True) + + # Setup logging + self.setup_logging(log_level) + self.logger = logging.getLogger(__name__) + + # Performance tracking + self.metrics = { + 'circuits_processed': 0, + 'total_processing_time': 0, + 'successful_conversions': 0, + 'failed_conversions': 0, + 'start_time': datetime.now() + } + + self.logger.info("HDH Deployment Manager initialized") + self.logger.info(f"Output directory: {self.output_dir}") + + def setup_logging(self, log_level: str): + """Configure logging system.""" + log_file = self.output_dir / "hdh_deployment.log" + + logging.basicConfig( + level=getattr(logging, log_level.upper()), + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler(log_file), + logging.StreamHandler(sys.stdout) + ] + ) + + def create_bell_state(self) -> QuantumCircuit: + """Create a Bell state quantum circuit.""" + qc = QuantumCircuit(2, 2) + qc.h(0) + qc.cx(0, 1) + qc.measure_all() + qc.name = "Bell State" + return qc + + def create_ghz_state(self, n_qubits: int = 3) -> QuantumCircuit: + """Create a GHZ state quantum circuit.""" + qc = QuantumCircuit(n_qubits, n_qubits) + qc.h(0) + for i in range(1, n_qubits): + qc.cx(0, i) + qc.measure_all() + qc.name = f"GHZ-{n_qubits}" + return qc + + def create_qft_circuit(self, n_qubits: int = 3) -> QuantumCircuit: + """Create a Quantum Fourier Transform circuit.""" + qc = QuantumCircuit(n_qubits, n_qubits) + qft = QFT(n_qubits) + qc.compose(qft, inplace=True) + qc.measure_all() + qc.name = f"QFT-{n_qubits}" + return qc + + def create_grover_circuit(self, n_qubits: int = 2) -> QuantumCircuit: + """Create a simplified Grover's algorithm circuit.""" + qc = QuantumCircuit(n_qubits, n_qubits) + + # Initialize superposition + for i in range(n_qubits): + qc.h(i) + + # Oracle (mark |11āŸ© for 2 qubits) + if n_qubits == 2: + qc.cz(0, 1) + + # Diffusion operator + for i in range(n_qubits): + qc.h(i) + qc.x(i) + + if n_qubits == 2: + qc.cz(0, 1) + + for i in range(n_qubits): + qc.x(i) + qc.h(i) + + qc.measure_all() + qc.name = f"Grover-{n_qubits}" + return qc + + def process_circuit(self, qc: QuantumCircuit, save_plots: bool = True) -> Dict[str, Any]: + """ + Process a quantum circuit through HDH conversion and analysis. + + Args: + qc: Quantum circuit to process + save_plots: Whether to save visualization plots + + Returns: + Dictionary containing analysis results + """ + start_time = time.time() + circuit_name = getattr(qc, 'name', f'Circuit_{self.metrics["circuits_processed"]}') + + try: + self.logger.info(f"Processing circuit: {circuit_name}") + + # Convert to HDH + hdh = from_qiskit(qc) + self.logger.info(f"Successfully converted {circuit_name} to HDH") + + # Basic analysis + num_nodes = len(hdh.S) + num_edges = len(hdh.C) + num_timesteps = len(hdh.T) + + self.logger.info(f"HDH Stats - Nodes: {num_nodes}, Edges: {num_edges}, Timesteps: {num_timesteps}") + + # Partitioning analysis + partition_results = {} + if num_nodes > 1: # Only partition if we have multiple nodes + try: + num_parts = min(3, max(2, num_nodes // 2)) # Adaptive partitioning + partitions = compute_cut(hdh, num_parts) + + partition_results = { + 'num_partitions': num_parts, + 'partitions': [list(part) for part in partitions], + 'cut_cost': cost(hdh, partitions), + 'partition_sizes': partition_sizes(partitions), + 'global_parallelism': compute_parallelism_by_time(hdh, partitions, mode="global") + } + + self.logger.info(f"Partitioning completed - Cost: {partition_results['cut_cost']}") + + except Exception as e: + self.logger.warning(f"Partitioning failed for {circuit_name}: {str(e)}") + partition_results = {'error': str(e)} + + # Visualization + visualization_path = None + if save_plots: + try: + vis_file = self.output_dir / f"{circuit_name.replace(' ', '_')}_hdh.png" + plt.figure(figsize=(12, 8)) + plot_hdh(hdh, save_path=str(vis_file)) + plt.title(f"HDH Visualization: {circuit_name}") + plt.tight_layout() + plt.savefig(vis_file, dpi=300, bbox_inches='tight') + plt.close() + visualization_path = str(vis_file) + self.logger.info(f"Visualization saved: {vis_file}") + + except Exception as e: + self.logger.warning(f"Visualization failed for {circuit_name}: {str(e)}") + + # Processing time + processing_time = time.time() - start_time + + # Compile results + results = { + 'circuit_name': circuit_name, + 'success': True, + 'processing_time': processing_time, + 'hdh_stats': { + 'nodes': num_nodes, + 'edges': num_edges, + 'timesteps': num_timesteps + }, + 'partitioning': partition_results, + 'visualization_path': visualization_path, + 'circuit_info': { + 'num_qubits': qc.num_qubits, + 'num_clbits': qc.num_clbits, + 'depth': qc.depth(), + 'size': qc.size() + } + } + + # Update metrics + self.metrics['circuits_processed'] += 1 + self.metrics['successful_conversions'] += 1 + self.metrics['total_processing_time'] += processing_time + + return results + + except Exception as e: + processing_time = time.time() - start_time + self.logger.error(f"Failed to process {circuit_name}: {str(e)}") + + self.metrics['circuits_processed'] += 1 + self.metrics['failed_conversions'] += 1 + self.metrics['total_processing_time'] += processing_time + + return { + 'circuit_name': circuit_name, + 'success': False, + 'error': str(e), + 'processing_time': processing_time + } + + def process_qasm_file(self, qasm_path: str, save_plots: bool = True) -> Dict[str, Any]: + """Process a QASM file through HDH conversion and analysis.""" + if not os.path.exists(qasm_path): + self.logger.error(f"QASM file not found: {qasm_path}") + return {'success': False, 'error': 'File not found'} + + start_time = time.time() + file_name = Path(qasm_path).stem + + try: + self.logger.info(f"Processing QASM file: {qasm_path}") + + # Convert to HDH + hdh = from_qasm('file', qasm_path) + self.logger.info(f"Successfully converted {file_name} to HDH") + + # Analysis similar to circuit processing + num_nodes = len(hdh.S) + num_edges = len(hdh.C) + num_timesteps = len(hdh.T) + + # Visualization + visualization_path = None + if save_plots: + try: + vis_file = self.output_dir / f"{file_name}_hdh.png" + plt.figure(figsize=(12, 8)) + plot_hdh(hdh, save_path=str(vis_file)) + plt.title(f"HDH Visualization: {file_name}") + plt.tight_layout() + plt.savefig(vis_file, dpi=300, bbox_inches='tight') + plt.close() + visualization_path = str(vis_file) + self.logger.info(f"Visualization saved: {vis_file}") + + except Exception as e: + self.logger.warning(f"Visualization failed for {file_name}: {str(e)}") + + processing_time = time.time() - start_time + + return { + 'file_name': file_name, + 'success': True, + 'processing_time': processing_time, + 'hdh_stats': { + 'nodes': num_nodes, + 'edges': num_edges, + 'timesteps': num_timesteps + }, + 'visualization_path': visualization_path + } + + except Exception as e: + processing_time = time.time() - start_time + self.logger.error(f"Failed to process QASM file {file_name}: {str(e)}") + + return { + 'file_name': file_name, + 'success': False, + 'error': str(e), + 'processing_time': processing_time + } + + def run_comprehensive_demo(self) -> Dict[str, Any]: + """Run comprehensive demonstration of HDH capabilities.""" + self.logger.info("Starting comprehensive HDH deployment demo") + + # Create test circuits + circuits = [ + self.create_bell_state(), + self.create_ghz_state(3), + self.create_ghz_state(4), + self.create_qft_circuit(3), + self.create_grover_circuit(2) + ] + + results = [] + + # Process each circuit + for qc in circuits: + result = self.process_circuit(qc, save_plots=True) + results.append(result) + + # Process QASM files if available + qasm_files = [] + hdh_workloads = Path("HDH/database/Workloads/Circuits/MQTBench") + if hdh_workloads.exists(): + qasm_files = list(hdh_workloads.glob("*.qasm"))[:3] # Process first 3 + + for qasm_file in qasm_files: + result = self.process_qasm_file(str(qasm_file), save_plots=True) + results.append(result) + + # Generate summary + summary = self.generate_summary(results) + + # Save results + results_file = self.output_dir / "deployment_results.json" + with open(results_file, 'w') as f: + json.dump({ + 'summary': summary, + 'detailed_results': results, + 'metrics': self.metrics + }, f, indent=2, default=str) + + self.logger.info(f"Demo completed. Results saved to {results_file}") + return summary + + def generate_summary(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: + """Generate summary of processing results.""" + successful = [r for r in results if r.get('success', False)] + failed = [r for r in results if not r.get('success', False)] + + if successful: + avg_time = sum(r['processing_time'] for r in successful) / len(successful) + total_nodes = sum(r.get('hdh_stats', {}).get('nodes', 0) for r in successful) + total_edges = sum(r.get('hdh_stats', {}).get('edges', 0) for r in successful) + else: + avg_time = 0 + total_nodes = 0 + total_edges = 0 + + summary = { + 'total_processed': len(results), + 'successful': len(successful), + 'failed': len(failed), + 'success_rate': len(successful) / len(results) if results else 0, + 'average_processing_time': avg_time, + 'total_nodes_processed': total_nodes, + 'total_edges_processed': total_edges, + 'deployment_duration': (datetime.now() - self.metrics['start_time']).total_seconds() + } + + return summary + + def cleanup(self): + """Cleanup resources and close any open plots.""" + plt.close('all') + self.logger.info("HDH Deployment Manager cleanup completed") + + +def main(): + """Main deployment function.""" + parser = argparse.ArgumentParser(description="HDH Deployment Example") + parser.add_argument("--output-dir", default="hdh_results", help="Output directory for results") + parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) + parser.add_argument("--no-plots", action="store_true", help="Disable plot generation") + parser.add_argument("--qasm-file", help="Specific QASM file to process") + parser.add_argument("--demo-mode", action="store_true", help="Run comprehensive demo") + + args = parser.parse_args() + + # Initialize deployment manager + manager = HDHDeploymentManager( + output_dir=args.output_dir, + log_level=args.log_level + ) + + try: + if args.qasm_file: + # Process specific QASM file + result = manager.process_qasm_file(args.qasm_file, save_plots=not args.no_plots) + print(f"Processing result: {json.dumps(result, indent=2, default=str)}") + + elif args.demo_mode: + # Run comprehensive demo + summary = manager.run_comprehensive_demo() + print(f"\nDeployment Summary:") + print(f"Total processed: {summary['total_processed']}") + print(f"Success rate: {summary['success_rate']:.2%}") + print(f"Average processing time: {summary['average_processing_time']:.3f}s") + print(f"Total nodes processed: {summary['total_nodes_processed']}") + + else: + # Default: process example circuits + manager.logger.info("Running default circuit processing examples") + + circuits = [ + manager.create_bell_state(), + manager.create_ghz_state(3), + manager.create_qft_circuit(3) + ] + + for qc in circuits: + result = manager.process_circuit(qc, save_plots=not args.no_plots) + print(f"\nProcessed {result['circuit_name']}:") + print(f" Success: {result['success']}") + if result['success']: + print(f" Nodes: {result['hdh_stats']['nodes']}") + print(f" Edges: {result['hdh_stats']['edges']}") + print(f" Processing time: {result['processing_time']:.3f}s") + + print(f"\nšŸŽ‰ Thank you Maria Gragera Garces for the excellent HDH library! šŸŽ‰") + print(f"Results saved in: {manager.output_dir}") + + except KeyboardInterrupt: + manager.logger.info("Deployment interrupted by user") + except Exception as e: + manager.logger.error(f"Deployment failed: {str(e)}") + raise + finally: + manager.cleanup() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..37c0eca --- /dev/null +++ b/requirements.txt @@ -0,0 +1,45 @@ +# HDH Deployment Example Requirements +# Dependencies for the comprehensive HDH deployment demonstration + +# Core HDH dependencies (from pyproject.toml) +qiskit>=1.0 +networkx>=3.0 +matplotlib>=3.0 +metis==0.2a5 + +# Additional deployment dependencies +numpy>=1.21.0 +scipy>=1.7.0 +pandas>=1.3.0 + +# Logging and configuration +pyyaml>=6.0 +python-dotenv>=0.19.0 + +# CLI enhancements +click>=8.0.0 +rich>=12.0.0 +tqdm>=4.64.0 + +# Testing and development +pytest>=7.0.0 +pytest-cov>=4.0.0 +black>=22.0.0 +isort>=5.10.0 +flake8>=5.0.0 + +# Documentation +sphinx>=5.0.0 +sphinx-rtd-theme>=1.0.0 + +# Performance monitoring +psutil>=5.9.0 +memory-profiler>=0.60.0 + +# Data handling +h5py>=3.7.0 +joblib>=1.2.0 + +# Jupyter support (optional) +jupyter>=1.0.0 +ipykernel>=6.15.0 \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..8767493 --- /dev/null +++ b/setup.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +""" +Setup script for HDH Deployment Example + +This script helps set up the deployment environment for the HDH library examples. +Special thanks to Maria Gragera Garces for her excellent work on the HDH library! +""" + +from setuptools import setup, find_packages +from pathlib import Path + +# Read requirements +requirements_file = Path(__file__).parent / "requirements.txt" +with open(requirements_file) as f: + requirements = [ + line.strip() + for line in f + if line.strip() and not line.startswith('#') + ] + +# Read README for long description +readme_file = Path(__file__).parent / "README.md" +long_description = "" +if readme_file.exists(): + with open(readme_file, encoding='utf-8') as f: + long_description = f.read() + +setup( + name="hdh-deployment-example", + version="1.0.0", + description="Comprehensive deployment example for HDH (Hybrid Dependency Hypergraph) library", + long_description=long_description, + long_description_content_type="text/markdown", + author="HDH Deployment Team", + author_email="example@hdh-deployment.com", + url="https://github.com/grageragarces/hdh", + packages=find_packages(), + install_requires=requirements, + python_requires=">=3.10", + classifiers=[ + "Development Status :: 4 - Beta", + "Intended Audience :: Science/Research", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Scientific/Engineering :: Physics", + "Topic :: Software Development :: Libraries :: Python Modules", + ], + keywords=[ + "quantum computing", + "hypergraph", + "quantum circuits", + "hdh", + "dependency analysis", + "quantum compilation" + ], + entry_points={ + 'console_scripts': [ + 'hdh-deploy=main:main', + 'hdh-benchmark=benchmark:main', + 'hdh-cli=cli:main', + ], + }, + extras_require={ + 'dev': [ + 'pytest>=7.0.0', + 'pytest-cov>=4.0.0', + 'black>=22.0.0', + 'isort>=5.10.0', + 'flake8>=5.0.0', + ], + 'docs': [ + 'sphinx>=5.0.0', + 'sphinx-rtd-theme>=1.0.0', + ], + 'jupyter': [ + 'jupyter>=1.0.0', + 'ipykernel>=6.15.0', + ], + }, + project_urls={ + 'Bug Reports': 'https://github.com/grageragarces/hdh/issues', + 'Source': 'https://github.com/grageragarces/hdh', + 'Documentation': 'https://grageragarces.github.io/HDH/', + }, +) \ No newline at end of file diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..f46c80c --- /dev/null +++ b/utils.py @@ -0,0 +1,524 @@ +#!/usr/bin/env python3 +""" +HDH Deployment Utilities +======================== + +Utility functions and helper classes for the HDH deployment example. + +Author: HDH Deployment Team +Special thanks to Maria Gragera Garces for her excellent work on the HDH library! +""" + +import os +import sys +import json +import yaml +import time +import logging +from pathlib import Path +from typing import Dict, Any, List, Optional, Union +from datetime import datetime +import matplotlib.pyplot as plt +import numpy as np + + +class ConfigManager: + """Configuration management for HDH deployment.""" + + def __init__(self, config_file: str = "config.yaml"): + """Initialize configuration manager.""" + self.config_file = Path(config_file) + self.config = self.load_config() + + def load_config(self) -> Dict[str, Any]: + """Load configuration from YAML file.""" + if self.config_file.exists(): + with open(self.config_file, 'r') as f: + return yaml.safe_load(f) + else: + return self.get_default_config() + + def get_default_config(self) -> Dict[str, Any]: + """Get default configuration.""" + return { + "logging": { + "level": "INFO", + "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s", + "file": "hdh_deployment.log" + }, + "output": { + "directory": "hdh_results", + "save_plots": True, + "plot_format": "png", + "plot_dpi": 300 + }, + "circuits": { + "max_qubits": 10, + "default_partitions": 3, + "enable_visualization": True, + "save_intermediate": False + }, + "performance": { + "timeout_seconds": 300, + "max_memory_gb": 8, + "parallel_processing": False + } + } + + def save_config(self): + """Save current configuration to file.""" + with open(self.config_file, 'w') as f: + yaml.dump(self.config, f, default_flow_style=False, indent=2) + + def get(self, key_path: str, default=None): + """Get configuration value using dot notation.""" + keys = key_path.split('.') + value = self.config + + for key in keys: + if isinstance(value, dict) and key in value: + value = value[key] + else: + return default + + return value + + def set(self, key_path: str, value: Any): + """Set configuration value using dot notation.""" + keys = key_path.split('.') + config = self.config + + for key in keys[:-1]: + if key not in config: + config[key] = {} + config = config[key] + + config[keys[-1]] = value + + +class ResultsManager: + """Manage and analyze HDH deployment results.""" + + def __init__(self, results_dir: str = "hdh_results"): + """Initialize results manager.""" + self.results_dir = Path(results_dir) + self.results_dir.mkdir(exist_ok=True) + + def save_results(self, results: Dict[str, Any], filename: str = None): + """Save results to JSON file.""" + if filename is None: + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + filename = f"hdh_results_{timestamp}.json" + + filepath = self.results_dir / filename + + with open(filepath, 'w') as f: + json.dump(results, f, indent=2, default=str) + + return filepath + + def load_results(self, filename: str) -> Dict[str, Any]: + """Load results from JSON file.""" + filepath = self.results_dir / filename + + with open(filepath, 'r') as f: + return json.load(f) + + def list_result_files(self) -> List[Path]: + """List all result files in the directory.""" + return list(self.results_dir.glob("*.json")) + + def get_latest_results(self) -> Optional[Dict[str, Any]]: + """Get the most recent results file.""" + result_files = self.list_result_files() + + if not result_files: + return None + + # Sort by modification time + latest_file = max(result_files, key=lambda f: f.stat().st_mtime) + return self.load_results(latest_file.name) + + def merge_results(self, result_files: List[str]) -> Dict[str, Any]: + """Merge multiple result files.""" + merged = { + "merged_at": datetime.now().isoformat(), + "source_files": result_files, + "results": [] + } + + for filename in result_files: + try: + results = self.load_results(filename) + merged["results"].append({ + "filename": filename, + "data": results + }) + except Exception as e: + merged["results"].append({ + "filename": filename, + "error": str(e) + }) + + return merged + + def generate_summary_report(self) -> Dict[str, Any]: + """Generate summary report from all results.""" + result_files = self.list_result_files() + + if not result_files: + return {"error": "No result files found"} + + summary = { + "generated_at": datetime.now().isoformat(), + "total_files": len(result_files), + "file_analysis": [] + } + + for result_file in result_files: + try: + results = self.load_results(result_file.name) + + analysis = { + "filename": result_file.name, + "file_size": result_file.stat().st_size, + "modified_at": datetime.fromtimestamp(result_file.stat().st_mtime).isoformat() + } + + # Analyze content if it has expected structure + if isinstance(results, dict): + if "detailed_results" in results: + detailed = results["detailed_results"] + if isinstance(detailed, list): + analysis["circuits_count"] = len(detailed) + analysis["successful_circuits"] = sum(1 for r in detailed if r.get("success", False)) + + if "summary" in results: + summary_data = results["summary"] + if isinstance(summary_data, dict): + analysis["summary_data"] = summary_data + + summary["file_analysis"].append(analysis) + + except Exception as e: + summary["file_analysis"].append({ + "filename": result_file.name, + "error": str(e) + }) + + return summary + + +class PlotManager: + """Enhanced plotting utilities for HDH visualization.""" + + def __init__(self, output_dir: str = "plots", dpi: int = 300): + """Initialize plot manager.""" + self.output_dir = Path(output_dir) + self.output_dir.mkdir(exist_ok=True) + self.dpi = dpi + + # Set matplotlib style + plt.style.use('default') + self.setup_matplotlib() + + def setup_matplotlib(self): + """Configure matplotlib settings.""" + plt.rcParams['figure.figsize'] = (12, 8) + plt.rcParams['font.size'] = 12 + plt.rcParams['axes.labelsize'] = 14 + plt.rcParams['axes.titlesize'] = 16 + plt.rcParams['legend.fontsize'] = 12 + plt.rcParams['xtick.labelsize'] = 10 + plt.rcParams['ytick.labelsize'] = 10 + + def create_comparison_plot(self, data: Dict[str, List[float]], + title: str, xlabel: str, ylabel: str, + filename: str = None) -> Path: + """Create a comparison plot with multiple data series.""" + fig, ax = plt.subplots(figsize=(12, 8)) + + colors = plt.cm.Set1(np.linspace(0, 1, len(data))) + + for (label, values), color in zip(data.items(), colors): + x_values = range(len(values)) + ax.plot(x_values, values, 'o-', label=label, color=color, + linewidth=2, markersize=6, alpha=0.8) + + ax.set_xlabel(xlabel) + ax.set_ylabel(ylabel) + ax.set_title(title) + ax.legend() + ax.grid(True, alpha=0.3) + + if filename is None: + filename = f"comparison_{title.lower().replace(' ', '_')}.png" + + filepath = self.output_dir / filename + plt.savefig(filepath, dpi=self.dpi, bbox_inches='tight') + plt.close() + + return filepath + + def create_histogram(self, data: List[float], title: str, + xlabel: str, ylabel: str = "Frequency", + bins: int = 20, filename: str = None) -> Path: + """Create a histogram plot.""" + fig, ax = plt.subplots(figsize=(10, 6)) + + ax.hist(data, bins=bins, alpha=0.7, color='skyblue', + edgecolor='black', linewidth=1) + + ax.set_xlabel(xlabel) + ax.set_ylabel(ylabel) + ax.set_title(title) + ax.grid(True, alpha=0.3, axis='y') + + # Add statistics + mean_val = np.mean(data) + std_val = np.std(data) + ax.axvline(mean_val, color='red', linestyle='--', + label=f'Mean: {mean_val:.3f}') + ax.axvline(mean_val + std_val, color='orange', linestyle='--', + alpha=0.7, label=f'Ā±1Ļƒ: {std_val:.3f}') + ax.axvline(mean_val - std_val, color='orange', linestyle='--', alpha=0.7) + ax.legend() + + if filename is None: + filename = f"histogram_{title.lower().replace(' ', '_')}.png" + + filepath = self.output_dir / filename + plt.savefig(filepath, dpi=self.dpi, bbox_inches='tight') + plt.close() + + return filepath + + def create_scatter_plot(self, x_data: List[float], y_data: List[float], + title: str, xlabel: str, ylabel: str, + labels: List[str] = None, filename: str = None) -> Path: + """Create a scatter plot with optional labels.""" + fig, ax = plt.subplots(figsize=(10, 8)) + + scatter = ax.scatter(x_data, y_data, alpha=0.6, s=60, + c=range(len(x_data)), cmap='viridis') + + # Add labels if provided + if labels: + for i, label in enumerate(labels): + ax.annotate(label, (x_data[i], y_data[i]), + xytext=(5, 5), textcoords='offset points', + fontsize=8, alpha=0.8) + + ax.set_xlabel(xlabel) + ax.set_ylabel(ylabel) + ax.set_title(title) + ax.grid(True, alpha=0.3) + + # Add colorbar + plt.colorbar(scatter, ax=ax, label='Data Point Index') + + if filename is None: + filename = f"scatter_{title.lower().replace(' ', '_')}.png" + + filepath = self.output_dir / filename + plt.savefig(filepath, dpi=self.dpi, bbox_inches='tight') + plt.close() + + return filepath + + +class PerformanceProfiler: + """Performance profiling utilities for HDH operations.""" + + def __init__(self): + """Initialize performance profiler.""" + self.profiles = {} + self.active_profiles = {} + + def start_profile(self, name: str): + """Start profiling a named operation.""" + self.active_profiles[name] = { + 'start_time': time.perf_counter(), + 'start_memory': self.get_memory_usage() + } + + def end_profile(self, name: str) -> Dict[str, float]: + """End profiling and return metrics.""" + if name not in self.active_profiles: + raise ValueError(f"No active profile named '{name}'") + + profile_data = self.active_profiles.pop(name) + + metrics = { + 'duration': time.perf_counter() - profile_data['start_time'], + 'memory_delta': self.get_memory_usage() - profile_data['start_memory'], + 'timestamp': datetime.now().isoformat() + } + + if name not in self.profiles: + self.profiles[name] = [] + + self.profiles[name].append(metrics) + return metrics + + def get_memory_usage(self) -> float: + """Get current memory usage in MB.""" + try: + import psutil + process = psutil.Process() + return process.memory_info().rss / 1024 / 1024 + except ImportError: + return 0.0 + + def get_profile_summary(self, name: str) -> Dict[str, Any]: + """Get summary statistics for a named profile.""" + if name not in self.profiles: + return {"error": f"No profiles found for '{name}'"} + + profiles = self.profiles[name] + durations = [p['duration'] for p in profiles] + memory_deltas = [p['memory_delta'] for p in profiles] + + return { + 'name': name, + 'count': len(profiles), + 'duration_stats': { + 'mean': np.mean(durations), + 'median': np.median(durations), + 'min': np.min(durations), + 'max': np.max(durations), + 'std': np.std(durations) + }, + 'memory_stats': { + 'mean': np.mean(memory_deltas), + 'median': np.median(memory_deltas), + 'min': np.min(memory_deltas), + 'max': np.max(memory_deltas), + 'std': np.std(memory_deltas) + } + } + + def export_profiles(self, filepath: str): + """Export all profiles to JSON file.""" + export_data = { + 'exported_at': datetime.now().isoformat(), + 'profiles': self.profiles, + 'summaries': {name: self.get_profile_summary(name) + for name in self.profiles.keys()} + } + + with open(filepath, 'w') as f: + json.dump(export_data, f, indent=2, default=str) + + +def setup_logging(log_level: str = "INFO", log_file: str = None) -> logging.Logger: + """Setup standardized logging for HDH deployment.""" + logger = logging.getLogger("hdh_deployment") + logger.setLevel(getattr(logging, log_level.upper())) + + # Clear any existing handlers + logger.handlers.clear() + + # Console handler + console_handler = logging.StreamHandler(sys.stdout) + console_formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + console_handler.setFormatter(console_formatter) + logger.addHandler(console_handler) + + # File handler if specified + if log_file: + file_handler = logging.FileHandler(log_file) + file_formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s' + ) + file_handler.setFormatter(file_formatter) + logger.addHandler(file_handler) + + return logger + + +def validate_hdh_environment() -> Dict[str, Any]: + """Validate that the HDH environment is properly set up.""" + validation_results = { + 'timestamp': datetime.now().isoformat(), + 'valid': True, + 'issues': [], + 'warnings': [] + } + + # Check HDH import + try: + import hdh + validation_results['hdh_version'] = getattr(hdh, '__version__', 'unknown') + except ImportError as e: + validation_results['valid'] = False + validation_results['issues'].append(f"HDH import failed: {str(e)}") + + # Check required dependencies + required_packages = ['qiskit', 'networkx', 'matplotlib', 'numpy'] + + for package in required_packages: + try: + __import__(package) + except ImportError: + validation_results['issues'].append(f"Missing required package: {package}") + validation_results['valid'] = False + + # Check optional dependencies + optional_packages = ['metis', 'psutil', 'rich', 'click'] + + for package in optional_packages: + try: + __import__(package) + except ImportError: + validation_results['warnings'].append(f"Optional package not available: {package}") + + # Check system resources + try: + import psutil + memory_gb = psutil.virtual_memory().total / (1024**3) + if memory_gb < 4: + validation_results['warnings'].append(f"Low system memory: {memory_gb:.1f}GB") + validation_results['system_memory_gb'] = memory_gb + except ImportError: + validation_results['warnings'].append("Cannot check system memory (psutil not available)") + + return validation_results + + +if __name__ == "__main__": + """Utility testing and validation.""" + print("HDH Deployment Utilities") + print("=" * 50) + print("Special thanks to Maria Gragera Garces for the HDH library!") + print() + + # Validate environment + validation = validate_hdh_environment() + print(f"Environment valid: {validation['valid']}") + + if validation['issues']: + print("Issues found:") + for issue in validation['issues']: + print(f" - {issue}") + + if validation['warnings']: + print("Warnings:") + for warning in validation['warnings']: + print(f" - {warning}") + + # Test configuration manager + print("\nTesting ConfigManager...") + config_mgr = ConfigManager() + print(f"Default output directory: {config_mgr.get('output.directory')}") + + # Test results manager + print("\nTesting ResultsManager...") + results_mgr = ResultsManager() + test_results = {"test": "data", "timestamp": datetime.now().isoformat()} + saved_file = results_mgr.save_results(test_results, "test_results.json") + print(f"Test results saved to: {saved_file}") + + print("\nUtilities test completed!") \ No newline at end of file