sp-hydra-veil-core/core/infrastructure/plumbing.py

from core.infrastructure.thread_dispatch import MainThreadDispatcher
from core.infrastructure.error_popup import PopupProtocol
from core.infrastructure.service_registry import ServiceLocator

from core.models.Result import Result, ResultError
from core.infrastructure.action_registry import ENUM_MAP
from core.errors.logger import logger

# from infrastructure.observer_service import notify


# generic:
from concurrent.futures import ThreadPoolExecutor, Future
from typing import Callable, Any, Optional, Type
import inspect
from functools import partial

MAX_ATTEMPTS = 3
executor = ThreadPoolExecutor(max_workers=1)

class ThreadPlumbing:
    """
    Manage Execution of functions on threads, and orchestrate their retries,
    then pass the data through to popup prompts for the user to decide how to solve any error, by enum type.
    """
    def __init__(self, action: Callable, ui_dispatcher: MainThreadDispatcher, error_implementation: Type[PopupProtocol], auto_navigate_dict: dict, ui_caller: Optional[Callable] = None, *args, **kwargs):
        self.action = action
        self.ui_dispatcher = ui_dispatcher
        self.ui_caller = ui_caller
        self.error_implementation = error_implementation
        self.auto_navigate_dict = auto_navigate_dict
        self.navigator = ServiceLocator.get("navigator")
        self.args = args
        self.kwargs = kwargs
        self.attempt_count = 0
    
    def execute(self):
        """Entry point: kick off the first attempt"""
        self.attempt_count += 1
        self._spawn_action(action=self.action, first_time=True)
    
    def _spawn_action(self, action: Callable, first_time: bool = None):
        """Spawn action on executor"""
        if first_time:
            future = executor.submit(action, *self.args, **self.kwargs) # original args/kwargs
        else:
            future = executor.submit(action) # this is a pre-loaded partial
        future.add_done_callback(self._on_complete)
    
    def _on_complete(self, future: Future):
        """Handle thread completion"""
        try:
            logger.info("The plumbing module started it's 'on complete' phase..")
            result = future.result()
            if not isinstance(result, Result):
                result = convert_to_result(result)
            if not result.valid:
                result.function_choice = resolve_prescription(result, *self.args, **self.kwargs)
        except Exception as e:
            error_msg = f"There was an error: {str(e)}"
            # notify(error_msg)
            logger.error(error_msg)
            result = Result(valid=False, error_type=ResultError.UNKNOWN, message=str(e))
        
        # Back to main thread
        print(f"result in plumbing is {result}")
        self.ui_dispatcher.run_on_main_thread(self._on_result_ready, result)
    
    def _on_result_ready(self, result: Result):
        """Decide: retry, show popup, or final callback"""
        if result.valid or self.attempt_count >= MAX_ATTEMPTS:
            if self.ui_caller:
                print("calling back with result")
                if not result:
                    result = Result(valid=False, error_type=ResultError.UNKNOWN, message="Plumbing is missing a result.")
                self.ui_caller(result)
            return

        # does this have an auto-navigate page?
        if self.navigator and self.auto_navigate_dict:
            self.check_auto_navigation(result)

        # If the result sender explicitly asked us to navigate to a page with a ui callback, then bypass the popup:
        if result.error_type == ResultError.NAVIGATE_TO:
            if self.ui_caller:
                self.ui_caller(result)

        # notify(f"Error prompt {self.attempt_count} of {MAX_ATTEMPTS}")

        # Show popup and wait for user choice
        choice = bool(result.function_choice)
        print(f"is there a choice? {choice}")
        popup = self.error_implementation(
            choice=bool(result.function_choice),
            title=f"Error: {result.error_type.value}",
            message=result.message,
            on_answer=self._on_popup_answer,
            passed_result=result,
            end_app=False
        )
        # framework neutral show method
        popup.show()

    def _on_popup_answer(self, user_choice: bool, passed_result: Result):
        """User responded to popup"""
        if user_choice and passed_result.function_choice:
            self.attempt_count += 1
            self._spawn_action(passed_result.function_choice)
        else:
            if self.ui_caller:
                self.ui_caller(passed_result)

    def check_auto_navigation(self, result: Result):
        auto_navigate_page = self.auto_navigate_dict.get(result.error_type)
        if auto_navigate_page is None:
            return

        self.navigator.request_navigation.emit(auto_navigate_page)

