88########################################################################### 
99"""Module with `Node` sub class for workchain processes.""" 
1010
11- from  typing  import  Optional , Tuple 
11+ from  typing  import  TYPE_CHECKING ,  Optional , Tuple 
1212
13+ from  aiida .common  import  exceptions 
1314from  aiida .common .lang  import  classproperty 
1415
1516from  .workflow  import  WorkflowNode 
1617
18+ if  TYPE_CHECKING :
19+     from  aiida .tools .workflows  import  WorkflowTools 
20+ 
1721__all__  =  ('WorkChainNode' ,)
1822
1923
@@ -22,6 +26,40 @@ class WorkChainNode(WorkflowNode):
2226
2327    STEPPER_STATE_INFO_KEY  =  'stepper_state_info' 
2428
29+     # An optional entry point for a CalculationTools instance 
30+     _tools  =  None 
31+ 
32+     @property  
33+     def  tools (self ) ->  Optional ['WorkflowTools' ]:
34+         """Return the calculation tools that are registered for the process type associated with this calculation. 
35+ 
36+         If the entry point name stored in the `process_type` of the CalcJobNode has an accompanying entry point in the 
37+         `aiida.tools.calculations` entry point category, it will attempt to load the entry point and instantiate it 
38+         passing the node to the constructor. If the entry point does not exist, cannot be resolved or loaded, a warning 
39+         will be logged and the base CalculationTools class will be instantiated and returned. 
40+ 
41+         :return: CalculationTools instance 
42+         """ 
43+         from  aiida .plugins .entry_point  import  get_entry_point_from_string , is_valid_entry_point_string , load_entry_point 
44+         from  aiida .tools .workflows  import  WorkflowTools 
45+ 
46+         if  self ._tools  is  None :
47+             entry_point_string  =  self .process_type 
48+ 
49+             if  entry_point_string  and  is_valid_entry_point_string (entry_point_string ):
50+                 entry_point  =  get_entry_point_from_string (entry_point_string )
51+ 
52+                 try :
53+                     tools_class  =  load_entry_point ('aiida.tools.workflows' , entry_point .name )
54+                     self ._tools  =  tools_class (self )
55+                 except  exceptions .EntryPointError  as  exception :
56+                     self ._tools  =  WorkflowTools (self )
57+                     self .logger .warning (
58+                         f'could not load the workflow tools entry point { entry_point .name }  : { exception }  ' 
59+                     )
60+ 
61+         return  self ._tools 
62+ 
2563    @classproperty  
2664    def  _updatable_attributes (cls ) ->  Tuple [str , ...]:  # type: ignore[override]  # noqa: N805 
2765        return  super ()._updatable_attributes  +  (cls .STEPPER_STATE_INFO_KEY ,)
0 commit comments