"""
Python Bridge to .NET EMDS Engines (NetWeaver and CDP)

This module provides a clean Python interface to communicate with the .NET engine CLI wrappers
using subprocess, JSON for commands, and Apache Arrow IPC for efficient data transfer.

Author: EMDS QGIS Port POC
Date: 2025-11-02
"""

import subprocess
import sys
import json
import stat
import base64
import time
import shutil
import tempfile
from pathlib import Path
from typing import Dict, List, Optional, Any
import pandas as pd
try:
    import pyarrow as pa
except ModuleNotFoundError:
    pa = None
from dataclasses import dataclass, asdict

try:
    from emds_qgis_plugin import debug_logger
except Exception:
    debug_logger = None


@dataclass
class EngineRequest:
    """Request structure for engine commands"""
    command: str
    model_file_path: Optional[str] = None
    model_json: Optional[str] = None
    data_arrow_file_path: Optional[str] = None
    data_arrow_base64: Optional[str] = None
    data_json: Optional[List[Dict]] = None
    result_arrow_file_path: Optional[str] = None
    parameters: Optional[Dict[str, Any]] = None


@dataclass
class EngineResponse:
    """Response structure from engine"""
    success: bool
    error_message: Optional[str] = None
    message: Optional[str] = None
    result_arrow_file_path: Optional[str] = None
    result_arrow_base64: Optional[str] = None
    rows_processed: int = 0
    processing_time_ms: int = 0
    model_info: Optional[Dict[str, Any]] = None
    engine_version: Optional[str] = None
    model_schema_version: Optional[str] = None


