225 lines
8.5 KiB
Python
225 lines
8.5 KiB
Python
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
|
|
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:
|
|
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()
|