signature:
def GetColumnHeaders ( self ) -> list :
    """Call IUIAutomationTablePattern : : GetCurrentColumnHeaders .
 Return list , a list of ` Control ` subclasses , representing all the column headers in a table . .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtablepattern - getcurrentcolumnheaders"""

implementation:
eleArray = self . pattern . GetCurrentColumnHeaders ( )
if eleArray :
    controls = [ ]
    for i in range ( eleArray . Length ) :
        ele = eleArray . GetElement ( i )
        con = Control . CreateControlFromElement ( element = ele )
        if con :
            controls . append ( con )
    return controls
return [ ]


signature:
def GetRowHeaders ( self ) -> list :
    """Call IUIAutomationTablePattern : : GetCurrentRowHeaders .
 Return list , a list of ` Control ` subclasses , representing all the row headers in a table .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtablepattern - getcurrentrowheaders"""

implementation:
eleArray = self . pattern . GetCurrentRowHeaders ( )
if eleArray :
    controls = [ ]
    for i in range ( eleArray . Length ) :
        ele = eleArray . GetElement ( i )
        con = Control . CreateControlFromElement ( element = ele )
        if con :
            controls . append ( con )
    return controls
return [ ]


signature:
def Compare ( self , textRange : 'TextRange' ) -> bool :
    """Call IUIAutomationTextRange : : Compare .
 textRange : ` TextRange ` .
 Return bool , specifies whether this text range has the same endpoints as another text range .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - compare"""

implementation: return bool ( self . textRange . Compare ( textRange . textRange ) )


signature:
def CompareEndpoints ( self , srcEndPoint : int , textRange : 'TextRange' , targetEndPoint : int ) -> int :
    """Call IUIAutomationTextRange : : CompareEndpoints .
 srcEndPoint : int , a value in class ` TextPatternRangeEndpoint ` .
 textRange : ` TextRange ` .
 targetEndPoint : int , a value in class ` TextPatternRangeEndpoint ` .
 Return int , a negative value if the caller ' s endpoint occurs earlier in the text than the target endpoint ;
 0 if the caller ' s endpoint is at the same location as the target endpoint ;
 or a positive value if the caller ' s endpoint occurs later in the text than the target endpoint .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - compareendpoints"""

implementation: return self . textRange . CompareEndpoints ( srcEndPoint , textRange , targetEndPoint )


signature:
def FindAttribute ( self , textAttributeId : int , val , backward : bool ) -> 'TextRange' :
    """Call IUIAutomationTextRange : : FindAttribute .
 textAttributeID : int , a value in class ` TextAttributeId ` .
 val : COM VARIANT according to textAttributeId ? todo .
 backward : bool , True if the last occurring text range should be returned instead of the first ; otherwise False .
 return ` TextRange ` or None , a text range subset that has the specified text attribute value .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - findattribute"""

implementation:
textRange = self . textRange . FindAttribute ( textAttributeId , val , int ( backward ) )
if textRange :
    return TextRange ( textRange = textRange )


signature:
def FindText ( self , text : str , backward : bool , ignoreCase : bool ) -> 'TextRange' :
    """Call IUIAutomationTextRange : : FindText .
 text : str ,
 backward : bool , True if the last occurring text range should be returned instead of the first ; otherwise False .
 ignoreCase : bool , True if case should be ignored ; otherwise False .
 return ` TextRange ` or None , a text range subset that contains the specified text .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - findtext"""

implementation:
textRange = self . textRange . FindText ( text , int ( backward ) , int ( ignoreCase ) )
if textRange :
    return TextRange ( textRange = textRange )


signature:
def GetAttributeValue ( self , textAttributeId : int ) -> ctypes . POINTER ( comtypes . IUnknown ) :
    """Call IUIAutomationTextRange : : GetAttributeValue .
 textAttributeId : int , a value in class ` TextAttributeId ` .
 Return ` ctypes . POINTER ( comtypes . IUnknown ) ` or None , the value of the specified text attribute across the entire text range , todo .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - getattributevalue"""

implementation: return self . textRange . GetAttributeValue ( textAttributeId )


signature:
def GetBoundingRectangles ( self ) -> list :
    """Call IUIAutomationTextRange : : GetBoundingRectangles .
 textAttributeId : int , a value in class ` TextAttributeId ` .
 Return list , a list of ` Rect ` .
 bounding rectangles for each fully or partially visible line of text in a text range . .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - getboundingrectangles
 for rect in textRange . GetBoundingRectangles ( ) :
 print ( rect . left , rect . top , rect . right , rect . bottom , rect . width ( ) , rect . height ( ) , rect . xcenter ( ) , rect . ycenter ( ) )"""