class EngineBridge:
    """Base class for communicating with .NET engine CLI wrappers"""

    def __init__(self, engine_exe_path: str, engine_name: str = "EMDS Engine"):
        """
        Initialize the engine bridge

        Args:
            engine_exe_path: Path to the .NET engine executable
            engine_name: Human-readable engine name for logging
        """
        self.engine_exe_path = Path(engine_exe_path)
        self.engine_name = engine_name

        # Provenance: metadata from the most recent engine call (docs/provenance/00).
        # Callers read these after process_data / process_with_mapping instead of
        # the response being silently discarded.
        self.last_model_info: Optional[Dict[str, Any]] = None
        self.last_engine_version: str = ""
        self.last_model_schema_version: str = ""
        self.last_processing_time_ms: int = 0

        if not self.engine_exe_path.exists():
            raise FileNotFoundError(
                f"{engine_name} executable not found at: {self.engine_exe_path}"
            )

    @staticmethod
    def _normalize_model_path(model_path: str) -> str:
        return str(Path(model_path).expanduser().absolute())

    @staticmethod
    def _ensure_native_apphost_executable(engine_path: Path) -> None:
        """Restore Unix execute bits lost by QGIS ZIP extraction."""
        if sys.platform == "win32":
            return

        if sys.platform == "darwin":
            # Belt-and-braces at launch time: a freshly downloaded plugin ZIP
            # can leave the engine quarantined, and Gatekeeper then kills it.
            # first_run_setup.repair_engine_payload does this for the whole
            # payload on plugin load; this covers engines resolved from
            # non-packaged (dev) locations too. Failures are ignored.
            try:
                subprocess.run(
                    ["/usr/bin/xattr", "-d", "com.apple.quarantine", str(engine_path)],
                    capture_output=True,
                    timeout=15,
                )
            except Exception:
                pass

        execute_bits = stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
        try:
            current_mode = engine_path.stat().st_mode
            if (current_mode & execute_bits) == execute_bits:
                return
            engine_path.chmod(current_mode | execute_bits)
        except OSError as exc:
            raise PermissionError(
                f"Could not mark EMDS engine executable: {engine_path}"
            ) from exc

    @staticmethod
    def _resolve_dotnet_host() -> Optional[str]:
        dotnet_path = shutil.which("dotnet")
        if dotnet_path:
            return dotnet_path

        candidate_paths = [
            Path(r"C:\Program Files\dotnet\dotnet.exe"),
            Path(r"C:\Program Files (x86)\dotnet\dotnet.exe"),
        ]

        for candidate in candidate_paths:
            if candidate.exists():
                return str(candidate)

        return None

    def _call_engine(self, request: EngineRequest, progress_callback: Optional[callable] = None) -> EngineResponse:
        """
        Call the .NET engine via subprocess with progress streaming

        Args:
            request: Engine request object
            progress_callback: Optional callback(current, total, percentage) for progress updates

        Returns:
            EngineResponse object
        """
        # Convert request to JSON with PascalCase for .NET
        request_dict_raw = {k: v for k, v in asdict(request).items() if v is not None}
        request_dict = {
            'Command': request_dict_raw.get('command'),
            'ModelFilePath': request_dict_raw.get('model_file_path'),
            'ModelJson': request_dict_raw.get('model_json'),
            'DataArrowFilePath': request_dict_raw.get('data_arrow_file_path'),
            'DataArrowBase64': request_dict_raw.get('data_arrow_base64'),
            'DataJson': request_dict_raw.get('data_json'),
            'ResultArrowFilePath': request_dict_raw.get('result_arrow_file_path'),
            'Parameters': request_dict_raw.get('parameters')
        }
        # Remove None values
        request_dict = {k: v for k, v in request_dict.items() if v is not None}
        request_json = json.dumps(request_dict)

        # A framework-dependent managed assembly (*.dll) must be launched through
        # the `dotnet` muxer. A native self-contained apphost bundles its own
        # runtime and MUST be launched directly: this is `NetWeaverCLI.exe` on
        # Windows but an EXTENSIONLESS binary (`NetWeaverCLI`) on macOS/Linux.
        # Prefixing a native apphost with `dotnet` makes the muxer try to load it
        # as managed IL, which fails with BadImageFormatException ("Bad IL
        # format"). So branch on ".dll", not on ".exe" -- the latter would miss
        # the extensionless macOS/Linux apphosts.
        engine_path = Path(self.engine_exe_path)
        launch_via_dotnet = engine_path.suffix.lower() == ".dll"
        dotnet_host = self._resolve_dotnet_host() if launch_via_dotnet else None
        if debug_logger:
            debug_logger.debug("Engine bridge request", {
                "engine_name": self.engine_name,
                "command": request.command,
                "engine_path": str(self.engine_exe_path),
                "launch_via_dotnet": launch_via_dotnet,
                "dotnet_host": dotnet_host or "",
                "dotnet_found_on_path": bool(shutil.which("dotnet")),
                "has_model_path": bool(request.model_file_path),
                "model_file_path": request.model_file_path or "",
                "model_file_exists": Path(request.model_file_path).exists() if request.model_file_path else False,
                "has_arrow_file_payload": bool(request.data_arrow_file_path),
                "has_arrow_payload": bool(request.data_arrow_base64),
                "arrow_payload_length": len(request.data_arrow_base64 or ""),
                "result_arrow_file_path": request.result_arrow_file_path or "",
                "has_json_payload": bool(request.data_json),
            })

        process = None
        try:
            if launch_via_dotnet:
                if not dotnet_host:
                    raise FileNotFoundError(
                        "dotnet executable was not found. Checked PATH and common install locations."
                    )
                launch_command = [dotnet_host, str(self.engine_exe_path)]
            else:
                # Native self-contained apphost (.exe on Windows, extensionless
                # on macOS/Linux): run directly, no system .NET needed.
                self._ensure_native_apphost_executable(engine_path)
                launch_command = [str(self.engine_exe_path)]
            if debug_logger:
                debug_logger.debug("Launching engine process", {
                    "engine_name": self.engine_name,
                    "launch_command": launch_command,
                })

            # Use Popen for streaming stderr (progress updates)
            popen_kwargs = {}
            if sys.platform == 'win32':
                popen_kwargs['creationflags'] = subprocess.CREATE_NO_WINDOW

            process = subprocess.Popen(
                launch_command,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=False,  # Binary mode for stdin/stdout, text for stderr monitoring
                **popen_kwargs
            )

            # Send request
            process.stdin.write(request_json.encode('utf-8'))
            process.stdin.close()

            # Monitor stderr for progress updates
            stdout_data = b''
            stderr_data = b''

            import threading

            # Background thread to read stderr for progress updates
            stderr_lines = []

            def _read_stderr():
                """Read stderr line-by-line in a background thread."""
                try:
                    while True:
                        line = process.stderr.readline()
                        if not line:
                            break
                        stderr_lines.append(line)
                        decoded = line.decode('utf-8', errors='replace').strip()
                        if decoded.startswith('PROGRESS:'):
                            try:
                                parts = decoded.split(':')
                                if len(parts) >= 3:
                                    current_total = parts[1].split('/')
                                    current = int(current_total[0])
                                    total = int(current_total[1])
                                    percentage = int(parts[2])
                                    if progress_callback:
                                        progress_callback(current, total, percentage)
                            except (ValueError, IndexError):
                                pass
                except Exception:
                    pass

            stderr_thread = threading.Thread(target=_read_stderr, daemon=True)
            stderr_thread.start()

            # Wait for stdout (blocking read)
            stdout_data = process.stdout.read()

            # Wait for stderr thread and process to finish
            stderr_thread.join(timeout=10)
            stderr_data = b''.join(stderr_lines)

            # Drain any remaining stderr
            try:
                remaining = process.stderr.read()
                if remaining:
                    stderr_data += remaining
            except Exception:
                pass

            return_code = process.wait(timeout=600)

            # Parse response from stdout
            output = stdout_data.decode('utf-8')

            if return_code != 0 and not output:
                stderr_text = stderr_data.decode('utf-8')
                raise RuntimeError(
                    f"{self.engine_name} failed: {stderr_text}"
                )

            # Extract JSON from output (handle license messages and other non-JSON
            # lines that the .NET engine writes to stdout before/after the response).
            # Use find (not rfind) to locate the FIRST '{' which opens the JSON object,
            # then raw_decode so trailing text after the closing '}' is silently ignored.
            json_start = output.find('{')
            if json_start == -1:
                raise RuntimeError(f"No JSON response found in output:\n{output}")

            try:
                response_dict, _ = json.JSONDecoder().raw_decode(output, json_start)
            except json.JSONDecodeError as exc:
                raise RuntimeError(
                    f"Failed to parse {self.engine_name} response: {exc}\n"
                    f"Raw output: {output}"
                )

            # Convert PascalCase from .NET to snake_case for Python
            response_dict_snake = {
                'success': response_dict.get('Success', False),
                'error_message': response_dict.get('ErrorMessage'),
                'message': response_dict.get('Message'),
                'result_arrow_file_path': response_dict.get('ResultArrowFilePath'),
                'result_arrow_base64': response_dict.get('ResultArrowBase64'),
                'rows_processed': response_dict.get('RowsProcessed', 0),
                'processing_time_ms': response_dict.get('ProcessingTimeMs', 0),
                'model_info': response_dict.get('ModelInfo'),
                'engine_version': response_dict.get('EngineVersion'),
                'model_schema_version': response_dict.get('ModelSchemaVersion'),
            }
            if debug_logger:
                debug_logger.debug("Engine bridge response", {
                    "engine_name": self.engine_name,
                    "success": response_dict_snake['success'],
                    "error_message": response_dict_snake['error_message'],
                    "rows_processed": response_dict_snake['rows_processed'],
                    "processing_time_ms": response_dict_snake['processing_time_ms'],
                    "result_arrow_file_path": response_dict_snake['result_arrow_file_path'] or "",
                    "message": response_dict_snake['message'],
                })

            engine_response = EngineResponse(**response_dict_snake)
            # Provenance: retain the metadata instead of discarding it so the
            # capture layer can record which engine/model produced a result.
            self.last_model_info = engine_response.model_info
            self.last_engine_version = engine_response.engine_version or ""
            self.last_model_schema_version = engine_response.model_schema_version or ""
            self.last_processing_time_ms = engine_response.processing_time_ms or 0
            return engine_response

        except subprocess.TimeoutExpired:
            process.kill()
            raise TimeoutError(f"{self.engine_name} execution timed out (10 minutes)")
        except FileNotFoundError as e:
            if debug_logger:
                debug_logger.error("Engine launch failed", {
                    "engine_name": self.engine_name,
                    "engine_path": str(self.engine_exe_path),
                    "resolved_dotnet_host": dotnet_host or "",
                    "exception": str(e),
                })
            raise RuntimeError(f"{self.engine_name} launch error: {str(e)}")
        except json.JSONDecodeError as e:
            raise RuntimeError(
                f"Failed to parse {self.engine_name} response: {e}\n"
                f"Raw output: {output}"
            )
        except Exception as e:
            raise RuntimeError(f"{self.engine_name} error: {str(e)}")
        finally:
            if process is not None:
                for stream in (process.stdin, process.stdout, process.stderr):
                    if stream is None:
                        continue
                    try:
                        stream.close()
                    except Exception:
                        pass

    def _require_pyarrow(self):
        """Return pyarrow or raise a clear QGIS-runtime dependency error."""
        if pa is None:
            raise RuntimeError(
                "The EMDS engine bridge requires the Python package 'pyarrow'. "
                "Install pyarrow into the QGIS Python environment, then restart QGIS."
            )
        return pa

    def _dataframe_to_arrow_base64(self, df: pd.DataFrame) -> str:
        """
        Convert pandas DataFrame to Arrow IPC format (base64 encoded)

        Args:
            df: Input DataFrame

        Returns:
            Base64 encoded Arrow IPC stream
        """
        pa_module = self._require_pyarrow()

        # Convert DataFrame to Arrow Table
        table = pa_module.Table.from_pandas(df)

        # Serialize to Arrow IPC stream format
        sink = pa_module.BufferOutputStream()
        with pa_module.ipc.new_stream(sink, table.schema) as writer:
            writer.write_table(table)

        # Get bytes and encode as base64
        arrow_bytes = sink.getvalue().to_pybytes()
        return base64.b64encode(arrow_bytes).decode('ascii')

    def _dataframe_to_arrow_file(self, df: pd.DataFrame, file_path: str) -> None:
        """
        Serialize pandas DataFrame to an Arrow IPC stream file.
        """
        pa_module = self._require_pyarrow()
        table = pa_module.Table.from_pandas(df)
        with pa_module.OSFile(file_path, 'wb') as sink:
            with pa_module.ipc.new_stream(sink, table.schema) as writer:
                writer.write_table(table)

    def _arrow_base64_to_dataframe(self, arrow_base64: str) -> pd.DataFrame:
        """
        Convert base64 encoded Arrow IPC to pandas DataFrame

        Args:
            arrow_base64: Base64 encoded Arrow IPC stream

        Returns:
            pandas DataFrame
        """
        pa_module = self._require_pyarrow()

        # Decode base64
        arrow_bytes = base64.b64decode(arrow_base64)

        # Read Arrow IPC stream
        reader = pa_module.ipc.open_stream(arrow_bytes)
        table = reader.read_all()

        # Convert to pandas
        return table.to_pandas()

    def _materialize_model_blob(self, model_blob: bytes, source_format: str = "") -> str:
        """Write embedded model bytes to a temp file (provenance materialize shim).

        Lets a caller run the exact stored model snapshot without depending on
        the original file still existing/unchanged on disk. The engine still
        loads from a path, so its file-based contract is unchanged (additive).
        """
        if source_format and not source_format.startswith("."):
            source_format = f".{source_format}"
        suffix = source_format or ".model"
        tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
        try:
            tmp.write(model_blob)
        finally:
            tmp.close()
        return tmp.name

    def get_model_summary(self, model_path: str) -> Dict[str, Any]:
        """Return the deterministic model summary (contract section 4) + engine version.

        Returns a dict with keys: ``model_summary`` (the structural summary),
        ``engine_version`` and ``model_schema_version``. Used by the snapshot
        service when first capturing a model.
        """
        request = EngineRequest(
            command="get_model_summary",
            model_file_path=self._normalize_model_path(model_path),
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get model summary: {response.error_message}")
        info = response.model_info or {}
        summary = info.get("ModelSummary") or info.get("model_summary") or {}
        return {
            "model_summary": summary,
            "engine_version": response.engine_version or "",
            "model_schema_version": response.model_schema_version or "",
        }

    def get_editable_weights(self, model_path: str) -> Dict[str, Any]:
        """Return the CDP model's editable weight edges in modeling-space ids.

        Drives the Scenario Sandbox weight grid. ``blocks`` carries
        id/name/level/is_lowest/parent_id per criterion; ``weights`` carries
        parent_id/child_id/pvector per edge. Ids are CdpEditor modeling-space
        block ids — pass the same ids straight back to :meth:`set_weights`
        (no engine-ordinal translation). CDP engine only.
        """
        request = EngineRequest(
            command="get_editable_weights",
            model_file_path=self._normalize_model_path(model_path),
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get editable weights: {response.error_message}")
        info = response.model_info or {}
        return {
            "name": info.get("Name") or info.get("name") or "",
            "blocks": info.get("Blocks") or info.get("blocks") or [],
            "weights": info.get("Weights") or info.get("weights") or [],
            # Fidelity warning: EngineCdpxExporter does not round-trip these, so
            # editing+saving such a model loses them. The UI warns when set.
            "has_custom_value_functions": bool(
                info.get("HasCustomValueFunctions") or info.get("has_custom_value_functions")
            ),
            "has_uncertainty": bool(info.get("HasUncertainty") or info.get("has_uncertainty")),
            "engine_version": response.engine_version or "",
        }

    def set_weights(
        self,
        model_path: str,
        edits: List[Dict[str, Any]],
        output_path: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Apply lowest-criteria weight edits; return the edited, engine-loadable model.

        ``edits`` is a list of ``{"parent_id", "child_id", "weight"}`` dicts using
        the modeling-space ids from :meth:`get_editable_weights`. Each weight
        (0..1) is set on the parent→child edge and the siblings are renormalized
        so the parent's children sum to 1 (the CDP-correct adjustment). The
        original file is never touched — the edit is written to a fresh .cdpx.

        Returns ``model_bytes`` (the edited .cdpx, ready to snapshot),
        ``output_path``, post-edit ``model_summary`` (deterministic), the
        ``applied_edits`` echo and ``engine_version``. CDP engine only.
        """
        if not edits:
            raise ValueError(
                "set_weights requires at least one {parent_id, child_id, weight} edit"
            )
        params: Dict[str, Any] = {"weights": edits}
        if output_path:
            params["output_path"] = str(Path(output_path).expanduser().absolute())
        request = EngineRequest(
            command="set_weights",
            model_file_path=self._normalize_model_path(model_path),
            parameters=params,
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to set weights: {response.error_message}")
        info = response.model_info or {}
        model_b64 = info.get("ModelBase64") or info.get("model_base64") or ""
        model_bytes = base64.b64decode(model_b64) if model_b64 else b""
        return {
            "model_bytes": model_bytes,
            "output_path": info.get("OutputPath") or info.get("output_path") or "",
            "model_summary": info.get("ModelSummary") or info.get("model_summary") or {},
            "applied_edits": info.get("AppliedEdits") or info.get("applied_edits") or [],
            "engine_version": response.engine_version or "",
        }

    def get_editable_nodes(self, model_path: str) -> Dict[str, Any]:
        """Return the NetWeaver model's toggleable nodes + current persistent-ignore.

        Drives the Scenario Sandbox ignore-tree. Each node is
        ``{name, type, persistent_ignore, conditional_ignore}``; identity is by
        ``name`` (unique per KB) — pass names straight back to
        :meth:`set_node_ignores`. NetWeaver engine only.
        """
        request = EngineRequest(
            command="get_editable_nodes",
            model_file_path=self._normalize_model_path(model_path),
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get editable nodes: {response.error_message}")
        info = response.model_info or {}
        return {
            "name": info.get("Name") or info.get("name") or "",
            "nodes": info.get("Nodes") or info.get("nodes") or [],
            "engine_version": response.engine_version or "",
        }

    def set_node_ignores(
        self,
        model_path: str,
        ignore_nodes: Optional[List[str]] = None,
        clear_nodes: Optional[List[str]] = None,
        output_path: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Set/clear persistent-ignore on named nodes; return the edited model.

        ``ignore_nodes`` get ``PersistentIgnore = True`` (the serialized flag —
        not the runtime ``.Ignore``), ``clear_nodes`` get it cleared. The edited
        model is written to a fresh ``.nw2`` (the original is untouched). Returns
        ``model_bytes`` (ready to snapshot), ``output_path``, post-edit
        ``model_summary``, ``changed_count``, ``missing_nodes`` (names not found)
        and ``engine_version``. NetWeaver engine only.
        """
        ignore_nodes = list(ignore_nodes or [])
        clear_nodes = list(clear_nodes or [])
        if not ignore_nodes and not clear_nodes:
            raise ValueError("set_node_ignores requires at least one node to ignore or clear")
        params: Dict[str, Any] = {"ignore_nodes": ignore_nodes, "clear_nodes": clear_nodes}
        if output_path:
            params["output_path"] = str(Path(output_path).expanduser().absolute())
        request = EngineRequest(
            command="set_node_ignore",
            model_file_path=self._normalize_model_path(model_path),
            parameters=params,
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to set node ignore flags: {response.error_message}")
        info = response.model_info or {}
        model_b64 = info.get("ModelBase64") or info.get("model_base64") or ""
        model_bytes = base64.b64decode(model_b64) if model_b64 else b""
        return {
            "model_bytes": model_bytes,
            "output_path": info.get("OutputPath") or info.get("output_path") or "",
            "model_summary": info.get("ModelSummary") or info.get("model_summary") or {},
            "changed_count": info.get("ChangedCount") or info.get("changed_count") or 0,
            "missing_nodes": info.get("MissingNodes") or info.get("missing_nodes") or [],
            "engine_version": response.engine_version or "",
        }

    def process_data(
        self,
        data: pd.DataFrame,
        model_path: Optional[str] = None,
        model_json: Optional[str] = None,
        parameters: Optional[Dict[str, Any]] = None,
        use_arrow: bool = True,
        progress_callback: Optional[callable] = None,
        model_blob: Optional[bytes] = None,
        model_source_format: str = "",
    ) -> pd.DataFrame:
        """
        Process data through the engine

        Args:
            data: Input data as pandas DataFrame
            model_path: Path to model file (NetWeaver .nw2 or CDP .cdpx)
            model_json: Model as JSON string (alternative to model_path)
            parameters: Additional engine parameters
            use_arrow: Use Arrow IPC for data transfer (faster for large datasets)
            progress_callback: Optional callback(current, total, percentage) for progress updates

        Returns:
            Result data as pandas DataFrame
        """
        start_time = time.time()

        # Prepare request
        request = EngineRequest(command="process", parameters=parameters)
        if debug_logger:
            debug_logger.debug("Preparing engine process request", {
                "engine_name": self.engine_name,
                "model_path": model_path or "",
                "rows": len(data.index),
                "columns": list(data.columns),
            })

        materialized_model_path = None
        if model_blob is not None:
            materialized_model_path = self._materialize_model_blob(model_blob, model_source_format)
            request.model_file_path = materialized_model_path
        elif model_path:
            request.model_file_path = self._normalize_model_path(model_path)
        elif model_json:
            request.model_json = model_json

        # Convert data -- prefer file-based Arrow IPC to avoid base64 overhead.
        tmp_input = None
        tmp_output = None
        try:
            if use_arrow:
                tmp_input = tempfile.NamedTemporaryFile(suffix=".arrow", delete=False)
                tmp_input.close()
                tmp_output = tempfile.NamedTemporaryFile(suffix=".arrow", delete=False)
                tmp_output.close()
                self._dataframe_to_arrow_file(data, tmp_input.name)
                request.data_arrow_file_path = tmp_input.name
                request.result_arrow_file_path = tmp_output.name
            else:
                request.data_json = data.to_dict('records')

            # Call engine with progress callback
            response = self._call_engine(request, progress_callback=progress_callback)

            if not response.success:
                raise RuntimeError(
                    f"{self.engine_name} processing failed: {response.error_message}"
                )

            # Convert result -- prefer file-based path, fall back to base64
            if response.result_arrow_file_path and Path(response.result_arrow_file_path).exists():
                arrow_bytes = Path(response.result_arrow_file_path).read_bytes()
                reader = self._require_pyarrow().ipc.open_stream(arrow_bytes)
                result_df = reader.read_all().to_pandas()
            elif response.result_arrow_base64:
                result_df = self._arrow_base64_to_dataframe(response.result_arrow_base64)
            else:
                raise RuntimeError("No result data returned from engine")

            elapsed_ms = (time.time() - start_time) * 1000
            if debug_logger:
                debug_logger.debug("Engine processing complete", {
                    "engine_name": self.engine_name,
                    "rows_processed": response.rows_processed,
                    "processing_time_ms": response.processing_time_ms,
                    "transfer_overhead_ms": max(0, round(elapsed_ms - response.processing_time_ms)),
                })

            return result_df
        finally:
            # Clean up temp Arrow files
            import os
            for tmp in (tmp_input, tmp_output):
                if tmp is not None:
                    try:
                        os.unlink(tmp.name)
                    except OSError:
                        pass
            if materialized_model_path:
                try:
                    os.unlink(materialized_model_path)
                except OSError:
                    pass

    def validate_model(self, model_path: str) -> bool:
        """
        Validate a model file

        Args:
            model_path: Path to model file

        Returns:
            True if model is valid
        """
        request = EngineRequest(
            command="validate_model",
            model_file_path=self._normalize_model_path(model_path)
        )

        response = self._call_engine(request)

        if not response.success:
            raise RuntimeError(f"Model validation failed: {response.error_message}")

        return True

    def get_model_info(self, model_path: str) -> Dict[str, Any]:
        """
        Get information about a model

        Args:
            model_path: Path to model file

        Returns:
            Dictionary with model information
        """
        request = EngineRequest(
            command="get_model_info",
            model_file_path=self._normalize_model_path(model_path)
        )

        response = self._call_engine(request)

        if not response.success:
            raise RuntimeError(f"Failed to get model info: {response.error_message}")

        return response.model_info if response.model_info else {}

    def get_topic_tree(self, model_path: str) -> List[Dict[str, Any]]:
        """
        Get the model topic hierarchy as a flat tree list with parent-child IDs.
        """
        request = EngineRequest(
            command="get_topic_tree",
            model_file_path=self._normalize_model_path(model_path)
        )

        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get topic tree: {response.error_message}")

        model_info = response.model_info if response.model_info else {}
        return model_info.get("TopicTree", [])

    def get_topic_variables(self, model_path: str, topics: List[str]) -> Dict[str, Any]:
        """
        Get variables for the selected topics plus all child topics, deduplicated.
        """
        request = EngineRequest(
            command="get_topic_variables",
            model_file_path=self._normalize_model_path(model_path),
            parameters={"topics": topics}
        )

        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get topic variables: {response.error_message}")

        return response.model_info if response.model_info else {}


class NetWeaverBridge(EngineBridge):
    """Python bridge to NetWeaver .NET engine"""

    def __init__(self, netweaver_cli_path: str):
        """
        Initialize NetWeaver bridge

        Args:
            netweaver_cli_path: Path to NetWeaverCLI.dll
        """
        super().__init__(netweaver_cli_path, "NetWeaver")

    @staticmethod
    def _json_safe_cell(value: Any) -> Any:
        """Coerce a row cell (possibly a numpy scalar / NaN) to a JSON-safe
        primitive. NaN becomes None so the engine leaves the variable
        unevaluated instead of receiving a bogus number."""
        if value is None:
            return None
        if isinstance(value, bool):
            return value
        if isinstance(value, str):
            return value
        try:
            numeric = float(value)
        except (TypeError, ValueError):
            return str(value)
        return None if numeric != numeric else numeric

    def get_node_network(self, model_path: str, topic_name: str) -> Dict[str, Any]:
        """Return one topic's full dependency subtree (parity plan Q1-1).

        The result is a flat occurrence list — ``nodes[i]["id"] == i`` — with
        ``name`` / ``type`` / ``label`` (operator glyphs like ``OR``, ``MIN``,
        ``xy``) / ``hint`` / ``children`` (ids) / ``connection`` (``normal`` |
        ``conditional`` | ``persistent``, the edge INTO the node) / ``db_field``
        (the KB's column hint for variable nodes). Topology only — cache it per
        (model mtime, topic) and color it with :meth:`evaluate_row`.
        """
        request = EngineRequest(
            command="get_node_network",
            model_file_path=self._normalize_model_path(model_path),
            parameters={"topic_name": topic_name},
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get node network: {response.error_message}")
        info = response.model_info or {}
        return {
            "topic": info.get("Topic") or info.get("topic") or topic_name,
            "nodes": info.get("Nodes") or info.get("nodes") or [],
            "truncated": bool(info.get("Truncated") or info.get("truncated")),
            "engine_version": response.engine_version or "",
        }

    def get_documents(
        self, model_path: str, topic_name: Optional[str] = None
    ) -> Dict[str, Any]:
        """The KB's authored documentation (parity plan Q6): KB-level doc
        items + hyperlink, and every named node carrying doc items, a
        hyperlink, or a hint. Public docs only. ``topic_name`` scopes nodes
        to that topic's subtree. Doc items are ``{key, title, text,
        has_rtf}`` — the plain-text body is the cross-product floor."""
        params: Dict[str, Any] = {}
        if topic_name:
            params["topic_name"] = topic_name
        request = EngineRequest(
            command="get_documents",
            model_file_path=self._normalize_model_path(model_path),
            parameters=params or None,
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get documents: {response.error_message}")
        info = response.model_info or {}
        return {
            "kb": info.get("Kb") or info.get("kb") or {},
            "nodes": info.get("Nodes") or info.get("nodes") or [],
            "topic_filter": info.get("TopicFilter") or info.get("topic_filter"),
            "engine_version": response.engine_version or "",
        }

    def evaluate_row(
        self,
        model_path: str,
        row: Dict[str, Any],
        field_mappings: Optional[Dict[str, str]] = None,
        ignore_overrides: Optional[Dict[str, Dict[str, bool]]] = None,
        topic_name: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Evaluate ONE feature's values through a fresh engine (plan Q1-2).

        ``row`` is the feature's column → value map (pass the whole attribute
        row; unresolvable columns are simply unused). Optional
        ``field_mappings`` (model var → column) win over the KB's own hints;
        ``ignore_overrides`` (``{node: {"conditional": bool, "persistent":
        bool}}``) are stateless per-call what-ifs — nothing is written, so
        resetting is just calling again without them. When ``topic_name`` is
        given, ``topic_nodes`` carries per-occurrence values in the SAME id
        order as :meth:`get_node_network` for coloring anonymous operators.

        Returns ``node_values`` (name → value, None = no data — matches the
        persisted −9999.888777 sentinel rule), ``no_data``, ``ignored_nodes``,
        ``applied_inputs``, ``unresolved_variables``,
        ``missing_override_nodes`` and ``topic_nodes``.
        """
        safe_row = {str(k): self._json_safe_cell(v) for k, v in dict(row).items()}
        params: Dict[str, Any] = {"row": safe_row}
        if field_mappings:
            params["field_mappings"] = {str(k): str(v) for k, v in field_mappings.items()}
        if ignore_overrides:
            params["ignore_overrides"] = {
                str(name): {
                    key: bool(flag)
                    for key, flag in dict(flags).items()
                    if key in ("conditional", "persistent")
                }
                for name, flags in ignore_overrides.items()
            }
        if topic_name:
            params["topic_name"] = topic_name

        request = EngineRequest(
            command="evaluate_row",
            model_file_path=self._normalize_model_path(model_path),
            parameters=params,
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to evaluate row: {response.error_message}")
        info = response.model_info or {}
        return {
            "applied_inputs": int(info.get("AppliedInputs") or info.get("applied_inputs") or 0),
            "unresolved_variables": info.get("UnresolvedVariables")
                or info.get("unresolved_variables") or [],
            # variable -> the row column it was actually read from (Q3-3)
            "resolved_inputs": info.get("ResolvedInputs")
                or info.get("resolved_inputs") or {},
            "node_values": info.get("NodeValues") or info.get("node_values") or {},
            "no_data": info.get("NoData") or info.get("no_data") or [],
            "ignored_nodes": info.get("IgnoredNodes") or info.get("ignored_nodes") or [],
            "overrides_applied": int(info.get("OverridesApplied")
                or info.get("overrides_applied") or 0),
            "missing_override_nodes": info.get("MissingOverrideNodes")
                or info.get("missing_override_nodes") or [],
            "topic_nodes": info.get("TopicNodes") or info.get("topic_nodes") or [],
            "truncated": bool(info.get("Truncated") or info.get("truncated")),
            "engine_version": response.engine_version or "",
        }


class CDPBridge(EngineBridge):
    """Python bridge to CDP .NET engine"""

    def __init__(self, cdp_cli_path: str):
        """
        Initialize CDP bridge

        Args:
            cdp_cli_path: Path to CDPCLI.dll
        """
        super().__init__(cdp_cli_path, "CDP")

    def get_mapping_schema(self, model_path: str) -> Dict[str, Any]:
        request = EngineRequest(
            command="get_mapping_schema",
            model_file_path=self._normalize_model_path(model_path)
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get CDP mapping schema: {response.error_message}")
        return response.model_info if response.model_info else {}

    def get_model_graph(self, model_path: str) -> Dict[str, Any]:
        request = EngineRequest(
            command="get_model_graph",
            model_file_path=self._normalize_model_path(model_path)
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get CDP model graph: {response.error_message}")
        return response.model_info if response.model_info else {}

    def get_model_diagram(self, model_path: str) -> Dict[str, Any]:
        request = EngineRequest(
            command="get_model_diagram",
            model_file_path=self._normalize_model_path(model_path)
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to get CDP model diagram: {response.error_message}")
        return response.model_info if response.model_info else {}

    def get_rules(self, model_path: str) -> List[Dict[str, Any]]:
        """Return the list of rules defined inside the CDP model file.

        Each rule is a dict with at least Name and Description keys. If the
        underlying CLI does not yet support `get_rules` (older builds), an
        empty list is returned so callers can fall back to a "No rules
        defined" placeholder.
        """
        request = EngineRequest(
            command="get_rules",
            model_file_path=self._normalize_model_path(model_path),
        )
        response = self._call_engine(request)
        if not response.success:
            return []
        info = response.model_info if response.model_info else {}
        rules = info.get("Rules") or info.get("rules") or []
        return list(rules) if isinstance(rules, list) else []

    def get_weights(self, model_path: str) -> List[Dict[str, Any]]:
        """Return the model's pair-wise / standard weights.

        Each weight entry is a dict, format dictated by the CLI. Returns an
        empty list when the CLI does not implement `get_weights`.
        """
        request = EngineRequest(
            command="get_weights",
            model_file_path=self._normalize_model_path(model_path),
        )
        response = self._call_engine(request)
        if not response.success:
            return []
        info = response.model_info if response.model_info else {}
        weights = info.get("Weights") or info.get("weights") or []
        return list(weights) if isinstance(weights, list) else []

    def _read_arrow_response_dataframe(self, response: EngineResponse) -> pd.DataFrame:
        if response.result_arrow_file_path and Path(response.result_arrow_file_path).exists():
            arrow_bytes = Path(response.result_arrow_file_path).read_bytes()
            reader = self._require_pyarrow().ipc.open_stream(arrow_bytes)
            return reader.read_all().to_pandas()
        if response.result_arrow_base64:
            return self._arrow_base64_to_dataframe(response.result_arrow_base64)
        return pd.DataFrame()

    def _run_table_command(
        self,
        *,
        command: str,
        data: Optional[pd.DataFrame] = None,
        model_path: str,
        parameters: Optional[Dict[str, Any]] = None,
        progress_callback: Optional[callable] = None,
    ) -> Dict[str, Any]:
        request = EngineRequest(
            command=command,
            model_file_path=self._normalize_model_path(model_path),
            parameters=parameters or {},
        )

        tmp_input = None
        tmp_output = None
        try:
            if data is not None:
                tmp_input = tempfile.NamedTemporaryFile(suffix=".arrow", delete=False)
                tmp_input.close()
                self._dataframe_to_arrow_file(data, tmp_input.name)
                request.data_arrow_file_path = tmp_input.name

            tmp_output = tempfile.NamedTemporaryFile(suffix=".arrow", delete=False)
            tmp_output.close()
            request.result_arrow_file_path = tmp_output.name

            response = self._call_engine(request, progress_callback=progress_callback)
            if not response.success:
                raise RuntimeError(f"CDP command '{command}' failed: {response.error_message}")

            return {
                "table": self._read_arrow_response_dataframe(response),
                "model_info": response.model_info if response.model_info else {},
                "message": response.message or "",
                "rows_processed": response.rows_processed,
                "processing_time_ms": response.processing_time_ms,
            }
        finally:
            import os
            for tmp in (tmp_input, tmp_output):
                if tmp is not None:
                    try:
                        os.unlink(tmp.name)
                    except OSError:
                        pass

    @staticmethod
    def _build_mapping_parameters(
        *,
        field_mappings: Optional[list[dict[str, Any]]] = None,
        mapping_rows: Optional[list[dict[str, Any]]] = None,
        display_field: str = "",
        primary_key: str = "",
        input_table_name: str = "",
        validation_settings: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        return {
            "field_mappings": field_mappings or [],
            "mapping_rows": mapping_rows or [],
            "display_field": display_field,
            "primary_key": primary_key,
            "input_table_name": input_table_name,
            "validation_settings": validation_settings or {},
        }

    def process_with_mapping(
        self,
        data: pd.DataFrame,
        model_path: str,
        *,
        field_mappings: Optional[list[dict[str, Any]]] = None,
        mapping_rows: Optional[list[dict[str, Any]]] = None,
        display_field: str = "",
        primary_key: str = "",
        input_table_name: str = "",
        validation_settings: Optional[Dict[str, Any]] = None,
        progress_callback: Optional[callable] = None,
        model_blob: Optional[bytes] = None,
        model_source_format: str = "",
    ) -> pd.DataFrame:
        parameters = self._build_mapping_parameters(
            field_mappings=field_mappings,
            mapping_rows=mapping_rows,
            display_field=display_field,
            primary_key=primary_key,
            input_table_name=input_table_name,
            validation_settings=validation_settings,
        )
        request = EngineRequest(
            command="process_with_mapping",
            model_file_path=self._normalize_model_path(model_path),
            parameters=parameters,
        )
        materialized_model_path = None
        if model_blob is not None:
            materialized_model_path = self._materialize_model_blob(model_blob, model_source_format)
            request.model_file_path = materialized_model_path
        # Use file-based Arrow IPC to avoid base64 overhead
        tmp_input = None
        tmp_output = None
        try:
            tmp_input = tempfile.NamedTemporaryFile(suffix=".arrow", delete=False)
            tmp_input.close()
            tmp_output = tempfile.NamedTemporaryFile(suffix=".arrow", delete=False)
            tmp_output.close()
            self._dataframe_to_arrow_file(data, tmp_input.name)
            request.data_arrow_file_path = tmp_input.name
            request.result_arrow_file_path = tmp_output.name

            response = self._call_engine(request, progress_callback=progress_callback)
            if not response.success:
                raise RuntimeError(f"CDP mapped processing failed: {response.error_message}")

            result_df = self._read_arrow_response_dataframe(response)
            if result_df.empty and not response.result_arrow_base64 and not response.result_arrow_file_path:
                raise RuntimeError("No mapped CDP result data returned from engine")
            return result_df
        finally:
            import os
            for tmp in (tmp_input, tmp_output):
                if tmp is not None:
                    try:
                        os.unlink(tmp.name)
                    except OSError:
                        pass
            if materialized_model_path:
                try:
                    os.unlink(materialized_model_path)
                except OSError:
                    pass

    def normalize_model(self, model_path: str, output_path: str = "") -> Dict[str, Any]:
        request = EngineRequest(
            command="normalize_model",
            model_file_path=self._normalize_model_path(model_path),
            parameters={"output_path": output_path},
        )
        response = self._call_engine(request)
        if not response.success:
            raise RuntimeError(f"Failed to normalize CDP model: {response.error_message}")
        return response.model_info if response.model_info else {}

    def get_lowest_criteria(self, model_path: str) -> Dict[str, Any]:
        return self._run_table_command(
            command="get_lowest_criteria",
            model_path=model_path,
        )

    def get_contributions(
        self,
        data: pd.DataFrame,
        model_path: str,
        *,
        field_mappings: Optional[list[dict[str, Any]]] = None,
        mapping_rows: Optional[list[dict[str, Any]]] = None,
        display_field: str = "",
        primary_key: str = "",
        input_table_name: str = "",
        validation_settings: Optional[Dict[str, Any]] = None,
        source_level_index: int = 1,
        target_hblock_id: int = 0,
    ) -> Dict[str, Any]:
        parameters = self._build_mapping_parameters(
            field_mappings=field_mappings,
            mapping_rows=mapping_rows,
            display_field=display_field,
            primary_key=primary_key,
            input_table_name=input_table_name,
            validation_settings=validation_settings,
        )
        parameters["source_level_index"] = source_level_index
        parameters["target_hblock_id"] = target_hblock_id
        return self._run_table_command(
            command="get_contributions",
            data=data,
            model_path=model_path,
            parameters=parameters,
        )

    def get_sensitivity(
        self,
        data: pd.DataFrame,
        model_path: str,
        *,
        field_mappings: Optional[list[dict[str, Any]]] = None,
        mapping_rows: Optional[list[dict[str, Any]]] = None,
        display_field: str = "",
        primary_key: str = "",
        input_table_name: str = "",
        validation_settings: Optional[Dict[str, Any]] = None,
        focus_alt_rank: int = -1,
        criterion_name: str = "",
    ) -> Dict[str, Any]:
        parameters = self._build_mapping_parameters(
            field_mappings=field_mappings,
            mapping_rows=mapping_rows,
            display_field=display_field,
            primary_key=primary_key,
            input_table_name=input_table_name,
            validation_settings=validation_settings,
        )
        parameters["focus_alt_rank"] = focus_alt_rank
        parameters["criterion_name"] = criterion_name
        return self._run_table_command(
            command="get_sensitivity",
            data=data,
            model_path=model_path,
            parameters=parameters,
        )

    def get_tradeoffs(
        self,
        data: pd.DataFrame,
        model_path: str,
        *,
        field_mappings: Optional[list[dict[str, Any]]] = None,
        mapping_rows: Optional[list[dict[str, Any]]] = None,
        display_field: str = "",
        primary_key: str = "",
        input_table_name: str = "",
        validation_settings: Optional[Dict[str, Any]] = None,
        reference_hblock_id: int = 0,
    ) -> Dict[str, Any]:
        parameters = self._build_mapping_parameters(
            field_mappings=field_mappings,
            mapping_rows=mapping_rows,
            display_field=display_field,
            primary_key=primary_key,
            input_table_name=input_table_name,
            validation_settings=validation_settings,
        )
        parameters["reference_hblock_id"] = reference_hblock_id
        return self._run_table_command(
            command="get_tradeoffs",
            data=data,
            model_path=model_path,
            parameters=parameters,
        )


# Example usage
if __name__ == "__main__":
    # This is a demo - paths would be configured based on deployment
    print("EMDS Engine Bridge - Proof of Concept")
    print("=" * 60)
    print()
    print("This module provides a Python interface to .NET EMDS engines")
    print("using subprocess communication with Apache Arrow IPC for")
    print("efficient data transfer.")
    print()
    print("Key features:")
    print("  - JSON for commands and metadata")
    print("  - Apache Arrow IPC for large datasets (binary, zero-copy)")
    print("  - Fallback to JSON for small datasets")
    print("  - Type-safe request/response with dataclasses")
    print()
    print("Example usage:")
    print()
    print("  # Initialize bridge")
    print("  nw = NetWeaverBridge('/path/to/NetWeaverCLI.dll')")
    print()
    print("  # Process data")
    print("  input_data = pd.DataFrame({")
    print("      'site_id': [1, 2, 3],")
    print("      'temperature': [15.5, 18.2, 12.3],")
    print("      'ph': [7.2, 6.8, 7.5]")
    print("  })")
    print()
    print("  results = nw.process_data(")
    print("      data=input_data,")
    print("      model_path='model.nw',")
    print("      use_arrow=True  # Fast binary transfer")
    print("  )")
