<s> from __future__ import absolute_import <EOL> from collections import defaultdict <EOL> from functools import total_ordering <EOL> from os . path import commonprefix <EOL> import re <EOL> from xml . sax . saxutils import unescape <EOL> from corehq . apps . app_manager import id_strings <EOL> from corehq . apps . app_manager . const import ( <EOL> RETURN_TO , ) <EOL> from corehq . apps . app_manager . exceptions import SuiteError , SuiteValidationError <EOL> from corehq . apps . app_manager . suite_xml . contributors import PostProcessor <EOL> from corehq . apps . app_manager . suite_xml . xml_models import StackDatum , Stack , CreateFrame <EOL> from corehq . apps . app_manager . xpath import CaseIDXPath , session_var , XPath <EOL> from dimagi . utils . decorators . memoized import memoized <EOL> class WorkflowHelper ( PostProcessor ) : <EOL> def __init__ ( self , suite , app , modules ) : <EOL> super ( WorkflowHelper , self ) . __init__ ( suite , app , modules ) <EOL> root_modules = [ module for module in self . modules if getattr ( module , '<STR_LIT>' , False ) ] <EOL> self . root_module_datums = [ datum for module in root_modules <EOL> for datum in self . get_module_datums ( u'<STR_LIT>' . format ( module . id ) ) . values ( ) ] <EOL> def update_suite ( self ) : <EOL> """<STR_LIT>""" <EOL> for module in self . modules : <EOL> for form in module . get_suite_forms ( ) : <EOL> form_command = id_strings . form_command ( form , module ) <EOL> if_prefix = None <EOL> stack_frames = [ ] <EOL> case_list_form_frames = CaseListFormWorkflow ( self ) . case_list_forms_frames ( form ) <EOL> stack_frames . extend ( case_list_form_frames ) <EOL> if case_list_form_frames : <EOL> if_prefix = session_var ( RETURN_TO ) . count ( ) . eq ( <NUM_LIT:0> ) <EOL> stack_frames . extend ( EndOfFormNavigationWorkflow ( self ) . form_workflow_frames ( if_prefix , module , form ) ) <EOL> self . create_workflow_stack ( form_command , stack_frames ) <EOL> def get_frame_children ( self , target_form , module_only = False ) : <EOL> """<STR_LIT>""" <EOL> target_form_command = id_strings . form_command ( target_form ) <EOL> target_module_id , target_form_id = target_form_command . split ( '<STR_LIT:->' ) <EOL> module_command = id_strings . menu_id ( target_form . get_module ( ) ) <EOL> module_datums = self . get_module_datums ( target_module_id ) <EOL> form_datums = module_datums [ target_form_id ] <EOL> if module_command == id_strings . ROOT : <EOL> datums_list = self . root_module_datums <EOL> else : <EOL> datums_list = module_datums . values ( ) <EOL> common_datums = commonprefix ( datums_list ) <EOL> remaining_datums = form_datums [ len ( common_datums ) : ] <EOL> frame_children = [ CommandId ( module_command ) ] if module_command != id_strings . ROOT else [ ] <EOL> frame_children . extend ( common_datums ) <EOL> if not module_only : <EOL> frame_children . append ( CommandId ( target_form_command ) ) <EOL> frame_children . extend ( remaining_datums ) <EOL> return frame_children <EOL> def create_workflow_stack ( self , form_command , frame_metas ) : <EOL> frames = filter ( None , [ meta . to_frame ( ) for meta in frame_metas ] ) <EOL> if not frames : <EOL> return <EOL> entry = self . get_form_entry ( form_command ) <EOL> if not entry . stack : <EOL> entry . stack = Stack ( ) <EOL> for frame in frames : <EOL> entry . stack . add_frame ( frame ) <EOL> def get_form_datums ( self , form ) : <EOL> """<STR_LIT>""" <EOL> module_id , form_id = id_strings . form_command ( form ) . split ( '<STR_LIT:->' ) <EOL> return self . get_module_datums ( module_id ) [ form_id ] <EOL> def get_module_datums ( self , module_id ) : <EOL> """<STR_LIT>""" <EOL> _ , datums = self . _get_entries_datums ( ) <EOL> return datums [ module_id ] <EOL> def get_form_entry ( self , form_command ) : <EOL> entries , _ = self . _get_entries_datums ( ) <EOL> return entries [ form_command ] <EOL> @ memoized <EOL> def _get_entries_datums ( self ) : <EOL> datums = defaultdict ( lambda : defaultdict ( list ) ) <EOL> entries = { } <EOL> def _include_datums ( entry ) : <EOL> return not entry . command . id . startswith ( '<STR_LIT>' ) and not entry . command . id . endswith ( '<STR_LIT>' ) <EOL> for e in filter ( _include_datums , self . suite . entries ) : <EOL> command = e . command . id <EOL> module_id , form_id = command . split ( '<STR_LIT:->' , <NUM_LIT:1> ) <EOL> entries [ command ] = e <EOL> if not e . datums : <EOL> datums [ module_id ] [ form_id ] = [ ] <EOL> else : <EOL> for d in e . datums : <EOL> datums [ module_id ] [ form_id ] . append ( WorkflowDatumMeta . from_session_datum ( d ) ) <EOL> return entries , datums <EOL> class EndOfFormNavigationWorkflow ( object ) : <EOL> def __init__ ( self , helper ) : <EOL> self . helper = helper <EOL> def form_workflow_frames ( self , if_prefix , module , form ) : <EOL> """<STR_LIT>""" <EOL> from corehq . apps . app_manager . models import ( <EOL> WORKFLOW_PREVIOUS , WORKFLOW_MODULE , WORKFLOW_ROOT , WORKFLOW_FORM , WORKFLOW_PARENT_MODULE <EOL> ) <EOL> def frame_children_for_module ( module_ , include_user_selections = True ) : <EOL> frame_children = [ ] <EOL> if module_ . root_module : <EOL> frame_children . extend ( frame_children_for_module ( module_ . root_module ) ) <EOL> if include_user_selections : <EOL> this_module_children = self . helper . get_frame_children ( module_ . get_form ( <NUM_LIT:0> ) , module_only = True ) <EOL> for child in this_module_children : <EOL> if child not in frame_children : <EOL> frame_children . append ( child ) <EOL> else : <EOL> module_command = id_strings . menu_id ( module_ ) <EOL> if module_command != id_strings . ROOT : <EOL> frame_children . append ( CommandId ( module_command ) ) <EOL> return frame_children <EOL> stack_frames = [ ] <EOL> if form . post_form_workflow == WORKFLOW_ROOT : <EOL> stack_frames . append ( StackFrameMeta ( if_prefix , None , [ ] , allow_empty_frame = True ) ) <EOL> elif form . post_form_workflow == WORKFLOW_MODULE : <EOL> frame_children = frame_children_for_module ( module , include_user_selections = False ) <EOL> stack_frames . append ( StackFrameMeta ( if_prefix , None , frame_children ) ) <EOL> elif form . post_form_workflow == WORKFLOW_PARENT_MODULE : <EOL> root_module = module . root_module <EOL> frame_children = frame_children_for_module ( root_module ) <EOL> stack_frames . append ( StackFrameMeta ( if_prefix , None , frame_children ) ) <EOL> elif form . post_form_workflow == WORKFLOW_PREVIOUS : <EOL> frame_children = self . helper . get_frame_children ( form ) <EOL> last = frame_children . pop ( ) <EOL> while isinstance ( last , WorkflowDatumMeta ) and not last . requires_selection : <EOL> last = frame_children . pop ( ) <EOL> stack_frames . append ( StackFrameMeta ( if_prefix , None , frame_children ) ) <EOL> elif form . post_form_workflow == WORKFLOW_FORM : <EOL> source_form_datums = self . helper . get_form_datums ( form ) <EOL> for link in form . form_links : <EOL> target_form = self . helper . app . get_form ( link . form_id ) <EOL> target_module = target_form . get_module ( ) <EOL> target_frame_children = self . helper . get_frame_children ( target_form ) <EOL> if link . datums : <EOL> frame_children = EndOfFormNavigationWorkflow . get_datums_matched_to_manual_values ( <EOL> target_frame_children , link . datums <EOL> ) <EOL> else : <EOL> frame_children = EndOfFormNavigationWorkflow . get_datums_matched_to_source ( <EOL> target_frame_children , source_form_datums <EOL> ) <EOL> if target_module in module . get_child_modules ( ) : <EOL> parent_frame_children = self . helper . get_frame_children ( module . get_form ( <NUM_LIT:0> ) , module_only = True ) <EOL> parent_ids = { parent . id for parent in parent_frame_children } <EOL> frame_children = parent_frame_children + [ <EOL> child for child in frame_children <EOL> if child . id not in parent_ids <EOL> ] <EOL> stack_frames . append ( StackFrameMeta ( if_prefix , link . xpath , frame_children ) ) <EOL> return stack_frames <EOL> @ staticmethod <EOL> def get_datums_matched_to_source ( target_frame_elements , source_datums ) : <EOL> """<STR_LIT>""" <EOL> unused_source_datums = list ( source_datums ) <EOL> for target_datum in target_frame_elements : <EOL> if not isinstance ( target_datum , WorkflowDatumMeta ) or not target_datum . requires_selection : <EOL> yield target_datum <EOL> else : <EOL> match = EndOfFormNavigationWorkflow . find_best_match ( target_datum , unused_source_datums ) <EOL> if match : <EOL> unused_source_datums = [ datum for datum in unused_source_datums if datum . id != match . id ] <EOL> yield match if match else target_datum . to_stack_datum ( ) <EOL> @ staticmethod <EOL> def find_best_match ( target_datum , source_datums ) : <EOL> """<STR_LIT>""" <EOL> candidate = None <EOL> for source_datum in source_datums : <EOL> if target_datum . id == source_datum . id : <EOL> if source_datum . case_type and source_datum . case_type == target_datum . case_type : <EOL> candidate = target_datum . to_stack_datum ( ) <EOL> break <EOL> elif not source_datum . case_type : <EOL> candidate = target_datum . to_stack_datum ( ) <EOL> else : <EOL> if source_datum . case_type and source_datum . case_type == target_datum . case_type : <EOL> candidate = target_datum . to_stack_datum ( source_id = source_datum . id ) <EOL> break <EOL> elif not source_datum . case_type : <EOL> if not candidate : <EOL> candidate = target_datum . to_stack_datum ( source_id = source_datum . id ) <EOL> return candidate <EOL> @ staticmethod <EOL> def get_datums_matched_to_manual_values ( target_frame_elements , manual_values ) : <EOL> """<STR_LIT>""" <EOL> manual_values_by_name = { datum . name : datum . xpath for datum in manual_values } <EOL> for child in target_frame_elements : <EOL> if not isinstance ( child , WorkflowDatumMeta ) or not child . requires_selection : <EOL> yield child <EOL> else : <EOL> manual_values = manual_values_by_name . get ( child . id ) <EOL> if manual_values : <EOL> yield StackDatum ( id = child . id , value = manual_values ) <EOL> else : <EOL> raise SuiteValidationError ( "<STR_LIT>" . format ( <EOL> child . id <EOL> ) ) <EOL> class CaseListFormWorkflow ( object ) : <EOL> def __init__ ( self , helper ) : <EOL> self . helper = helper <EOL> def case_list_forms_frames ( self , form ) : <EOL> if not form . is_case_list_form : <EOL> return [ ] <EOL> stack_frames = [ ] <EOL> for target_module in form . case_list_modules : <EOL> target_case_type = target_module . case_type <EOL> if not form . is_registration_form ( target_case_type ) : <EOL> continue <EOL> stack_frames . extend ( self . get_stack_frames_for_case_list_form_target ( target_module , form ) ) <EOL> return stack_frames <EOL> def get_stack_frames_for_case_list_form_target ( self , target_module , form ) : <EOL> stack_frames = [ ] <EOL> target_command = id_strings . menu_id ( target_module ) <EOL> if form . form_type == '<STR_LIT>' : <EOL> [ reg_action ] = form . get_registration_actions ( target_module . case_type ) <EOL> source_session_var = form . session_var_for_action ( reg_action ) <EOL> if form . form_type == '<STR_LIT>' : <EOL> reg_action = form . get_registration_actions ( target_module . case_type ) [ <NUM_LIT:0> ] <EOL> source_session_var = reg_action . case_session_var <EOL> source_case_id = session_var ( source_session_var ) <EOL> case_count = CaseIDXPath ( source_case_id ) . case ( ) . count ( ) <EOL> frame_case_created = StackFrameMeta ( None , self . get_if_clause ( case_count . gt ( <NUM_LIT:0> ) , target_command ) ) <EOL> stack_frames . append ( frame_case_created ) <EOL> frame_case_not_created = StackFrameMeta ( None , self . get_if_clause ( case_count . eq ( <NUM_LIT:0> ) , target_command ) ) <EOL> stack_frames . append ( frame_case_not_created ) <EOL> def add_datums_for_target ( module , source_form_dm , allow_missing = False ) : <EOL> """<STR_LIT>""" <EOL> target_form_dm = self . helper . get_frame_children ( module . get_form ( <NUM_LIT:0> ) , module_only = True ) <EOL> used = set ( ) <EOL> for source_meta in source_form_dm : <EOL> if source_meta . case_type : <EOL> try : <EOL> target_dm = self . get_target_dm ( target_form_dm , source_meta . case_type , module ) <EOL> except SuiteError : <EOL> if source_meta . requires_selection : <EOL> raise <EOL> else : <EOL> used . add ( source_meta ) <EOL> meta = WorkflowDatumMeta . from_session_datum ( source_meta ) <EOL> frame_case_created . add_child ( meta . to_stack_datum ( datum_id = target_dm . id ) ) <EOL> frame_case_not_created . add_child ( meta . to_stack_datum ( datum_id = target_dm . id ) ) <EOL> else : <EOL> source_case_type = self . get_case_type_created_by_form ( form , target_module ) <EOL> try : <EOL> target_dm = self . get_target_dm ( target_form_dm , source_case_type , module ) <EOL> except SuiteError : <EOL> if not allow_missing : <EOL> raise <EOL> else : <EOL> used . add ( source_meta ) <EOL> datum_meta = WorkflowDatumMeta . from_session_datum ( target_dm ) <EOL> frame_case_created . add_child ( datum_meta . to_stack_datum ( source_id = source_meta . id ) ) <EOL> return [ dm for dm in source_form_dm if dm not in used ] <EOL> source_form_dm = self . helper . get_form_datums ( form ) <EOL> if target_module . root_module_id : <EOL> root_module = target_module . root_module <EOL> root_module_command = CommandId ( id_strings . menu_id ( root_module ) ) <EOL> frame_case_created . add_child ( root_module_command ) <EOL> frame_case_not_created . add_child ( root_module_command ) <EOL> source_form_dm = add_datums_for_target ( root_module , source_form_dm , allow_missing = True ) <EOL> frame_case_created . add_child ( CommandId ( target_command ) ) <EOL> frame_case_not_created . add_child ( CommandId ( target_command ) ) <EOL> add_datums_for_target ( target_module , source_form_dm ) <EOL> return stack_frames <EOL> @ staticmethod <EOL> def get_target_dm ( target_form_dm , case_type , module ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> [ target_dm ] = [ <EOL> target_meta for target_meta in target_form_dm <EOL> if getattr ( target_meta , '<STR_LIT>' , None ) == case_type <EOL> ] <EOL> except ValueError : <EOL> raise SuiteValidationError ( <EOL> u"<STR_LIT>" <EOL> u"<STR_LIT>" . format ( module . default_name ( ) ) <EOL> ) <EOL> return target_dm <EOL> @ staticmethod <EOL> def get_if_clause ( case_count_xpath , target_command ) : <EOL> return_to = session_var ( RETURN_TO ) <EOL> return XPath . and_ ( <EOL> return_to . count ( ) . eq ( <NUM_LIT:1> ) , <EOL> return_to . eq ( XPath . string ( target_command ) ) , <EOL> case_count_xpath <EOL> ) <EOL> @ staticmethod <EOL> def get_case_type_created_by_form ( form , target_module ) : <EOL> if form . form_type == '<STR_LIT>' : <EOL> [ reg_action ] = form . get_registration_actions ( target_module . case_type ) <EOL> if reg_action == '<STR_LIT>' : <EOL> return form . get_module ( ) . case_type <EOL> else : <EOL> return reg_action . case_type <EOL> elif form . form_type == '<STR_LIT>' : <EOL> return form . get_registration_actions ( target_module . case_type ) [ <NUM_LIT:0> ] . case_type <EOL> class StackFrameMeta ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , if_prefix , if_clause , children = None , allow_empty_frame = False ) : <EOL> if if_prefix : <EOL> template = '<STR_LIT>' . format ( if_clause ) if if_clause else '<STR_LIT:{}>' <EOL> if_clause = template . format ( if_prefix ) <EOL> self . if_clause = unescape ( if_clause ) if if_clause else None <EOL> self . children = [ ] <EOL> self . allow_empty_frame = allow_empty_frame <EOL> if children : <EOL> for child in children : <EOL> self . add_child ( child ) <EOL> def add_child ( self , child ) : <EOL> if isinstance ( child , WorkflowDatumMeta ) : <EOL> child = child . to_stack_datum ( ) <EOL> self . children . append ( child ) <EOL> def to_frame ( self ) : <EOL> if not self . children and not self . allow_empty_frame : <EOL> return <EOL> frame = CreateFrame ( if_clause = self . if_clause ) <EOL> for child in self . children : <EOL> if isinstance ( child , CommandId ) : <EOL> frame . add_command ( XPath . string ( child . id ) ) <EOL> elif isinstance ( child , StackDatum ) : <EOL> frame . add_datum ( child ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" . format ( type ( child ) , child ) ) <EOL> return frame <EOL> @ total_ordering <EOL> class CommandId ( object ) : <EOL> def __init__ ( self , command ) : <EOL> self . id = command <EOL> def __lt__ ( self , other ) : <EOL> return self . id < other . id <EOL> def __eq__ ( self , other ) : <EOL> return self . id == other . id <EOL> def __ne__ ( self , other ) : <EOL> return not self == other <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . id ) <EOL> @ total_ordering <EOL> class WorkflowDatumMeta ( object ) : <EOL> """<STR_LIT>""" <EOL> type_regex = re . compile ( "<STR_LIT>" ) <EOL> def __init__ ( self , datum_id , nodeset , function ) : <EOL> self . id = datum_id <EOL> self . nodeset = nodeset <EOL> self . function = function <EOL> @ classmethod <EOL> def from_session_datum ( cls , session_datum ) : <EOL> return cls ( session_datum . id , session_datum . nodeset , session_datum . function ) <EOL> @ property <EOL> def requires_selection ( self ) : <EOL> return bool ( self . nodeset ) <EOL> @ property <EOL> @ memoized <EOL> def case_type ( self ) : <EOL> """<STR_LIT>""" <EOL> def _extract_type ( xpath ) : <EOL> match = self . type_regex . search ( xpath ) <EOL> return match . group ( <NUM_LIT:1> ) if match else None <EOL> if self . nodeset : <EOL> return _extract_type ( self . nodeset ) <EOL> elif self . function : <EOL> return _extract_type ( self . function ) <EOL> def to_stack_datum ( self , datum_id = None , source_id = None ) : <EOL> value = session_var ( source_id or self . id ) if self . requires_selection else self . function <EOL> return StackDatum ( id = datum_id or self . id , value = value ) <EOL> def __lt__ ( self , other ) : <EOL> return self . id < other . id <EOL> def __eq__ ( self , other ) : <EOL> return self . id == other . id <EOL> def __ne__ ( self , other ) : <EOL> return not self == other <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . id , self . case_type ) </s>
