Coverage for controller / workflow / state_machine.py: 100.00%

1from dataclasses import dataclass, field 

2from typing import Optional, List, TYPE_CHECKING, Callable 

3from enum import Enum, auto 

4from typing import Dict 

5 

6from controller.path_protocol import PathLike 

7from view.output_format import OutputFormat 

8from view.merge_mode import MergeMode 

9from domain.core.output_handler import OutputHandler 

10from domain.model.file import File 

11 

12@dataclass 

13class WorkflowContext: 

14 input_path: Optional['PathLike'] = None 

15 format_choice: Optional['OutputFormat'] = None 

16 merge_mode: Optional['MergeMode'] = None 

17 files: List['File'] = field(default_factory=list) 

18 handler: Optional['OutputHandler'] = None 

19 merged_filename: Optional[str] = None 

20 error_message: Optional[str] = None 

21 error_origin: Optional['WorkflowState'] = None 

22 

23class WorkflowState(Enum): 

24 SOURCE_INPUT = auto() 

25 FORMAT_SELECTION = auto() 

26 MERGE_MODE_SELECTION = auto() 

27 FILES_SELECTION = auto() 

28 PROCESSING = auto() 

29 COMPLETE = auto() 

30 ERROR = auto() 

31 

32 @property 

33 def display_name(self) -> str: 

34 return { 

35 WorkflowState.SOURCE_INPUT: "source", 

36 WorkflowState.FORMAT_SELECTION: "format", 

37 WorkflowState.MERGE_MODE_SELECTION: "merge mode", 

38 WorkflowState.FILES_SELECTION: "files", 

39 WorkflowState.PROCESSING: "processing", 

40 WorkflowState.COMPLETE: "complete", 

41 WorkflowState.ERROR: "error", 

42 }[self] 

43 

44@dataclass(frozen=True) 

45class StateTransition: 

46 next: Optional['WorkflowState'] 

47 back: Optional['WorkflowState'] 

48 

49WORKFLOW_TRANSITIONS: Dict[WorkflowState, StateTransition] = { 

50 WorkflowState.SOURCE_INPUT: StateTransition(WorkflowState.FORMAT_SELECTION, None), 

51 WorkflowState.FORMAT_SELECTION: StateTransition(WorkflowState.MERGE_MODE_SELECTION, WorkflowState.SOURCE_INPUT), 

52 WorkflowState.MERGE_MODE_SELECTION: StateTransition(WorkflowState.FILES_SELECTION, WorkflowState.FORMAT_SELECTION), 

53 WorkflowState.FILES_SELECTION: StateTransition(WorkflowState.PROCESSING, WorkflowState.MERGE_MODE_SELECTION), 

54 WorkflowState.PROCESSING: StateTransition(WorkflowState.COMPLETE, None), 

55 WorkflowState.COMPLETE: StateTransition(WorkflowState.SOURCE_INPUT, None), 

56 WorkflowState.ERROR: StateTransition(WorkflowState.SOURCE_INPUT, None), 

57} 

58 

59class ConversionWorkflow: 

60 def __init__(self, initial_state: WorkflowState = WorkflowState.SOURCE_INPUT, on_state_change: Optional[Callable[[], None]] = None): 

61 self.state = initial_state 

62 self.state_stack: List[WorkflowState] = [] 

63 self.context = WorkflowContext() 

64 self._on_state_change = on_state_change 

65 

66 def next(self) -> None: 

67 transition = WORKFLOW_TRANSITIONS[self.state] 

68 if transition.next is not None: 

69 self.state_stack.append(self.state) 

70 self.state = transition.next 

71 if self._on_state_change: 

72 self._on_state_change() 

73 

74 def back(self) -> None: 

75 if self.state_stack: 

76 self.state = self.state_stack.pop() 

77 if self._on_state_change: 

78 self._on_state_change() 

79 

80 def can_go_back(self) -> bool: 

81 return bool(self.state_stack) 

82 

83 def reset(self): 

84 self.state = WorkflowState.SOURCE_INPUT 

85 self.state_stack.clear() 

86 self.context = WorkflowContext() 

87 if self._on_state_change: 

88 self._on_state_change() 

89 

90 def get_state(self) -> WorkflowState: 

91 return self.state 

92 

93 def get_history(self) -> List[WorkflowState]: 

94 return self.state_stack