implementation:
floats = self . textRange . GetBoundingRectangles ( )
rects = [ ]
for i in range ( len ( floats ) // 4 ) :
    rect = Rect ( int ( floats [ i * 4 ] ) , int ( floats [ i * 4 + 1 ] ) , int ( floats [ i * 4 ] ) + int ( floats [ i * 4 + 2 ] ) , int ( floats [ i * 4 + 1 ] ) + int ( floats [ i * 4 + 3 ] ) )
    rects . append ( rect )
return rects


signature:
def GetChildren ( self ) -> list :
    """Call IUIAutomationTextRange : : GetChildren .
 textAttributeId : int , a value in class ` TextAttributeId ` .
 Return list , a list of ` Control ` subclasses , embedded objects that fall within the text range . .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - getchildren"""

implementation:
eleArray = self . textRange . GetChildren ( )
if eleArray :
    controls = [ ]
    for i in range ( eleArray . Length ) :
        ele = eleArray . GetElement ( i )
        con = Control . CreateControlFromElement ( element = ele )
        if con :
            controls . append ( con )
    return controls
return [ ]


signature:
def Move ( self , unit : int , count : int , waitTime : float = OPERATION_WAIT_TIME ) -> int :
    """Call IUIAutomationTextRange : : Move .
 Move the text range forward or backward by the specified number of text units .
 unit : int , a value in class ` TextUnit ` .
 count : int , the number of text units to move .
 A positive value moves the text range forward .
 A negative value moves the text range backward . Zero has no effect .
 waitTime : float .
 Return : int , the number of text units actually moved .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - move"""

implementation:
ret = self . textRange . Move ( unit , count )
time . sleep ( waitTime )
return ret


signature:
def MoveEndpointByRange ( self , srcEndPoint : int , textRange : 'TextRange' , targetEndPoint : int , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call IUIAutomationTextRange : : MoveEndpointByRange .
 Move one endpoint of the current text range to the specified endpoint of a second text range .
 srcEndPoint : int , a value in class ` TextPatternRangeEndpoint ` .
 textRange : ` TextRange ` .
 targetEndPoint : int , a value in class ` TextPatternRangeEndpoint ` .
 waitTime : float .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - moveendpointbyrange"""

implementation:
ret = self . textRange . MoveEndpointByRange ( srcEndPoint , textRange . textRange , targetEndPoint ) == S_OK
time . sleep ( waitTime )
return ret


signature:
def MoveEndpointByUnit ( self , endPoint : int , unit : int , count : int , waitTime : float = OPERATION_WAIT_TIME ) -> int :
    """Call IUIAutomationTextRange : : MoveEndpointByUnit .
 Move one endpoint of the text range the specified number of text units within the document range .
 endPoint : int , a value in class ` TextPatternRangeEndpoint ` .
 unit : int , a value in class ` TextUnit ` .
 count : int , the number of units to move .
 A positive count moves the endpoint forward .
 A negative count moves backward .
 A count of 0 has no effect .
 waitTime : float .
 Return int , the count of units actually moved .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - moveendpointbyunit"""

implementation:
ret = self . textRange . MoveEndpointByUnit ( endPoint , unit , count )
time . sleep ( waitTime )
return ret


signature:
def ScrollIntoView ( self , alignTop : bool = True , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call IUIAutomationTextRange : : ScrollIntoView .
 Cause the text control to scroll until the text range is visible in the viewport .
 alignTop : bool , True if the text control should be scrolled so that the text range is flush with the top of the viewport ;
 False if it should be flush with the bottom of the viewport .
 waitTime : float .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextrange - scrollintoview"""

implementation:
ret = self . textRange . ScrollIntoView ( int ( alignTop ) ) == S_OK
time . sleep ( waitTime )
return ret


signature:
def GetActiveComposition ( self ) -> TextRange :
    """Call IUIAutomationTextEditPattern : : GetActiveComposition .
 Return ` TextRange ` or None , the active composition .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtexteditpattern - getactivecomposition"""

implementation:
textRange = self . pattern . GetActiveComposition ( )
if textRange :
    return TextRange ( textRange = textRange )


signature:
def GetConversionTarget ( self ) -> TextRange :
    """Call IUIAutomationTextEditPattern : : GetConversionTarget .
 Return ` TextRange ` or None , the current conversion target range . .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtexteditpattern - getconversiontarget"""

implementation:
textRange = self . pattern . GetConversionTarget ( )
if textRange :
    return TextRange ( textRange = textRange )


signature:
def GetVisibleRanges ( self ) -> list :
    """Call IUIAutomationTextPattern : : GetVisibleRanges .
 Return list , a list of ` TextRange ` , disjoint text ranges from a text - based control
 where each text range represents a contiguous span of visible text .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextpattern - getvisibleranges"""

implementation:
eleArray = self . pattern . GetVisibleRanges ( )
if eleArray :
    textRanges = [ ]
    for i in range ( eleArray . Length ) :
        ele = eleArray . GetElement ( i )
        textRanges . append ( TextRange ( textRange = ele ) )
    return textRanges
return [ ]


signature:
def RangeFromChild ( self , child ) -> TextRange :
    """Call IUIAutomationTextPattern : : RangeFromChild .
 child : ` Control ` or its subclass .
 Return ` TextRange ` or None , a text range enclosing a child element such as an image ,
 hyperlink , Microsoft Excel spreadsheet , or other embedded object .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextpattern - rangefromchild"""

implementation:
textRange = self . pattern . RangeFromChild ( Control . Element )
if textRange :
    return TextRange ( textRange = textRange )


signature:
def RangeFromPoint ( self , x : int , y : int ) -> TextRange :
    """Call IUIAutomationTextPattern : : RangeFromPoint .
 child : ` Control ` or its subclass .
 Return ` TextRange ` or None , the degenerate ( empty ) text range nearest to the specified screen coordinates .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtextpattern - rangefrompoint"""

implementation:
textRange = self . pattern . RangeFromPoint ( ctypes . wintypes . POINT ( x , y ) )
if textRange :
    return TextRange ( textRange = textRange )


signature:
def Move ( self , x : int , y : int , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call IUIAutomationTransformPattern : : Move .
 Move the UI Automation element .
 x : int .
 y : int .
 waitTime : float .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtransformpattern - move"""

implementation:
ret = self . pattern . Move ( x , y ) == S_OK
time . sleep ( waitTime )
return ret


signature:
def Resize ( self , width : int , height : int , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call IUIAutomationTransformPattern : : Resize .
 Resize the UI Automation element .
 width : int .
 height : int .
 waitTime : float .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtransformpattern - resize"""

implementation:
ret = self . pattern . Resize ( width , height ) == S_OK
time . sleep ( waitTime )
return ret


signature:
def Rotate ( self , degrees : int , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call IUIAutomationTransformPattern : : Rotate .
 Rotates the UI Automation element .
 degrees : int .
 waitTime : float .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtransformpattern - rotate"""

implementation:
ret = self . pattern . Rotate ( degrees ) == S_OK
time . sleep ( waitTime )
return ret


signature:
def Zoom ( self , zoomLevel : float , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call IUIAutomationTransformPattern2 : : Zoom .
 Zoom the viewport of the control .
 zoomLevel : float for int .
 waitTime : float .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtransformpattern2 - zoom"""

implementation:
ret = self . pattern . Zoom ( zoomLevel ) == S_OK
time . sleep ( waitTime )
return ret


signature:
def ZoomByUnit ( self , zoomUnit : int , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call IUIAutomationTransformPattern2 : : ZoomByUnit .
 Zoom the viewport of the control by the specified unit .
 zoomUnit : int , a value in class ` ZoomUnit ` .
 waitTime : float .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationtransformpattern2 - zoombyunit"""

implementation:
ret = self . pattern . ZoomByUnit ( zoomUnit ) == S_OK
time . sleep ( waitTime )
return ret


signature:
def SetWindowVisualState ( self , state : int , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call IUIAutomationWindowPattern : : SetWindowVisualState .
 Minimize , maximize , or restore the window .
 state : int , a value in class ` WindowVisualState ` .
 waitTime : float .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationwindowpattern - setwindowvisualstate"""

implementation:
ret = self . pattern . SetWindowVisualState ( state ) == S_OK
time . sleep ( waitTime )
return ret


signature:
def WaitForInputIdle ( self , milliseconds : int ) -> bool :
    '''Call IUIAutomationWindowPattern : : WaitForInputIdle .
 Cause the calling code to block for the specified time or
 until the associated process enters an idle state , whichever completes first .
 milliseconds : int .
 Return bool , True if succeed otherwise False .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationwindowpattern - waitforinputidle'''

implementation: return self . pattern . WaitForInputIdle ( milliseconds ) == S_OK


signature:
def CreateControlFromElement ( element ) -> 'Control' :
    """Create a concreate ` Control ` from a com type ` IUIAutomationElement ` .
 element : ` ctypes . POINTER ( IUIAutomationElement ) ` .
 Return a subclass of ` Control ` , an instance of the control ' s real type ."""

implementation:
if element :
    controlType = element . CurrentControlType
    if controlType in ControlConstructors :
        return ControlConstructors [ controlType ] ( element = element )
    else :
        Logger . WriteLine ( "element.CurrentControlType returns {}, invalid ControlType!" . format ( controlType ) , ConsoleColor . Red )


signature:
def AddSearchProperties ( self , ** searchProperties ) -> None :
    """Add search properties using ` dict . update ` .
 searchProperties : dict , same as searchProperties in ` Control . _ _ init _ _ ` ."""

implementation:
self . searchProperties . update ( searchProperties )
if 'Depth' in searchProperties :
    self . searchDepth = searchProperties [ 'Depth' ]
if 'RegexName' in searchProperties :
    regName = searchProperties [ 'RegexName' ]
    self . regexName = re . compile ( regName ) if regName else None


signature:
def RemoveSearchProperties ( self , ** searchProperties ) -> None :
    """searchProperties : dict , same as searchProperties in ` Control . _ _ init _ _ ` ."""

implementation:
for key in searchProperties :
    del self . searchProperties [ key ]
    if key == 'RegexName' :
        self . regexName = None


signature:
def GetColorfulSearchPropertiesStr ( self , keyColor = 'DarkGreen' , valueColor = 'DarkCyan' ) -> str :
    """keyColor , valueColor : str , color name in class ConsoleColor"""

implementation:
strs = [ '<Color={}>{}</Color>: <Color={}>{}</Color>' . format ( keyColor if k in Control . ValidKeys else 'DarkYellow' , k , valueColor , ControlTypeNames [ v ] if k == 'ControlType' else repr ( v ) ) for k , v in self . searchProperties . items ( ) ]
return '{' + ', ' . join ( strs ) + '}'


signature:
def BoundingRectangle ( self ) -> Rect :
    """Property BoundingRectangle .
 Call IUIAutomationElement : : get _ CurrentBoundingRectangle .
 Return ` Rect ` .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationelement - get _ currentboundingrectangle
 rect = control . BoundingRectangle
 print ( rect . left , rect . top , rect . right , rect . bottom , rect . width ( ) , rect . height ( ) , rect . xcenter ( ) , rect . ycenter ( ) )"""

implementation:
rect = self . Element . CurrentBoundingRectangle
return Rect ( rect . left , rect . top , rect . right , rect . bottom )


signature:
def GetClickablePoint ( self ) -> tuple :
    """Call IUIAutomationElement : : GetClickablePoint .
 Return tuple , ( x : int , y : int , gotClickable : bool ) , like ( 20 , 10 , True )
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationelement - getclickablepoint"""

implementation:
point , gotClickable = self . Element . GetClickablePoint ( )
return ( point . x , point . y , bool ( gotClickable ) )


signature:
def GetPattern ( self , patternId : int ) :
    """Call IUIAutomationElement : : GetCurrentPattern .
 Get a new pattern by pattern id if it supports the pattern .
 patternId : int , a value in class ` PatternId ` .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationelement - getcurrentpattern"""

implementation:
try :
    pattern = self . Element . GetCurrentPattern ( patternId )
    if pattern :
        subPattern = CreatePattern ( patternId , pattern )
        self . _supportedPatterns [ patternId ] = subPattern
        return subPattern
except comtypes . COMError as ex :
    pass


signature:
def GetPatternAs ( self , patternId : int , riid ) :
    """Call IUIAutomationElement : : GetCurrentPatternAs .
 Get a new pattern by pattern id if it supports the pattern , todo .
 patternId : int , a value in class ` PatternId ` .
 riid : GUID .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationelement - getcurrentpatternas"""

implementation: return self . Element . GetCurrentPatternAs ( patternId , riid )


signature:
def GetPropertyValueEx ( self , propertyId : int , ignoreDefaultValue : int ) -> Any :
    """Call IUIAutomationElement : : GetCurrentPropertyValueEx .
 propertyId : int , a value in class ` PropertyId ` .
 ignoreDefaultValue : int , 0 or 1.
 Return Any , corresponding type according to propertyId .
 Refer https : / / docs . microsoft . com / en - us / windows / desktop / api / uiautomationclient / nf - uiautomationclient - iuiautomationelement - getcurrentpropertyvalueex"""

implementation: return self . Element . GetCurrentPropertyValueEx ( propertyId , ignoreDefaultValue )


signature:
def Element ( self ) :
    """Property Element .
 Return ` ctypes . POINTER ( IUIAutomationElement ) ` ."""

implementation:
if not self . _element :
    self . Refind ( maxSearchSeconds = TIME_OUT_SECOND , searchIntervalSeconds = self . searchWaitTime )
return self . _element


signature:
def GetCachedPattern ( self , patternId : int , cache : bool ) :
    """Get a pattern by patternId .
 patternId : int , a value in class ` PatternId ` .
 Return a pattern if it supports the pattern else None .
 cache : bool , if True , store the pattern for later use , if False , get a new pattern by ` self . GetPattern ` ."""

implementation:
if cache :
    pattern = self . _supportedPatterns . get ( patternId , None )
    if pattern :
        return pattern
    else :
        pattern = self . GetPattern ( patternId )
        if pattern :
            self . _supportedPatterns [ patternId ] = pattern
            return pattern
else :
    pattern = self . GetPattern ( patternId )
    if pattern :
        self . _supportedPatterns [ patternId ] = pattern
        return pattern


signature:
def GetAncestorControl ( self , condition : Callable ) -> 'Control' :
    """Get a ancestor control that matches the condition .
 condition : Callable , function ( control : Control , depth : int ) - > bool ,
 depth starts with - 1 and decreses when search goes up .
 Return ` Control ` subclass or None ."""

implementation:
ancestor = self
depth = 0
while True :
    ancestor = ancestor . GetParentControl ( )
    depth -= 1
    if ancestor :
        if condition ( ancestor , depth ) :
            return ancestor
    else :
        break


signature:
def GetParentControl ( self ) -> 'Control' :
    """Return ` Control ` subclass or None ."""

implementation:
ele = _AutomationClient . instance ( ) . ViewWalker . GetParentElement ( self . Element )
return Control . CreateControlFromElement ( ele )


signature:
def GetFirstChildControl ( self ) -> 'Control' :
    """Return ` Control ` subclass or None ."""

implementation:
ele = _AutomationClient . instance ( ) . ViewWalker . GetFirstChildElement ( self . Element )
return Control . CreateControlFromElement ( ele )


signature:
def GetLastChildControl ( self ) -> 'Control' :
    """Return ` Control ` subclass or None ."""

implementation:
ele = _AutomationClient . instance ( ) . ViewWalker . GetLastChildElement ( self . Element )
return Control . CreateControlFromElement ( ele )


signature:
def GetNextSiblingControl ( self ) -> 'Control' :
    """Return ` Control ` subclass or None ."""

implementation:
ele = _AutomationClient . instance ( ) . ViewWalker . GetNextSiblingElement ( self . Element )
return Control . CreateControlFromElement ( ele )


signature:
def GetPreviousSiblingControl ( self ) -> 'Control' :
    """Return ` Control ` subclass or None ."""

implementation:
ele = _AutomationClient . instance ( ) . ViewWalker . GetPreviousSiblingElement ( self . Element )
return Control . CreateControlFromElement ( ele )


signature:
def GetSiblingControl ( self , condition : Callable , forward : bool = True ) -> 'Control' :
    """Find a SiblingControl by condition ( control : Control ) - > bool .
 forward : bool , if True , only search next siblings , if False , search pervious siblings first , then search next siblings .
 condition : Callable , function ( control : Control ) - > bool .
 Return ` Control ` subclass or None ."""

implementation:
if not forward :
    prev = self
    while True :
        prev = prev . GetPreviousSiblingControl ( )
        if prev :
            if condition ( prev ) :
                return prev
        else :
            break
next_ = self
while True :
    next_ = next_ . GetNextSiblingControl ( )
    if next_ :
        if condition ( next_ ) :
            return next_
    else :
        break


signature:
def GetChildren ( self ) -> list :
    """Return list , a list of ` Control ` subclasses ."""

implementation:
children = [ ]
child = self . GetFirstChildControl ( )
while child :
    children . append ( child )
    child = child . GetNextSiblingControl ( )
return children


signature:
def _CompareFunction ( self , control : 'Control' , depth : int ) -> bool :
    """Define how to search .
 control : ` Control ` or its subclass .
 depth : int , tree depth from searchFromControl .
 Return bool ."""

implementation:
for key , value in self . searchProperties . items ( ) :
    if 'ControlType' == key :
        if value != control . ControlType :
            return False
    elif 'ClassName' == key :
        if value != control . ClassName :
            return False
    elif 'AutomationId' == key :
        if value != control . AutomationId :
            return False
    elif 'Name' == key :
        if value != control . Name :
            return False
    elif 'SubName' == key :
        if value not in control . Name :
            return False
    elif 'RegexName' == key :
        if not self . regexName . match ( control . Name ) :
            return False
    elif 'Depth' == key :
        if value != depth :
            return False
    elif 'Compare' == key :
        if not value ( control , depth ) :
            return False
return True


signature:
def Exists ( self , maxSearchSeconds : float = 5 , searchIntervalSeconds : float = SEARCH_INTERVAL , printIfNotExist : bool = False ) -> bool :
    """maxSearchSeconds : float
 searchIntervalSeconds : float
 Find control every searchIntervalSeconds seconds in maxSearchSeconds seconds .
 Return bool , True if find"""

implementation:
if self . _element and self . _elementDirectAssign : # if element is directly assigned , not by searching , just check whether self . _ element is valid
# but I can ' t find an API in UIAutomation that can directly check
    rootElement = GetRootControl ( ) . Element
    if self . _element == rootElement :
        return True
    else :
        parentElement = _AutomationClient . instance ( ) . ViewWalker . GetParentElement ( self . _element )
        if parentElement :
            return True
        else :
            return False
# find the element
if len ( self . searchProperties ) == 0 :
    raise LookupError ( "control's searchProperties must not be empty!" )
self . _element = None
startTime = ProcessTime ( )
# Use same timeout ( s ) parameters for resolve all parents
prev = self . searchFromControl
if prev and not prev . _element and not prev . Exists ( maxSearchSeconds , searchIntervalSeconds ) :
    if printIfNotExist or DEBUG_EXIST_DISAPPEAR :
        Logger . ColorfullyWriteLine ( self . GetColorfulSearchPropertiesStr ( ) + '<Color=Red> does not exist.</Color>' )
    return False
startTime2 = ProcessTime ( )
if DEBUG_SEARCH_TIME :
    startDateTime = datetime . datetime . now ( )
while True :
    control = FindControl ( self . searchFromControl , self . _CompareFunction , self . searchDepth , False , self . foundIndex )
    if control :
        self . _element = control . Element
        control . _element = 0
        # control will be destroyed , but the element needs to be stroed in self . _ element
        if DEBUG_SEARCH_TIME :
            Logger . ColorfullyWriteLine ( '{} TraverseControls: <Color=Cyan>{}</Color>, SearchTime: <Color=Cyan>{:.3f}</Color>s[{} - {}]' . format ( self . GetColorfulSearchPropertiesStr ( ) , control . traverseCount , ProcessTime ( ) - startTime2 , startDateTime . time ( ) , datetime . datetime . now ( ) . time ( ) ) )
        return True
    else :
        remain = startTime + maxSearchSeconds - ProcessTime ( )
        if remain > 0 :
            time . sleep ( min ( remain , searchIntervalSeconds ) )
        else :
            if printIfNotExist or DEBUG_EXIST_DISAPPEAR :
                Logger . ColorfullyWriteLine ( self . GetColorfulSearchPropertiesStr ( ) + '<Color=Red> does not exist.</Color>' )
            return False


signature:
def Disappears ( self , maxSearchSeconds : float = 5 , searchIntervalSeconds : float = SEARCH_INTERVAL , printIfNotDisappear : bool = False ) -> bool :
    """maxSearchSeconds : float
 searchIntervalSeconds : float
 Check if control disappears every searchIntervalSeconds seconds in maxSearchSeconds seconds .
 Return bool , True if control disappears ."""

implementation:
global DEBUG_EXIST_DISAPPEAR
start = ProcessTime ( )
while True :
    temp = DEBUG_EXIST_DISAPPEAR
    DEBUG_EXIST_DISAPPEAR = False
    # do not print for Exists
    if not self . Exists ( 0 , 0 , False ) :
        DEBUG_EXIST_DISAPPEAR = temp
        return True
    DEBUG_EXIST_DISAPPEAR = temp
    remain = start + maxSearchSeconds - ProcessTime ( )
    if remain > 0 :
        time . sleep ( min ( remain , searchIntervalSeconds ) )
    else :
        if printIfNotDisappear or DEBUG_EXIST_DISAPPEAR :
            Logger . ColorfullyWriteLine ( self . GetColorfulSearchPropertiesStr ( ) + '<Color=Red> does not disappear.</Color>' )
        return False


signature:
def Refind ( self , maxSearchSeconds : float = TIME_OUT_SECOND , searchIntervalSeconds : float = SEARCH_INTERVAL , raiseException : bool = True ) -> bool :
    """Refind the control every searchIntervalSeconds seconds in maxSearchSeconds seconds .
 maxSearchSeconds : float .
 searchIntervalSeconds : float .
 raiseException : bool , if True , raise a LookupError if timeout .
 Return bool , True if find ."""

implementation:
if not self . Exists ( maxSearchSeconds , searchIntervalSeconds , False if raiseException else DEBUG_EXIST_DISAPPEAR ) :
    if raiseException :
        Logger . ColorfullyWriteLine ( '<Color=Red>Find Control Timeout: </Color>' + self . GetColorfulSearchPropertiesStr ( ) )
        raise LookupError ( 'Find Control Timeout: ' + self . GetSearchPropertiesStr ( ) )
    else :
        return False
return True


signature:
def MoveCursorToInnerPos ( self , x : int = None , y : int = None , ratioX : float = 0.5 , ratioY : float = 0.5 , simulateMove : bool = True ) -> tuple :
    """Move cursor to control ' s internal position , default to center .
 x : int , if < 0 , move to self . BoundingRectangle . right + x , if not None , ignore ratioX .
 y : int , if < 0 , move to self . BoundingRectangle . bottom + y , if not None , ignore ratioY .
 ratioX : float .
 ratioY : float .
 simulateMove : bool .
 Return tuple , two ints ( x , y ) , the cursor positon relative to screen ( 0,0 ) after moving or None if control ' s width or height = = 0."""

implementation:
rect = self . BoundingRectangle
if rect . width ( ) == 0 or rect . height ( ) == 0 :
    Logger . ColorfullyWriteLine ( '<Color=Yellow>Can not move curosr</Color>. {}\'s BoundingRectangle is {}. SearchProperties: {}' . format ( self . ControlTypeName , rect , self . GetColorfulSearchPropertiesStr ( ) ) )
    return
if x is None :
    x = rect . left + int ( rect . width ( ) * ratioX )
else :
    x = ( rect . left if x >= 0 else rect . right ) + x
if y is None :
    y = rect . top + int ( rect . height ( ) * ratioY )
else :
    y = ( rect . top if y >= 0 else rect . bottom ) + y
if simulateMove and MAX_MOVE_SECOND > 0 :
    MoveTo ( x , y , waitTime = 0 )
else :
    SetCursorPos ( x , y )
return x , y


signature:
def MoveCursorToMyCenter ( self , simulateMove : bool = True ) -> tuple :
    """Move cursor to control ' s center .
 Return tuple , two ints tuple ( x , y ) , the cursor positon relative to screen ( 0,0 ) after moving ."""

implementation: return self . MoveCursorToInnerPos ( simulateMove = simulateMove )


signature:
def Click ( self , x : int = None , y : int = None , ratioX : float = 0.5 , ratioY : float = 0.5 , simulateMove : bool = True , waitTime : float = OPERATION_WAIT_TIME ) -> None :
    """x : int , if < 0 , click self . BoundingRectangle . right + x , if not None , ignore ratioX .
 y : int , if < 0 , click self . BoundingRectangle . bottom + y , if not None , ignore ratioY .
 ratioX : float .
 ratioY : float .
 simulateMove : bool , if True , first move cursor to control smoothly .
 waitTime : float .
 Click ( ) , Click ( ratioX = 0.5 , ratioY = 0.5 ) : click center .
 Click ( 10 , 10 ) : click left + 10 , top + 10.
 Click ( - 10 , - 10 ) : click right - 10 , bottom - 10."""

implementation:
point = self . MoveCursorToInnerPos ( x , y , ratioX , ratioY , simulateMove )
if point :
    Click ( point [ 0 ] , point [ 1 ] , waitTime )


signature:
def DoubleClick ( self , x : int = None , y : int = None , ratioX : float = 0.5 , ratioY : float = 0.5 , simulateMove : bool = True , waitTime : float = OPERATION_WAIT_TIME ) -> None :
    """x : int , if < 0 , right click self . BoundingRectangle . right + x , if not None , ignore ratioX .
 y : int , if < 0 , right click self . BoundingRectangle . bottom + y , if not None , ignore ratioY .
 ratioX : float .
 ratioY : float .
 simulateMove : bool , if True , first move cursor to control smoothly .
 waitTime : float .
 DoubleClick ( ) , DoubleClick ( ratioX = 0.5 , ratioY = 0.5 ) : double click center .
 DoubleClick ( 10 , 10 ) : double click left + 10 , top + 10.
 DoubleClick ( - 10 , - 10 ) : double click right - 10 , bottom - 10."""

implementation:
x , y = self . MoveCursorToInnerPos ( x , y , ratioX , ratioY , simulateMove )
Click ( x , y , GetDoubleClickTime ( ) * 1.0 / 2000 )
Click ( x , y , waitTime )


signature:
def WheelDown ( self , wheelTimes : int = 1 , interval : float = 0.05 , waitTime : float = OPERATION_WAIT_TIME ) -> None :
    """Make control have focus first , move cursor to center and mouse wheel down .
 wheelTimes : int .
 interval : float .
 waitTime : float ."""

implementation:
x , y = GetCursorPos ( )
self . SetFocus ( )
self . MoveCursorToMyCenter ( False )
WheelDown ( wheelTimes , interval , waitTime )
SetCursorPos ( x , y )


signature:
def ShowWindow ( self , cmdShow : int , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Get a native handle from self or ancestors until valid and call native ` ShowWindow ` with cmdShow .
 cmdShow : int , a value in in class ` SW ` .
 waitTime : float .
 Return bool , True if succeed otherwise False ."""

implementation:
handle = self . NativeWindowHandle
if not handle :
    control = self
    while not handle :
        control = control . GetParentControl ( )
        handle = control . NativeWindowHandle
if handle :
    ret = ShowWindow ( handle , cmdShow )
    time . sleep ( waitTime )
    return ret


signature:
def Show ( self , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call native ` ShowWindow ( SW . Show ) ` .
 Return bool , True if succeed otherwise False ."""

implementation: return self . ShowWindow ( SW . Show , waitTime )


signature:
def Hide ( self , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Call native ` ShowWindow ( SW . Hide ) ` .
 waitTime : float
 Return bool , True if succeed otherwise False ."""

implementation: return self . ShowWindow ( SW . Hide , waitTime )


signature:
def MoveWindow ( self , x : int , y : int , width : int , height : int , repaint : bool = True ) -> bool :
    """Call native MoveWindow if control has a valid native handle .
 x : int .
 y : int .
 width : int .
 height : int .
 repaint : bool .
 Return bool , True if succeed otherwise False ."""

implementation:
handle = self . NativeWindowHandle
if handle :
    return MoveWindow ( handle , x , y , width , height , int ( repaint ) )
return False


signature:
def SetWindowText ( self , text : str ) -> bool :
    """Call native SetWindowText if control has a valid native handle ."""

implementation:
handle = self . NativeWindowHandle
if handle :
    return SetWindowText ( handle , text )
return False


signature:
def SendKey ( self , key : int , waitTime : float = OPERATION_WAIT_TIME ) -> None :
    """Make control have focus first and type a key .
 ` self . SetFocus ` may not work for some controls , you may need to click it to make it have focus .
 key : int , a key code value in class Keys .
 waitTime : float ."""

implementation:
self . SetFocus ( )
SendKey ( key , waitTime )


signature:
def SendKeys ( self , keys : str , interval : float = 0.01 , waitTime : float = OPERATION_WAIT_TIME ) -> None :
    """Make control have focus first and type keys .
 ` self . SetFocus ` may not work for some controls , you may need to click it to make it have focus .
 keys : str , keys to type , see the docstring of ` SendKeys ` .
 interval : float , seconds between keys ."""

implementation:
self . SetFocus ( )
SendKeys ( keys , interval , waitTime )


signature:
def GetPixelColor ( self , x : int , y : int ) -> int :
    """Call native ` GetPixelColor ` if control has a valid native handle .
 Use ` self . ToBitmap ` if control doesn ' t have a valid native handle or you get many pixels .
 x : int , internal x position .
 y : int , internal y position .
 Return int , a color value in bgr .
 r = bgr & 0x0000FF
 g = ( bgr & 0x00FF00 ) > > 8
 b = ( bgr & 0xFF0000 ) > > 16"""

implementation:
handle = self . NativeWindowHandle
if handle :
    return GetPixelColor ( x , y , handle )


signature:
def ToBitmap ( self , x : int = 0 , y : int = 0 , width : int = 0 , height : int = 0 ) -> Bitmap :
    """Capture control to a Bitmap object .
 x , y : int , the point in control ' s internal position ( from 0,0 ) .
 width , height : int , image ' s width and height from x , y , use 0 for entire area .
 If width ( or height ) < 0 , image size will be control ' s width ( or height ) - width ( or height ) ."""

implementation:
bitmap = Bitmap ( )
bitmap . FromControl ( self , x , y , width , height )
return bitmap


signature:
def CaptureToImage ( self , savePath : str , x : int = 0 , y : int = 0 , width : int = 0 , height : int = 0 ) -> bool :
    """Capture control to a image file .
 savePath : str , should end with . bmp , . jpg , . jpeg , . png , . gif , . tif , . tiff .
 x , y : int , the point in control ' s internal position ( from 0,0 ) .
 width , height : int , image ' s width and height from x , y , use 0 for entire area .
 If width ( or height ) < 0 , image size will be control ' s width ( or height ) - width ( or height ) .
 Return bool , True if succeed otherwise False ."""

implementation:
bitmap = Bitmap ( )
if bitmap . FromControl ( self , x , y , width , height ) :
    return bitmap . ToFile ( savePath )
return False


signature:
def IsTopLevel ( self ) -> bool :
    """Determine whether current control is top level ."""

implementation:
handle = self . NativeWindowHandle
if handle :
    return GetAncestor ( handle , GAFlag . Root ) == handle
return False


signature:
def GetTopLevelControl ( self ) -> 'Control' :
    """Get the top level control which current control lays .
 If current control is top level , return self .
 If current control is root control , return None .
 Return ` PaneControl ` or ` WindowControl ` or None ."""

implementation:
handle = self . NativeWindowHandle
if handle :
    topHandle = GetAncestor ( handle , GAFlag . Root )
    if topHandle :
        if topHandle == handle :
            return self
        else :
            return ControlFromHandle ( topHandle )
    else : # self is root control
        pass
else :
    control = self
    while True :
        control = control . GetParentControl ( )
        handle = control . NativeWindowHandle
        if handle :
            topHandle = GetAncestor ( handle , GAFlag . Root )
            return ControlFromHandle ( topHandle )


signature:
def Select ( self , itemName : str = '' , condition : Callable = None , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Show combobox ' s popup menu and select a item by name .
 itemName : str .
 condition : Callable function ( comboBoxItemName : str ) - > bool , if condition is valid , ignore itemName .
 waitTime : float .
 Some comboboxs doesn ' t support SelectionPattern , here is a workaround .
 This method tries to and selection support .
 It may not work for some comboboxes , such as comboboxes in older Qt version .
 If it doesn ' t work , you should write your own version Select , or it doesn ' t support selection at all ."""

implementation:
expandCollapsePattern = self . GetExpandCollapsePattern ( )
if expandCollapsePattern :
    expandCollapsePattern . Expand ( )
else : # Windows Form ' s ComboBoxControl doesn ' t support ExpandCollapsePattern
    self . Click ( x = - 10 , ratioY = 0.5 , simulateMove = False )
find = False
if condition :
    listItemControl = self . ListItemControl ( Compare = lambda c , d : condition ( c . Name ) )
else :
    listItemControl = self . ListItemControl ( Name = itemName )
if listItemControl . Exists ( 1 ) :
    scrollItemPattern = listItemControl . GetScrollItemPattern ( )
    if scrollItemPattern :
        scrollItemPattern . ScrollIntoView ( waitTime = 0.1 )
    listItemControl . Click ( waitTime = waitTime )
    find = True
else : # ComboBox ' s popup window is a child of root control
    listControl = ListControl ( searchDepth = 1 )
    if listControl . Exists ( 1 ) :
        if condition :
            listItemControl = self . ListItemControl ( Compare = lambda c , d : condition ( c . Name ) )
        else :
            listItemControl = self . ListItemControl ( Name = itemName )
        if listItemControl . Exists ( 0 , 0 ) :
            scrollItemPattern = listItemControl . GetScrollItemPattern ( )
            if scrollItemPattern :
                scrollItemPattern . ScrollIntoView ( waitTime = 0.1 )
            listItemControl . Click ( waitTime = waitTime )
            find = True
if not find :
    Logger . ColorfullyWriteLine ( 'Can\'t find <Color=Cyan>{}</Color> in ComboBoxControl or it does not support selection.' . format ( itemName ) , ConsoleColor . Yellow )
    if expandCollapsePattern :
        expandCollapsePattern . Collapse ( waitTime )
    else :
        self . Click ( x = - 10 , ratioY = 0.5 , simulateMove = False , waitTime = waitTime )
return find


signature:
def SetTopmost ( self , isTopmost : bool = True , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Set top level window topmost .
 isTopmost : bool .
 waitTime : float ."""

implementation:
if self . IsTopLevel ( ) :
    ret = SetWindowTopmost ( self . NativeWindowHandle , isTopmost )
    time . sleep ( waitTime )
    return ret
return False


signature:
def Maximize ( self , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Set top level window maximize ."""

implementation:
if self . IsTopLevel ( ) :
    return self . ShowWindow ( SW . ShowMaximized , waitTime )
return False


signature:
def MoveToCenter ( self ) -> bool :
    """Move window to screen center ."""

implementation:
if self . IsTopLevel ( ) :
    rect = self . BoundingRectangle
    screenWidth , screenHeight = GetScreenSize ( )
    x , y = ( screenWidth - rect . width ( ) ) // 2 , ( screenHeight - rect . height ( ) ) // 2
    if x < 0 :
        x = 0
    if y < 0 :
        y = 0
    return SetWindowPos ( self . NativeWindowHandle , SWP . HWND_Top , x , y , 0 , 0 , SWP . SWP_NoSize )
return False


signature:
def SetActive ( self , waitTime : float = OPERATION_WAIT_TIME ) -> bool :
    """Set top level window active ."""

implementation:
if self . IsTopLevel ( ) :
    handle = self . NativeWindowHandle
    if IsIconic ( handle ) :
        ret = ShowWindow ( handle , SW . Restore )
    elif not IsWindowVisible ( handle ) :
        ret = ShowWindow ( handle , SW . Show )
    ret = SetForegroundWindow ( handle )
    # may fail if foreground windows ' s process is not python
    time . sleep ( waitTime )
    return ret
return False


signature:
def MetroClose ( self , waitTime : float = OPERATION_WAIT_TIME ) -> None :
    """Only work on Windows 8/8.1 , if current window is Metro UI .
 waitTime : float ."""

implementation:
if self . ClassName == METRO_WINDOW_CLASS_NAME :
    screenWidth , screenHeight = GetScreenSize ( )
    MoveTo ( screenWidth // 2 , 0 , waitTime = 0 )
    DragDrop ( screenWidth // 2 , 0 , screenWidth // 2 , screenHeight , waitTime = waitTime )
else :
    Logger . WriteLine ( 'Window is not Metro!' , ConsoleColor . Yellow )


signature:
def DemoCN ( ) :
    """for Chinese language"""

implementation:
thisWindow = auto . GetConsoleWindow ( )
auto . Logger . ColorfullyWrite ( '我将运行<Color=Cyan>cmd</Color>并设置它的<Color=Cyan>屏幕缓冲区</Color>使<Color=Cyan>cmd</Color>一行能容纳很多字符\n\n' )
time . sleep ( 3 )
auto . SendKeys ( '{Win}r' )
while not isinstance ( auto . GetFocusedControl ( ) , auto . EditControl ) :
    time . sleep ( 1 )
auto . SendKeys ( 'cmd{Enter}' )
cmdWindow = auto . WindowControl ( RegexName = '.+cmd.exe' )
cmdWindow . TitleBarControl ( ) . RightClick ( )
auto . SendKey ( auto . Keys . VK_P )
optionWindow = cmdWindow . WindowControl ( SubName = '属性' )
optionWindow . TabItemControl ( SubName = '选项' ) . Click ( )
optionTab = optionWindow . PaneControl ( SubName = '选项' )
checkBox = optionTab . CheckBoxControl ( AutomationId = '103' )
if checkBox . GetTogglePattern ( ) . ToggleState != auto . ToggleState . On :
    checkBox . Click ( )
checkBox = optionTab . CheckBoxControl ( AutomationId = '104' )
if checkBox . GetTogglePattern ( ) . ToggleState != auto . ToggleState . On :
    checkBox . Click ( )
optionWindow . TabItemControl ( SubName = '布局' ) . Click ( )
layoutTab = optionWindow . PaneControl ( SubName = '布局' )
layoutTab . EditControl ( AutomationId = '301' ) . GetValuePattern ( ) . SetValue ( '300' )
layoutTab . EditControl ( AutomationId = '303' ) . GetValuePattern ( ) . SetValue ( '3000' )
layoutTab . EditControl ( AutomationId = '305' ) . GetValuePattern ( ) . SetValue ( '140' )
layoutTab . EditControl ( AutomationId = '307' ) . GetValuePattern ( ) . SetValue ( '30' )
optionWindow . ButtonControl ( AutomationId = '1' ) . Click ( )
cmdWindow . SetActive ( )
rect = cmdWindow . BoundingRectangle
auto . DragDrop ( rect . left + 50 , rect . top + 10 , 50 , 30 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( '我将运行<Color=Cyan>记事本</Color>并输入<Color=Cyan>Hello!!!</Color>\n\n' )
time . sleep ( 3 )
subprocess . Popen ( 'notepad' )
notepadWindow = auto . WindowControl ( searchDepth = 1 , ClassName = 'Notepad' )
cx , cy = auto . GetScreenSize ( )
notepadWindow . MoveWindow ( cx // 2 , 20 , cx // 2 , cy // 2 )
time . sleep ( 0.5 )
notepadWindow . EditControl ( ) . SendKeys ( 'Hello!!!' , 0.05 )
time . sleep ( 1 )
dir = os . path . dirname ( __file__ )
scriptPath = os . path . abspath ( os . path . join ( dir , '..\\automation.py' ) )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( '运行"<Color=Cyan>automation.py -h</Color>"显示帮助\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -h' . format ( scriptPath ) + '{Enter}' , 0.05 )
time . sleep ( 3 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( '运行"<Color=Cyan>automation.py -r -d1</Color>"显示所有顶层窗口, 即桌面的子窗口\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -r -d1 -t0' . format ( scriptPath ) + '{Enter}' , 0.05 )
time . sleep ( 3 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( '运行"<Color=Cyan>automation.py -c</Color>"显示鼠标光标下的控件\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -c -t3' . format ( scriptPath ) + '{Enter}' , 0.05 )
notepadWindow . SetActive ( )
notepadWindow . MoveCursorToMyCenter ( )
time . sleep ( 3 )
cmdWindow . SetActive ( waitTime = 2 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( '运行"<Color=Cyan>automation.py -a</Color>"显示鼠标光标下的控件和它的所有父控件\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -a -t3' . format ( scriptPath ) + '{Enter}' , 0.05 )
notepadWindow . SetActive ( )
notepadWindow . MoveCursorToMyCenter ( )
time . sleep ( 3 )
cmdWindow . SetActive ( waitTime = 2 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( '运行"<Color=Cyan>automation.py</Color>"显示当前激活窗口和它的所有子控件\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -t3' . format ( scriptPath ) + '{Enter}' , 0.05 )
notepadWindow . SetActive ( )
notepadWindow . EditControl ( ) . Click ( )
time . sleep ( 3 )
cmdWindow . SetActive ( waitTime = 2 )
time . sleep ( 3 )
thisWindow . SetActive ( )
auto . Logger . WriteLine ( '演示结束，按Enter退出' , auto . ConsoleColor . Green )
input ( )


signature:
def DemoEN ( ) :
    """for other language"""

implementation:
thisWindow = auto . GetConsoleWindow ( )
auto . Logger . ColorfullyWrite ( 'I will run <Color=Cyan>cmd</Color>\n\n' )
time . sleep ( 3 )
auto . SendKeys ( '{Win}r' )
while not isinstance ( auto . GetFocusedControl ( ) , auto . EditControl ) :
    time . sleep ( 1 )
auto . SendKeys ( 'cmd{Enter}' )
cmdWindow = auto . WindowControl ( SubName = 'cmd.exe' )
rect = cmdWindow . BoundingRectangle
auto . DragDrop ( rect . left + 50 , rect . top + 10 , 50 , 10 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( 'I will run <Color=Cyan>Notepad</Color> and type <Color=Cyan>Hello!!!</Color>\n\n' )
time . sleep ( 3 )
subprocess . Popen ( 'notepad' )
notepadWindow = auto . WindowControl ( searchDepth = 1 , ClassName = 'Notepad' )
cx , cy = auto . GetScreenSize ( )
notepadWindow . MoveWindow ( cx // 2 , 20 , cx // 2 , cy // 2 )
time . sleep ( 0.5 )
notepadWindow . EditControl ( ) . SendKeys ( 'Hello!!!' , 0.05 )
time . sleep ( 1 )
dir = os . path . dirname ( __file__ )
scriptPath = os . path . abspath ( os . path . join ( dir , '..\\automation.py' ) )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( 'run "<Color=Cyan>automation.py -h</Color>" to display the help\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -h' . format ( scriptPath ) + '{Enter}' , 0.05 )
time . sleep ( 3 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( 'run "<Color=Cyan>automation.py -r -d1</Color>" to display the top level windows, desktop\'s children\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -r -d1 -t0' . format ( scriptPath ) + '{Enter}' , 0.05 )
time . sleep ( 3 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( 'run "<Color=Cyan>automation.py -c</Color>" to display the control under mouse cursor\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -c -t3' . format ( scriptPath ) + '{Enter}' , 0.05 )
notepadWindow . SetActive ( )
notepadWindow . MoveCursorToMyCenter ( )
time . sleep ( 3 )
cmdWindow . SetActive ( waitTime = 2 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( 'run "<Color=Cyan>automation.py -a</Color>" to display the control under mouse cursor and its ancestors\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -a -t3' . format ( scriptPath ) + '{Enter}' , 0.05 )
notepadWindow . SetActive ( )
notepadWindow . MoveCursorToMyCenter ( )
time . sleep ( 3 )
cmdWindow . SetActive ( waitTime = 2 )
thisWindow . SetActive ( )
auto . Logger . ColorfullyWrite ( 'run "<Color=Cyan>automation.py</Color>" to display the active window\n\n' )
time . sleep ( 3 )
cmdWindow . SendKeys ( '"{}" -t3' . format ( scriptPath ) + '{Enter}' , 0.05 )
notepadWindow . SetActive ( )
notepadWindow . EditControl ( ) . Click ( )
time . sleep ( 3 )
cmdWindow . SetActive ( waitTime = 2 )
time . sleep ( 3 )
thisWindow . SetActive ( )
auto . Logger . WriteLine ( 'press Enter to exit' , auto . ConsoleColor . Green )
input ( )


signature:
def threadFunc ( root ) :
    """If you want to use functionalities related to Controls and Patterns in a new thread .
 You must call InitializeUIAutomationInCurrentThread first in the thread
 and call UninitializeUIAutomationInCurrentThread when the thread exits .
 But you can ' t use use a Control or a Pattern created in a different thread .
 So you can ' t create a Control or a Pattern in main thread and then pass it to a new thread and use it ."""

implementation:
# print ( root ) # you cannot use root because it is root control created in main thread
th = threading . currentThread ( )
auto . Logger . WriteLine ( '\nThis is running in a new thread. {} {}' . format ( th . ident , th . name ) , auto . ConsoleColor . Cyan )
time . sleep ( 2 )
auto . InitializeUIAutomationInCurrentThread ( )
auto . GetConsoleWindow ( ) . CaptureToImage ( 'console_newthread.png' )
newRoot = auto . GetRootControl ( )
# ok , root control created in new thread
auto . EnumAndLogControl ( newRoot , 1 )
auto . UninitializeUIAutomationInCurrentThread ( )
auto . Logger . WriteLine ( '\nThread exits. {} {}' . format ( th . ident , th . name ) , auto . ConsoleColor . Cyan )


signature:
def _saliency_map ( self , a , image , target , labels , mask , fast = False ) :
    """Implements Algorithm 3 in manuscript"""

implementation:
# pixel influence on target class
alphas = a . gradient ( image , target ) * mask
# pixel influence on sum of residual classes
# ( don ' t evaluate if fast = = True )
if fast :
    betas = - np . ones_like ( alphas )
else :
    betas = np . sum ( [ a . gradient ( image , label ) * mask - alphas for label in labels ] , 0 )
# compute saliency map
# ( take into account both pos . & neg . perturbations )
salmap = np . abs ( alphas ) * np . abs ( betas ) * np . sign ( alphas * betas )
# find optimal pixel & direction of perturbation
idx = np . argmin ( salmap )
idx = np . unravel_index ( idx , mask . shape )
pix_sign = np . sign ( alphas ) [ idx ]
return idx , pix_sign


signature:
def from_keras ( cls , model , bounds , input_shape = None , channel_axis = 3 , preprocessing = ( 0 , 1 ) ) :
    """Alternative constructor for a TensorFlowModel that
 accepts a ` tf . keras . Model ` instance .
 Parameters
 model : ` tensorflow . keras . Model `
 A ` tensorflow . keras . Model ` that accepts a single input tensor
 and returns a single output tensor representing logits .
 bounds : tuple
 Tuple of lower and upper bound for the pixel values , usually
 (0 , 1 ) or ( 0 , 255 ) .
 input _ shape : tuple
 The shape of a single input , e . g . ( 28 , 28 , 1 ) for MNIST .
 If None , tries to get the the shape from the model ' s
 input _ shape attribute .
 channel _ axis : int
 The index of the axis that represents color channels .
 preprocessing : 2 - element tuple with floats or numpy arrays
 Elementwises preprocessing of input ; we first subtract the first
 element of preprocessing from the input and then divide the input
 by the second element ."""

implementation:
import tensorflow as tf
if input_shape is None :
    try :
        input_shape = model . input_shape [ 1 : ]
    except AttributeError :
        raise ValueError ( 'Please specify input_shape manually or ' 'provide a model with an input_shape attribute' )
with tf . keras . backend . get_session ( ) . as_default ( ) :
    inputs = tf . placeholder ( tf . float32 , ( None , ) + input_shape )
    logits = model ( inputs )
    return cls ( inputs , logits , bounds = bounds , channel_axis = channel_axis , preprocessing = preprocessing )


signature:
def normalized_distance ( self , image ) :
    """Calculates the distance of a given image to the
 original image .
 Parameters
 image : ` numpy . ndarray `
 The image that should be compared to the original image .
 Returns
 : class : ` Distance `
 The distance between the given image and the original image ."""

implementation: return self . __distance ( self . __original_image_for_distance , image , bounds = self . bounds ( ) )


signature:
def __is_adversarial ( self , image , predictions , in_bounds ) :
    """Interface to criterion . is _ adverarial that calls
 _ _ new _ adversarial if necessary .
 Parameters
 predictions : : class : ` numpy . ndarray `
 A vector with the pre - softmax predictions for some image .
 label : int
 The label of the unperturbed reference image ."""

implementation:
is_adversarial = self . __criterion . is_adversarial ( predictions , self . __original_class )
assert isinstance ( is_adversarial , bool ) or isinstance ( is_adversarial , np . bool_ )
if is_adversarial :
    is_best , distance = self . __new_adversarial ( image , predictions , in_bounds )
else :
    is_best = False
    distance = None
return is_adversarial , is_best , distance


signature:
def channel_axis ( self , batch ) :
    """Interface to model . channel _ axis for attacks .
 Parameters
 batch : bool
 Controls whether the index of the axis for a batch of images
 (4 dimensions ) or a single image ( 3 dimensions ) should be returned ."""

implementation:
axis = self . __model . channel_axis ( )
if not batch :
    axis = axis - 1
return axis


signature:
def has_gradient ( self ) :
    """Returns true if _ backward and _ forward _ backward can be called
 by an attack , False otherwise ."""

implementation:
try :
    self . __model . gradient
    self . __model . predictions_and_gradient
except AttributeError :
    return False
else :
    return True


signature:
def predictions ( self , image , strict = True , return_details = False ) :
    """Interface to model . predictions for attacks .
 Parameters
 image : ` numpy . ndarray `
 Single input with shape as expected by the model
 ( without the batch dimension ) .
 strict : bool
 Controls if the bounds for the pixel values should be checked ."""

implementation:
in_bounds = self . in_bounds ( image )
assert not strict or in_bounds
self . _total_prediction_calls += 1
predictions = self . __model . predictions ( image )
is_adversarial , is_best , distance = self . __is_adversarial ( image , predictions , in_bounds )
assert predictions . ndim == 1
if return_details :
    return predictions , is_adversarial , is_best , distance
else :
    return predictions , is_adversarial


signature:
def batch_predictions ( self , images , greedy = False , strict = True , return_details = False ) :
    """Interface to model . batch _ predictions for attacks .
 Parameters
 images : ` numpy . ndarray `
 Batch of inputs with shape as expected by the model .
 greedy : bool
 Whether the first adversarial should be returned .
 strict : bool
 Controls if the bounds for the pixel values should be checked ."""

implementation:
if strict :
    in_bounds = self . in_bounds ( images )
    assert in_bounds
self . _total_prediction_calls += len ( images )
predictions = self . __model . batch_predictions ( images )
assert predictions . ndim == 2
assert predictions . shape [ 0 ] == images . shape [ 0 ]
if return_details :
    assert greedy
adversarials = [ ]
for i in range ( len ( predictions ) ) :
    if strict :
        in_bounds_i = True
    else :
        in_bounds_i = self . in_bounds ( images [ i ] )
    is_adversarial , is_best , distance = self . __is_adversarial ( images [ i ] , predictions [ i ] , in_bounds_i )
    if is_adversarial and greedy :
        if return_details :
            return predictions , is_adversarial , i , is_best , distance
        else :
            return predictions , is_adversarial , i
    adversarials . append ( is_adversarial )
if greedy : # pragma : no cover
# no adversarial found
    if return_details :
        return predictions , False , None , False , None
    else :
        return predictions , False , None
is_adversarial = np . array ( adversarials )
assert is_adversarial . ndim == 1
assert is_adversarial . shape [ 0 ] == images . shape [ 0 ]
return predictions , is_adversarial


signature:
def gradient ( self , image = None , label = None , strict = True ) :
    """Interface to model . gradient for attacks .
 Parameters
 image : ` numpy . ndarray `
 Single input with shape as expected by the model
 ( without the batch dimension ) .
 Defaults to the original image .
 label : int
 Label used to calculate the loss that is differentiated .
 Defaults to the original label .
 strict : bool
 Controls if the bounds for the pixel values should be checked ."""

implementation:
assert self . has_gradient ( )
if image is None :
    image = self . __original_image
if label is None :
    label = self . __original_class
assert not strict or self . in_bounds ( image )
self . _total_gradient_calls += 1
gradient = self . __model . gradient ( image , label )
assert gradient . shape == image . shape
return gradient


signature:
def predictions_and_gradient ( self , image = None , label = None , strict = True , return_details = False ) :
    """Interface to model . predictions _ and _ gradient for attacks .
 Parameters
 image : ` numpy . ndarray `
 Single input with shape as expected by the model
 ( without the batch dimension ) .
 Defaults to the original image .
 label : int
 Label used to calculate the loss that is differentiated .
 Defaults to the original label .
 strict : bool
 Controls if the bounds for the pixel values should be checked ."""

implementation:
assert self . has_gradient ( )
if image is None :
    image = self . __original_image
if label is None :
    label = self . __original_class
in_bounds = self . in_bounds ( image )
assert not strict or in_bounds
self . _total_prediction_calls += 1
self . _total_gradient_calls += 1
predictions , gradient = self . __model . predictions_and_gradient ( image , label )
# noqa : E501
is_adversarial , is_best , distance = self . __is_adversarial ( image , predictions , in_bounds )
assert predictions . ndim == 1
assert gradient . shape == image . shape
if return_details :
    return predictions , gradient , is_adversarial , is_best , distance
else :
    return predictions , gradient , is_adversarial


signature:
def backward ( self , gradient , image = None , strict = True ) :
    """Interface to model . backward for attacks .
 Parameters
 gradient : ` numpy . ndarray `
 Gradient of some loss w . r . t . the logits .
 image : ` numpy . ndarray `
 Single input with shape as expected by the model
 ( without the batch dimension ) .
 Returns
 gradient : ` numpy . ndarray `
 The gradient w . r . t the image .
 See Also
 : meth : ` gradient `"""

implementation:
assert self . has_gradient ( )
assert gradient . ndim == 1
if image is None :
    image = self . __original_image
assert not strict or self . in_bounds ( image )
self . _total_gradient_calls += 1
gradient = self . __model . backward ( gradient , image )
assert gradient . shape == image . shape
return gradient


signature:
def loss_function ( cls , const , a , x , logits , reconstructed_original , confidence , min_ , max_ ) :
    """Returns the loss and the gradient of the loss w . r . t . x ,
 assuming that logits = model ( x ) ."""

implementation:
targeted = a . target_class ( ) is not None
if targeted :
    c_minimize = cls . best_other_class ( logits , a . target_class ( ) )
    c_maximize = a . target_class ( )
else :
    c_minimize = a . original_class
    c_maximize = cls . best_other_class ( logits , a . original_class )
is_adv_loss = logits [ c_minimize ] - logits [ c_maximize ]
# is _ adv is True as soon as the is _ adv _ loss goes below 0
# but sometimes we want additional confidence
is_adv_loss += confidence
is_adv_loss = max ( 0 , is_adv_loss )
s = max_ - min_
squared_l2_distance = np . sum ( ( x - reconstructed_original ) ** 2 ) / s ** 2
total_loss = squared_l2_distance + const * is_adv_loss
# calculate the gradient of total _ loss w . r . t . x
logits_diff_grad = np . zeros_like ( logits )
logits_diff_grad [ c_minimize ] = 1
logits_diff_grad [ c_maximize ] = - 1
is_adv_loss_grad = a . backward ( logits_diff_grad , x )
assert is_adv_loss >= 0
if is_adv_loss == 0 :
    is_adv_loss_grad = 0
squared_l2_distance_grad = ( 2 / s ** 2 ) * ( x - reconstructed_original )
total_loss_grad = squared_l2_distance_grad + const * is_adv_loss_grad
return total_loss , total_loss_grad


signature:
def best_other_class ( logits , exclude ) :
    """Returns the index of the largest logit , ignoring the class that
 is passed as ` exclude ` ."""

implementation:
other_logits = logits - onehot_like ( logits , exclude , value = np . inf )
return np . argmax ( other_logits )


signature:
def name ( self ) :
    """Concatenates the names of the given criteria in alphabetical order .
 If a sub - criterion is itself a combined criterion , its name is
 first split into the individual names and the names of the
 sub - sub criteria is used instead of the name of the sub - criterion .
 This is done recursively to ensure that the order and the hierarchy
 of the criteria does not influence the name .
 Returns
 str
 The alphabetically sorted names of the sub - criteria concatenated
 using double underscores between them ."""

implementation:
names = ( criterion . name ( ) for criterion in self . _criteria )
return '__' . join ( sorted ( names ) )


signature:
def _difference_map ( image , color_axis ) :
    """Difference map of the image .
 Approximate derivatives of the function image [ c , : , : ]
 ( e . g . PyTorch ) or image [ : , : , c ] ( e . g . Keras ) .
 dfdx , dfdy = difference _ map ( image )
 In :
 image : numpy . ndarray
 of shape C x h x w or h x w x C , with C = 1 or C = 3
 ( color channels ) , h , w > = 3 , and [ type ] is ' Float ' or
 ' Double ' . Contains the values of functions f _ b :
 R ^ 2 - > R ^ C , b = 1 , . . . , B , on the grid
 {0 , . . . , h - 1 } x { 0 , . . . , w - 1 } .
 Out :
 dfdx : numpy . ndarray
 dfdy : numpy . ndarray
 of shape C x h x w or h x w x C contain the x and
 y derivatives of f at the points on the grid ,
 approximated by central differences ( except on
 boundaries ) :
 For c = 0 , . . . , C , i = 1 , . . . , h - 2,
 j = 1 , . . . , w - 2.
 e . g . for shape = c x h x w :
 dfdx [ c , i , j ] = ( image [ c , i , j + 1 ] -
 image [ c , i , j - 1 ] ) / 2
 dfdx [ c , i , j ] = ( image [ c , i + 1 , j ] -
 image [ c , i - 1 , j ] ) / 2
 positive x - direction is along rows from left to right .
 positive y - direction is along columns from above to below ."""

implementation:
if color_axis == 2 :
    image = _transpose_image ( image )
# Derivative in x direction ( rows from left to right )
dfdx = np . zeros_like ( image )
# forward difference in first column
dfdx [ : , : , 0 ] = image [ : , : , 1 ] - image [ : , : , 0 ]
# backwards difference in last column
dfdx [ : , : , - 1 ] = image [ : , : , - 1 ] - image [ : , : , - 2 ]
# central difference elsewhere
dfdx [ : , : , 1 : - 1 ] = 0.5 * ( image [ : , : , 2 : ] - image [ : , : , : - 2 ] )
# Derivative in y direction ( columns from above to below )
dfdy = np . zeros_like ( image )
# forward difference in first row
dfdy [ : , 0 , : ] = image [ : , 1 , : ] - image [ : , 0 , : ]
# backwards difference in last row
dfdy [ : , - 1 , : ] = image [ : , - 1 , : ] - image [ : , - 2 , : ]
# central difference elsewhere
dfdy [ : , 1 : - 1 , : ] = 0.5 * ( image [ : , 2 : , : ] - image [ : , : - 2 , : ] )
return dfdx , dfdy


signature:
def _compose ( image , vec_field , color_axis ) :
    """Calculate the composition of the function image with the vector
 field vec _ field by interpolation .
 new _ func = compose ( image , vec _ field )
 In :
 image : numpy . ndarray
 of shape C x h x w with C = 3 or C = 1 ( color channels ) ,
 h , w > = 2 , and [ type ] = ' Float ' or ' Double ' .
 Contains the values of a function f : R ^ 2 - > R ^ C
 on the grid { 0 , . . . , h - 1 } x { 0 , . . . , w - 1 } .
 vec _ field : numpy . array
 of shape ( h , w , 2)
 vec _ field [ y , x , 0 ] is the x - coordinate of the vector vec _ field [ y , x ]
 vec _ field [ y , x , 1 ] is the y - coordinate of the vector vec _ field [ y , x ]
 positive x - direction is along rows from left to right
 positive y - direction is along columns from above to below"""

implementation:
if color_axis == 2 :
    image = _transpose_image ( image )
c , h , w = image . shape
# colors , height , width
hrange = np . arange ( h )
wrange = np . arange ( w )
MGx , MGy = np . meshgrid ( wrange , hrange )
defMGx = ( MGx + vec_field [ : , : , 0 ] ) . clip ( 0 , w - 1 )
defMGy = ( MGy + vec_field [ : , : , 1 ] ) . clip ( 0 , h - 1 )
new_image = np . empty_like ( image )
for channel in range ( c ) : # Get a linear interpolation for this color channel .
    interpolation = RectBivariateSpline ( hrange , wrange , image [ channel ] , kx = 1 , ky = 1 )
    # grid = False since the deformed grid is irregular
    new_image [ channel ] = interpolation ( defMGy , defMGx , grid = False )
if color_axis == 2 :
    return _re_transpose_image ( new_image )
else :
    return new_image


signature:
def _create_vec_field ( fval , gradf , d1x , d2x , color_axis , smooth = 0 ) :
    """Calculate the deformation vector field
 In :
 fval : float
 gradf : numpy . ndarray
 of shape C x h x w with C = 3 or C = 1
 ( color channels ) , h , w > = 1.
 d1x : numpy . ndarray
 of shape C x h x w and [ type ] = ' Float ' or ' Double ' .
 d2x : numpy . ndarray
 of shape C x h x w and [ type ] = ' Float ' or ' Double ' .
 smooth : float
 Width of the Gaussian kernel used for smoothing
 ( default is 0 for no smoothing ) .
 Out :
 vec _ field : numpy . ndarray
 of shape ( 2 , h , w ) ."""

implementation:
if color_axis == 2 :
    gradf = _transpose_image ( gradf )
c , h , w = gradf . shape
# colors , height , width
# Sum over color channels
alpha1 = np . sum ( gradf * d1x , axis = 0 )
alpha2 = np . sum ( gradf * d2x , axis = 0 )
norm_squared_alpha = ( alpha1 ** 2 ) . sum ( ) + ( alpha2 ** 2 ) . sum ( )
# Smoothing
if smooth > 0 :
    alpha1 = gaussian_filter ( alpha1 , smooth )
    alpha2 = gaussian_filter ( alpha2 , smooth )
    norm_squared_alpha = ( alpha1 ** 2 ) . sum ( ) + ( alpha2 ** 2 ) . sum ( )
    # In theory , we need to apply the filter a second time .
    alpha1 = gaussian_filter ( alpha1 , smooth )
    alpha2 = gaussian_filter ( alpha2 , smooth )
vec_field = np . empty ( ( h , w , 2 ) )
vec_field [ : , : , 0 ] = - fval * alpha1 / norm_squared_alpha
vec_field [ : , : , 1 ] = - fval * alpha2 / norm_squared_alpha
return vec_field


signature:
def softmax ( logits ) :
    """Transforms predictions into probability values .
 Parameters
 logits : array _ like
 The logits predicted by the model .
 Returns
 ` numpy . ndarray `
 Probability values corresponding to the logits ."""

implementation:
assert logits . ndim == 1
# for numerical reasons we subtract the max logit
# ( mathematically it doesn ' t matter ! )
# otherwise exp ( logits ) might become too large or too small
logits = logits - np . max ( logits )
e = np . exp ( logits )
return e / np . sum ( e )


signature:
def crossentropy ( label , logits ) :
    """Calculates the cross - entropy .
 Parameters
 logits : array _ like
 The logits predicted by the model .
 label : int
 The label describing the target distribution .
 Returns
 float
 The cross - entropy between softmax ( logits ) and onehot ( label ) ."""

implementation:
assert logits . ndim == 1
# for numerical reasons we subtract the max logit
# ( mathematically it doesn ' t matter ! )
# otherwise exp ( logits ) might become too large or too small
logits = logits - np . max ( logits )
e = np . exp ( logits )
s = np . sum ( e )
ce = np . log ( s ) - logits [ label ]
return ce


signature:
def batch_crossentropy ( label , logits ) :
    """Calculates the cross - entropy for a batch of logits .
 Parameters
 logits : array _ like
 The logits predicted by the model for a batch of inputs .
 label : int
 The label describing the target distribution .
 Returns
 np . ndarray
 The cross - entropy between softmax ( logits [ i ] ) and onehot ( label )
 for all i ."""

implementation:
assert logits . ndim == 2
# for numerical reasons we subtract the max logit
# ( mathematically it doesn ' t matter ! )
# otherwise exp ( logits ) might become too large or too small
logits = logits - np . max ( logits , axis = 1 , keepdims = True )
e = np . exp ( logits )
s = np . sum ( e , axis = 1 )
ces = np . log ( s ) - logits [ : , label ]
return ces


signature:
def binarize ( x , values , threshold = None , included_in = 'upper' ) :
    """Binarizes the values of x .
 Parameters
 values : tuple of two floats
 The lower and upper value to which the inputs are mapped .
 threshold : float
 The threshold ; defaults to ( values [ 0 ] + values [ 1 ] ) / 2 if None .
 included _ in : str
 Whether the threshold value itself belongs to the lower or
 upper interval ."""

implementation:
lower , upper = values
if threshold is None :
    threshold = ( lower + upper ) / 2.
x = x . copy ( )
if included_in == 'lower' :
    x [ x <= threshold ] = lower
    x [ x > threshold ] = upper
elif included_in == 'upper' :
    x [ x < threshold ] = lower
    x [ x >= threshold ] = upper
else :
    raise ValueError ( 'included_in must be "lower" or "upper"' )
return x


signature:
def imagenet_example ( shape = ( 224 , 224 ) , data_format = 'channels_last' ) :
    """Returns an example image and its imagenet class label .
 Parameters
 shape : list of integers
 The shape of the returned image .
 data _ format : str
 " channels _ first " or " channels _ last "
 Returns
 image : array _ like
 The example image .
 label : int
 The imagenet label associated with the image .
 NOTE : This function is deprecated and will be removed in the future ."""

implementation:
assert len ( shape ) == 2
assert data_format in [ 'channels_first' , 'channels_last' ]
from PIL import Image
path = os . path . join ( os . path . dirname ( __file__ ) , 'example.png' )
image = Image . open ( path )
image = image . resize ( shape )
image = np . asarray ( image , dtype = np . float32 )
image = image [ : , : , : 3 ]
assert image . shape == shape + ( 3 , )
if data_format == 'channels_first' :
    image = np . transpose ( image , ( 2 , 0 , 1 ) )
return image , 282


signature:
def samples ( dataset = 'imagenet' , index = 0 , batchsize = 1 , shape = ( 224 , 224 ) , data_format = 'channels_last' ) :
    '''Returns a batch of example images and the corresponding labels
 Parameters
 dataset : string
 The data set to load ( options : imagenet , mnist , cifar10,
 cifar100 , fashionMNIST )
 index : int
 For each data set 20 example images exist . The returned batch
 contains the images with index [ index , index + 1 , index + 2 , . . . ]
 batchsize : int
 Size of batch .
 shape : list of integers
 The shape of the returned image ( only relevant for Imagenet ) .
 data _ format : str
 " channels _ first " or " channels _ last "
 Returns
 images : array _ like
 The batch of example images
 labels : array of int
 The labels associated with the images .'''

implementation:
from PIL import Image
images , labels = [ ] , [ ]
basepath = os . path . dirname ( __file__ )
samplepath = os . path . join ( basepath , 'data' )
files = os . listdir ( samplepath )
for idx in range ( index , index + batchsize ) :
    i = idx % 20
    # get filename and label
    file = [ n for n in files if '{}_{:02d}_' . format ( dataset , i ) in n ] [ 0 ]
    label = int ( file . split ( '.' ) [ 0 ] . split ( '_' ) [ - 1 ] )
    # open file
    path = os . path . join ( samplepath , file )
    image = Image . open ( path )
    if dataset == 'imagenet' :
        image = image . resize ( shape )
    image = np . asarray ( image , dtype = np . float32 )
    if dataset != 'mnist' and data_format == 'channels_first' :
        image = np . transpose ( image , ( 2 , 0 , 1 ) )
    images . append ( image )
    labels . append ( label )
labels = np . array ( labels )
images = np . stack ( images )
return images , labels


signature:
def onehot_like ( a , index , value = 1 ) :
    """Creates an array like a , with all values
 set to 0 except one .
 Parameters
 a : array _ like
 The returned one - hot array will have the same shape
 and dtype as this array
 index : int
 The index that should be set to ` value `
 value : single value compatible with a . dtype
 The value to set at the given index
 Returns
 ` numpy . ndarray `
 One - hot array with the given value at the given
 location and zeros everywhere else ."""

implementation:
x = np . zeros_like ( a )
x [ index ] = value
return x


signature:
def _get_output ( self , a , image ) :
    """Looks up the precomputed adversarial image for a given image ."""

implementation:
sd = np . square ( self . _input_images - image )
mses = np . mean ( sd , axis = tuple ( range ( 1 , sd . ndim ) ) )
index = np . argmin ( mses )
# if we run into numerical problems with this approach , we might
# need to add a very tiny threshold here
if mses [ index ] > 0 :
    raise ValueError ( 'No precomputed output image for this image' )
return self . _output_images [ index ]


signature:
def _process_gradient ( self , backward , dmdp ) :
    """backward : ` callable `
 callable that backpropagates the gradient of the model w . r . t to
 preprocessed input through the preprocessing to get the gradient
 of the model ' s output w . r . t . the input before preprocessing
 dmdp : gradient of model w . r . t . preprocessed input"""

implementation:
if backward is None : # pragma : no cover
    raise ValueError ( 'Your preprocessing function does not provide' ' an (approximate) gradient' )
dmdx = backward ( dmdp )
assert dmdx . dtype == dmdp . dtype
return dmdx


signature:
def predictions ( self , image ) :
    """Convenience method that calculates predictions for a single image .
 Parameters
 image : ` numpy . ndarray `
 Single input with shape as expected by the model
 ( without the batch dimension ) .
 Returns
 ` numpy . ndarray `
 Vector of predictions ( logits , i . e . before the softmax ) with
 shape ( number of classes , ) .
 See Also
 : meth : ` batch _ predictions `"""

implementation: return np . squeeze ( self . batch_predictions ( image [ np . newaxis ] ) , axis = 0 )


signature:
def gradient ( self , image , label ) :
    """Calculates the gradient of the cross - entropy loss w . r . t . the image .
 The default implementation calls predictions _ and _ gradient .
 Subclasses can provide more efficient implementations that
 only calculate the gradient .
 Parameters
 image : ` numpy . ndarray `
 Single input with shape as expected by the model
 ( without the batch dimension ) .
 label : int
 Reference label used to calculate the gradient .
 Returns
 gradient : ` numpy . ndarray `
 The gradient of the cross - entropy loss w . r . t . the image . Will
 have the same shape as the image .
 See Also
 : meth : ` gradient `"""

implementation:
_ , gradient = self . predictions_and_gradient ( image , label )
return gradient


signature:
def clone ( git_uri ) :
    """Clone a remote git repository to a local path .
 : param git _ uri : the URI to the git repository to be cloned
 : return : the generated local path where the repository has been cloned to"""

implementation:
hash_digest = sha256_hash ( git_uri )
local_path = home_directory_path ( FOLDER , hash_digest )
exists_locally = path_exists ( local_path )
if not exists_locally :
    _clone_repo ( git_uri , local_path )
else :
    logging . info ( # pragma : no cover
    "Git repository already exists locally." )
# pragma : no cover
return local_path


signature:
def run ( command , parser , cl_args , unknown_args ) :
    ''': param command :
 : param parser :
 : param cl _ args :
 : param unknown _ args :
 : return :'''

implementation:
Log . debug ( "Deactivate Args: %s" , cl_args )
return cli_helper . run ( command , cl_args , "deactivate topology" )


signature:
def poll ( self , timeout = 0.0 ) :
    """Modified version of poll ( ) from asyncore module"""

implementation:
if self . sock_map is None :
    Log . warning ( "Socket map is not registered to Gateway Looper" )
readable_lst = [ ]
writable_lst = [ ]
error_lst = [ ]
if self . sock_map is not None :
    for fd , obj in self . sock_map . items ( ) :
        is_r = obj . readable ( )
        is_w = obj . writable ( )
        if is_r :
            readable_lst . append ( fd )
        if is_w and not obj . accepting :
            writable_lst . append ( fd )
        if is_r or is_w :
            error_lst . append ( fd )
# Add wakeup fd
readable_lst . append ( self . pipe_r )
Log . debug ( "Will select() with timeout: " + str ( timeout ) + ", with map: " + str ( self . sock_map ) )
try :
    readable_lst , writable_lst , error_lst = select . select ( readable_lst , writable_lst , error_lst , timeout )
except select . error as err :
    Log . debug ( "Trivial error: " + str ( err ) )
    if err . args [ 0 ] != errno . EINTR :
        raise
    else :
        return
Log . debug ( "Selected [r]: " + str ( readable_lst ) + " [w]: " + str ( writable_lst ) + " [e]: " + str ( error_lst ) )
if self . pipe_r in readable_lst :
    Log . debug ( "Read from pipe" )
    os . read ( self . pipe_r , 1024 )
    readable_lst . remove ( self . pipe_r )
if self . sock_map is not None :
    for fd in readable_lst :
        obj = self . sock_map . get ( fd )
        if obj is None :
            continue
        asyncore . read ( obj )
    for fd in writable_lst :
        obj = self . sock_map . get ( fd )
        if obj is None :
            continue
        asyncore . write ( obj )
    for fd in error_lst :
        obj = self . sock_map . get ( fd )
        if obj is None :
            continue
        # pylint : disable = W0212
        asyncore . _exception ( obj )


signature:
def configure ( level , logfile = None ) :
    """configure logging"""

implementation:
log_format = "%(asctime)s-%(levelname)s: %(message)s"
date_format = '%a, %d %b %Y %H:%M:%S'
logging . basicConfig ( format = log_format , datefmt = date_format )
Log . setLevel ( level )
if logfile is not None :
    fh = logging . FileHandler ( logfile )
    fh . setFormatter ( logging . Formatter ( log_format ) )
    Log . addHandler ( fh )


signature:
def write_success_response ( self , result ) :
    """Result may be a python dictionary , array or a primitive type
 that can be converted to JSON for writing back the result ."""

implementation:
response = self . make_success_response ( result )
now = time . time ( )
spent = now - self . basehandler_starttime
response [ constants . RESPONSE_KEY_EXECUTION_TIME ] = spent
self . write_json_response ( response )


signature:
def write_error_response ( self , message ) :
    """Writes the message as part of the response and sets 404 status ."""

implementation:
self . set_status ( 404 )
response = self . make_error_response ( str ( message ) )
now = time . time ( )
spent = now - self . basehandler_starttime
response [ constants . RESPONSE_KEY_EXECUTION_TIME ] = spent
self . write_json_response ( response )


signature:
def write_json_response ( self , response ) :
    """write back json response"""

implementation:
self . write ( tornado . escape . json_encode ( response ) )
self . set_header ( "Content-Type" , "application/json" )


signature:
def make_response ( self , status ) :
    """Makes the base dict for the response .
 The status is the string value for
 the key " status " of the response . This
 should be " success " or " failure " ."""

implementation:
response = { constants . RESPONSE_KEY_STATUS : status , constants . RESPONSE_KEY_VERSION : constants . API_VERSION , constants . RESPONSE_KEY_EXECUTION_TIME : 0 , constants . RESPONSE_KEY_MESSAGE : "" , }
return response


signature:
def make_success_response ( self , result ) :
    """Makes the python dict corresponding to the
 JSON that needs to be sent for a successful
 response . Result is the actual payload
 that gets sent ."""

implementation:
response = self . make_response ( constants . RESPONSE_STATUS_SUCCESS )
response [ constants . RESPONSE_KEY_RESULT ] = result
return response


signature:
def make_error_response ( self , message ) :
    """Makes the python dict corresponding to the
 JSON that needs to be sent for a failed
 response . Message is the message that is
 sent as the reason for failure ."""

implementation:
response = self . make_response ( constants . RESPONSE_STATUS_FAILURE )
response [ constants . RESPONSE_KEY_MESSAGE ] = message
return response


signature:
def get_argument_cluster ( self ) :
    """Helper function to get request argument .
 Raises exception if argument is missing .
 Returns the cluster argument ."""

implementation:
try :
    return self . get_argument ( constants . PARAM_CLUSTER )
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_role ( self ) :
    """Helper function to get request argument .
 Raises exception if argument is missing .
 Returns the role argument ."""

implementation:
try :
    return self . get_argument ( constants . PARAM_ROLE , default = None )
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_environ ( self ) :
    """Helper function to get request argument .
 Raises exception if argument is missing .
 Returns the environ argument ."""

implementation:
try :
    return self . get_argument ( constants . PARAM_ENVIRON )
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_topology ( self ) :
    """Helper function to get topology argument .
 Raises exception if argument is missing .
 Returns the topology argument ."""

implementation:
try :
    topology = self . get_argument ( constants . PARAM_TOPOLOGY )
    return topology
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_component ( self ) :
    """Helper function to get component argument .
 Raises exception if argument is missing .
 Returns the component argument ."""

implementation:
try :
    component = self . get_argument ( constants . PARAM_COMPONENT )
    return component
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_instance ( self ) :
    """Helper function to get instance argument .
 Raises exception if argument is missing .
 Returns the instance argument ."""

implementation:
try :
    instance = self . get_argument ( constants . PARAM_INSTANCE )
    return instance
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_starttime ( self ) :
    """Helper function to get starttime argument .
 Raises exception if argument is missing .
 Returns the starttime argument ."""

implementation:
try :
    starttime = self . get_argument ( constants . PARAM_STARTTIME )
    return starttime
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_endtime ( self ) :
    """Helper function to get endtime argument .
 Raises exception if argument is missing .
 Returns the endtime argument ."""

implementation:
try :
    endtime = self . get_argument ( constants . PARAM_ENDTIME )
    return endtime
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_query ( self ) :
    """Helper function to get query argument .
 Raises exception if argument is missing .
 Returns the query argument ."""

implementation:
try :
    query = self . get_argument ( constants . PARAM_QUERY )
    return query
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_offset ( self ) :
    """Helper function to get offset argument .
 Raises exception if argument is missing .
 Returns the offset argument ."""

implementation:
try :
    offset = self . get_argument ( constants . PARAM_OFFSET )
    return offset
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_argument_length ( self ) :
    """Helper function to get length argument .
 Raises exception if argument is missing .
 Returns the length argument ."""

implementation:
try :
    length = self . get_argument ( constants . PARAM_LENGTH )
    return length
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def get_required_arguments_metricnames ( self ) :
    """Helper function to get metricname arguments .
 Notice that it is get _ argument " s " variation , which means that this can be repeated .
 Raises exception if argument is missing .
 Returns a list of metricname arguments"""

implementation:
try :
    metricnames = self . get_arguments ( constants . PARAM_METRICNAME )
    if not metricnames :
        raise tornado . web . MissingArgumentError ( constants . PARAM_METRICNAME )
    return metricnames
except tornado . web . MissingArgumentError as e :
    raise Exception ( e . log_message )


signature:
def validateInterval ( self , startTime , endTime ) :
    """Helper function to validate interval .
 An interval is valid if starttime and endtime are integrals ,
 and starttime is less than the endtime .
 Raises exception if interval is not valid ."""

implementation:
start = int ( startTime )
end = int ( endTime )
if start > end :
    raise Exception ( "starttime is greater than endtime." )


signature:
def start_connect ( self ) :
    """Tries to connect to the Heron Server
 ` ` loop ( ) ` ` method needs to be called after this ."""

implementation:
Log . debug ( "In start_connect() of %s" % self . _get_classname ( ) )
# TODO : specify buffer size , exception handling
self . create_socket ( socket . AF_INET , socket . SOCK_STREAM )
# when ready , handle _ connect is called
self . _connecting = True
self . connect ( self . endpoint )


signature:
def register_on_message ( self , msg_builder ) :
    """Registers protobuf message builders that this client wants to receive
 : param msg _ builder : callable to create a protobuf message that this client wants to receive"""

implementation:
message = msg_builder ( )
Log . debug ( "In register_on_message(): %s" % message . DESCRIPTOR . full_name )
self . registered_message_map [ message . DESCRIPTOR . full_name ] = msg_builder


signature:
def send_request ( self , request , context , response_type , timeout_sec ) :
    """Sends a request message ( REQID is non - zero )"""

implementation:
# generates a unique request id
reqid = REQID . generate ( )
Log . debug ( "%s: In send_request() with REQID: %s" % ( self . _get_classname ( ) , str ( reqid ) ) )
# register response message type
self . response_message_map [ reqid ] = response_type
self . context_map [ reqid ] = context
# Add timeout for this request if necessary
if timeout_sec > 0 :
    def timeout_task ( ) :
        self . handle_timeout ( reqid )
    self . looper . register_timer_task_in_sec ( timeout_task , timeout_sec )
outgoing_pkt = OutgoingPacket . create_packet ( reqid , request )
self . _send_packet ( outgoing_pkt )


signature:
def send_message ( self , message ) :
    """Sends a message ( REQID is zero )"""

implementation:
Log . debug ( "In send_message() of %s" % self . _get_classname ( ) )
outgoing_pkt = OutgoingPacket . create_packet ( REQID . generate_zero ( ) , message )
self . _send_packet ( outgoing_pkt )


signature:
def handle_timeout ( self , reqid ) :
    """Handles timeout"""

implementation:
if reqid in self . context_map :
    context = self . context_map . pop ( reqid )
    self . response_message_map . pop ( reqid )
    self . on_response ( StatusCode . TIMEOUT_ERROR , context , None )


signature:
def create_tar ( tar_filename , files , config_dir , config_files ) :
    '''Create a tar file with a given set of files'''

implementation:
with contextlib . closing ( tarfile . open ( tar_filename , 'w:gz' , dereference = True ) ) as tar :
    for filename in files :
        if os . path . isfile ( filename ) :
            tar . add ( filename , arcname = os . path . basename ( filename ) )
        else :
            raise Exception ( "%s is not an existing file" % filename )
    if os . path . isdir ( config_dir ) :
        tar . add ( config_dir , arcname = get_heron_sandbox_conf_dir ( ) )
    else :
        raise Exception ( "%s is not an existing directory" % config_dir )
    for filename in config_files :
        if os . path . isfile ( filename ) :
            arcfile = os . path . join ( get_heron_sandbox_conf_dir ( ) , os . path . basename ( filename ) )
            tar . add ( filename , arcname = arcfile )
        else :
            raise Exception ( "%s is not an existing file" % filename )


signature:
def get_subparser ( parser , command ) :
    '''Retrieve the given subparser from parser'''

implementation:
# pylint : disable = protected - access
subparsers_actions = [ action for action in parser . _actions if isinstance ( action , argparse . _SubParsersAction ) ]
# there will probably only be one subparser _ action ,
# but better save than sorry
for subparsers_action in subparsers_actions : # get all subparsers
    for choice , subparser in subparsers_action . choices . items ( ) :
        if choice == command :
            return subparser
return None


signature:
def get_heron_dir ( ) :
    """This will extract heron directory from . pex file .
 For example ,
 when _ _ file _ _ is ' / Users / heron - user / bin / heron / heron / tools / common / src / python / utils / config . pyc ' , and
 its real path is ' / Users / heron - user / . heron / bin / heron / tools / common / src / python / utils / config . pyc ' ,
 the internal variable ` ` path ` ` would be ' / Users / heron - user / . heron ' , which is the heron directory
 This means the variable ` go _ above _ dirs ` below is 9.
 : return : root location of the . pex file"""

implementation:
go_above_dirs = 9
path = "/" . join ( os . path . realpath ( __file__ ) . split ( '/' ) [ : - go_above_dirs ] )
return normalized_class_path ( path )


signature:
def get_heron_libs ( local_jars ) :
    """Get all the heron lib jars with the absolute paths"""

implementation:
heron_lib_dir = get_heron_lib_dir ( )
heron_libs = [ os . path . join ( heron_lib_dir , f ) for f in local_jars ]
return heron_libs


signature:
def parse_cluster_role_env ( cluster_role_env , config_path ) :
    """Parse cluster / [ role ] / [ environ ] , supply default , if not provided , not required"""

implementation:
parts = cluster_role_env . split ( '/' ) [ : 3 ]
if not os . path . isdir ( config_path ) :
    Log . error ( "Config path cluster directory does not exist: %s" % config_path )
    raise Exception ( "Invalid config path" )
# if cluster / role / env is not completely provided , check further
if len ( parts ) < 3 :
    cli_conf_file = os . path . join ( config_path , CLIENT_YAML )
    # if client conf doesn ' t exist , use default value
    if not os . path . isfile ( cli_conf_file ) :
        if len ( parts ) == 1 :
            parts . append ( getpass . getuser ( ) )
        if len ( parts ) == 2 :
            parts . append ( ENVIRON )
    else :
        cli_confs = { }
        with open ( cli_conf_file , 'r' ) as conf_file :
            tmp_confs = yaml . load ( conf_file )
            # the return value of yaml . load can be None if conf _ file is an empty file
            if tmp_confs is not None :
                cli_confs = tmp_confs
            else :
                print ( "Failed to read: %s due to it is empty" % ( CLIENT_YAML ) )
        # if role is required but not provided , raise exception
        if len ( parts ) == 1 :
            if ( ROLE_REQUIRED in cli_confs ) and ( cli_confs [ ROLE_REQUIRED ] is True ) :
                raise Exception ( "role required but not provided (cluster/role/env = %s). See %s in %s" % ( cluster_role_env , ROLE_REQUIRED , cli_conf_file ) )
            else :
                parts . append ( getpass . getuser ( ) )
        # if environ is required but not provided , raise exception
        if len ( parts ) == 2 :
            if ( ENV_REQUIRED in cli_confs ) and ( cli_confs [ ENV_REQUIRED ] is True ) :
                raise Exception ( "environ required but not provided (cluster/role/env = %s). See %s in %s" % ( cluster_role_env , ENV_REQUIRED , cli_conf_file ) )
            else :
                parts . append ( ENVIRON )
# if cluster or role or environ is empty , print
if len ( parts [ 0 ] ) == 0 or len ( parts [ 1 ] ) == 0 or len ( parts [ 2 ] ) == 0 :
    print ( "Failed to parse" )
    sys . exit ( 1 )
return ( parts [ 0 ] , parts [ 1 ] , parts [ 2 ] )


signature:
def get_cluster_role_env ( cluster_role_env ) :
    """Parse cluster / [ role ] / [ environ ] , supply empty string , if not provided"""

implementation:
parts = cluster_role_env . split ( '/' ) [ : 3 ]
if len ( parts ) == 3 :
    return ( parts [ 0 ] , parts [ 1 ] , parts [ 2 ] )
if len ( parts ) == 2 :
    return ( parts [ 0 ] , parts [ 1 ] , "" )
if len ( parts ) == 1 :
    return ( parts [ 0 ] , "" , "" )
return ( "" , "" , "" )


signature:
def direct_mode_cluster_role_env ( cluster_role_env , config_path ) :
    """Check cluster / [ role ] / [ environ ] , if they are required"""

implementation:
# otherwise , get the client . yaml file
cli_conf_file = os . path . join ( config_path , CLIENT_YAML )
# if client conf doesn ' t exist , use default value
if not os . path . isfile ( cli_conf_file ) :
    return True
client_confs = { }
with open ( cli_conf_file , 'r' ) as conf_file :
    client_confs = yaml . load ( conf_file )
    # the return value of yaml . load can be None if conf _ file is an empty file
    if not client_confs :
        return True
    # if role is required but not provided , raise exception
    role_present = True if len ( cluster_role_env [ 1 ] ) > 0 else False
    if ROLE_REQUIRED in client_confs and client_confs [ ROLE_REQUIRED ] and not role_present :
        raise Exception ( "role required but not provided (cluster/role/env = %s). See %s in %s" % ( cluster_role_env , ROLE_REQUIRED , cli_conf_file ) )
    # if environ is required but not provided , raise exception
    environ_present = True if len ( cluster_role_env [ 2 ] ) > 0 else False
    if ENV_REQUIRED in client_confs and client_confs [ ENV_REQUIRED ] and not environ_present :
        raise Exception ( "environ required but not provided (cluster/role/env = %s). See %s in %s" % ( cluster_role_env , ENV_REQUIRED , cli_conf_file ) )
return True