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], ENUM_MAP: dict, 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.ENUM_MAP = ENUM_MAP 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: if not result.function_choice: result.function_choice = resolve_prescription(result, self.ENUM_MAP, *self.args, **self.kwargs) else: logger.info("Using the pre-existing function choice for this error.") 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, ENUM_MAP: dict, *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 # logger.info(f"self.ENUM_MAP is {self.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()