def normalize_seq(s):
    """Ensures input is a tuple regardless of if it was a list or tuple."""
    return tuple(s) if s is not None else ()

def resolve_prescription(result: Result, *args, **kwargs):
    """
    Find the correct solution to the type of enum problem,
    And then add the args from either the original caller's args,
    or the defaults for that error type
    """

    # Priority 1: If the function explicitly provided a full partial, use it.
    if result.function_choice:
        return result.function_choice

    # Priority 2: Look up the Enum map
    if result.error_type not in ENUM_MAP:
        return None  # No solution

    enum_value = ENUM_MAP[result.error_type]
    
    # Handle the None case (no solution)
    if enum_value is None:
        return None

    # Unpack the enum value - could be just a function or (function, default_args)
    if isinstance(enum_value, tuple):
        func, default_args = enum_value
    else:
        func = enum_value
        default_args = None

    # What does this function require?
    sig = inspect.signature(func)
    function_requirements = sig.parameters

    # requires nothing, we're done, return it.
    if not function_requirements:
        return func

    # Priority 3: Use original args/kwargs
    try:
        bound = sig.bind(*args, **kwargs)
        bound.apply_defaults()
        return partial(func, **bound.arguments)
    except TypeError:
        pass  # Not enough, move to priority 4

    # Priority 4: Use original args/kwargs PLUS defaults
    if default_args is None:
        return False

    try:
        if isinstance(default_args, dict):
            # original kwargs take precedence over default_args
            merged_kwargs = {**default_args, **kwargs}
            bound = sig.bind(**merged_kwargs)
        else:
            # 1. Handle Type Mismatch: Normalize both to tuples
            d_tuple = normalize_seq(default_args)
            o_tuple = normalize_seq(args)
            
            # 2. Positional Override: Replace defaults with original args by index
            merged_args = list(d_tuple)
            for i, val in enumerate(o_tuple):
                if i < len(merged_args):
                    merged_args[i] = val # Override default at this position
                else:
                    merged_args.append(val) # Append if original args exceed defaults
            
            # 3. Collision Fix: Convert positional merge to a name-map first.
            # This prevents "multiple values for argument" if the same arg 
            # is present in both merged_args and kwargs.
            pos_bound = sig.bind(*merged_args)
            
            # Now merge the positional map with kwargs (kwargs win)
            final_kwargs = {**pos_bound.arguments, **kwargs}
            bound = sig.bind(**final_kwargs)
        
        bound.apply_defaults()
        return partial(func, **bound.arguments)
    except TypeError:
        pass

    return False  # Still args or kwargs lacking


def convert_to_result(result: Any) -> Result:
    """Turn a non-Result object into a Result object"""
    error_msg = "Internal error where the result was returned in the wrong format."
    logger.error(error_msg)

    if isinstance(result, bool):
        return Result(valid=result, error_type=ResultError.INVALID_INPUT, message=error_msg)
    
    elif isinstance(result, dict):
        valid = result.get('valid', False)
        message = result.get('message', False)
        if not message:
            message = error_msg
        return Result(valid=valid, error_type=ResultError.INVALID_INPUT, message=error_msg)

    else:
        return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)


# Optional Standalone Public API
def take_plumbing_action(action: Callable, ui_dispatcher: MainThreadDispatcher, error_implementation: Type[PopupProtocol], ui_caller: Callable, *args, **kwargs):
    controller = ThreadPlumbing(action, ui_dispatcher, ui_caller, *args, **kwargs)
    controller.execute()