signature:
def write ( self , data ) :
    """Just quote out stuff before sending it out"""

implementation:
args = parse_qs ( self . handler . environ . get ( "QUERY_STRING" ) )
if "i" in args :
    i = args [ "i" ]
else :
    i = "0"
# TODO : don ' t we need to quote this data in here ?
super ( JSONPolling , self ) . write ( "io.j[%s]('%s');" % ( i , data ) )


signature:
def emit_to_room ( self , room , event , * args ) :
    """This is sent to all in the room ( in this particular Namespace )"""

implementation:
pkt = dict ( type = "event" , name = event , args = args , endpoint = self . ns_name )
room_name = self . _get_room_name ( room )
for sessid , socket in six . iteritems ( self . socket . server . sockets ) :
    if 'rooms' not in socket . session :
        continue
    if room_name in socket . session [ 'rooms' ] and self . socket != socket :
        socket . send_packet ( pkt )


signature:
def broadcast_event ( self , event , * args ) :
    """This is sent to all in the sockets in this particular Namespace ,
 including itself ."""

implementation:
pkt = dict ( type = "event" , name = event , args = args , endpoint = self . ns_name )
for sessid , socket in six . iteritems ( self . socket . server . sockets ) :
    socket . send_packet ( pkt )


signature:
def add_parent ( self , parent ) :
    """Add a parent to this role ,
 and add role itself to the parent ' s children set .
 you should override this function if neccessary .
 Example : :
 logged _ user = RoleMixin ( ' logged _ user ' )
 student = RoleMixin ( ' student ' )
 student . add _ parent ( logged _ user )
 : param parent : Parent role to add in ."""

implementation:
parent . children . add ( self )
self . parents . add ( parent )


signature:
def allow ( self , role , method , resource , with_children = True ) :
    """Add allowing rules .
 : param role : Role of this rule .
 : param method : Method to allow in rule , include GET , POST , PUT etc .
 : param resource : Resource also view function .
 : param with _ children : Allow role ' s children in rule as well
 if with _ children is ` True `"""

implementation:
if with_children :
    for r in role . get_children ( ) :
        permission = ( r . get_name ( ) , method , resource )
        if permission not in self . _allowed :
            self . _allowed . append ( permission )
if role == 'anonymous' :
    permission = ( role , method , resource )
else :
    permission = ( role . get_name ( ) , method , resource )
if permission not in self . _allowed :
    self . _allowed . append ( permission )


signature:
def deny ( self , role , method , resource , with_children = False ) :
    """Add denying rules .
 : param role : Role of this rule .
 : param method : Method to deny in rule , include GET , POST , PUT etc .
 : param resource : Resource also view function .
 : param with _ children : Deny role ' s children in rule as well
 if with _ children is ` True `"""

implementation:
if with_children :
    for r in role . get_children ( ) :
        permission = ( r . get_name ( ) , method , resource )
        if permission not in self . _denied :
            self . _denied . append ( permission )
permission = ( role . get_name ( ) , method , resource )
if permission not in self . _denied :
    self . _denied . append ( permission )


signature:
def exempt ( self , resource ) :
    """Exempt a view function from being checked permission
 : param resource : The view function exempt from checking ."""

implementation:
if resource not in self . _exempt :
    self . _exempt . append ( resource )


signature:
def is_allowed ( self , role , method , resource ) :
    """Check whether role is allowed to access resource
 : param role : Role to be checked .
 : param method : Method to be checked .
 : param resource : View function to be checked ."""

implementation: return ( role , method , resource ) in self . _allowed


signature:
def is_denied ( self , role , method , resource ) :
    """Check wherther role is denied to access resource
 : param role : Role to be checked .
 : param method : Method to be checked .
 : param resource : View function to be checked ."""

implementation: return ( role , method , resource ) in self . _denied


signature:
def init_app ( self , app ) :
    """Initialize application in Flask - RBAC .
 Adds ( RBAC , app ) to flask extensions .
 Adds hook to authenticate permission before request .
 : param app : Flask object"""

implementation:
app . config . setdefault ( 'RBAC_USE_WHITE' , False )
self . use_white = app . config [ 'RBAC_USE_WHITE' ]
if not hasattr ( app , 'extensions' ) :
    app . extensions = { }
app . extensions [ 'rbac' ] = _RBACState ( self , app )
self . acl . allow ( anonymous , 'GET' , 'static' )
app . before_first_request ( self . _setup_acl )
app . before_request ( self . _authenticate )


signature:
def has_permission ( self , method , endpoint , user = None ) :
    """Return does the current user can access the resource .
 Example : :
 @ app . route ( ' / some _ url ' , methods = [ ' GET ' , ' POST ' ] )
 @ rbac . allow ( [ ' anonymous ' ] , [ ' GET ' ] )
 def a _ view _ func ( ) :
 return Response ( ' Blah Blah . . . ' )
 If you are not logged .
 ` rbac . has _ permission ( ' GET ' , ' a _ view _ func ' ) ` return True .
 ` rbac . has _ permission ( ' POST ' , ' a _ view _ func ' ) ` return False .
 : param method : The method wait to check .
 : param endpoint : The application endpoint .
 : param user : user who you need to check . Current user by default ."""

implementation:
app = self . get_app ( )
_user = user or self . _user_loader ( )
if not hasattr ( _user , 'get_roles' ) :
    roles = [ anonymous ]
else :
    roles = _user . get_roles ( )
return self . _check_permission ( roles , method , endpoint )


signature:
def allow ( self , roles , methods , with_children = True ) :
    """This is a decorator function .
 You can allow roles to access the view func with it .
 An example : :
 @ app . route ( ' / website / setting ' , methods = [ ' GET ' , ' POST ' ] )
 @ rbac . allow ( [ ' administrator ' , ' super _ user ' ] , [ ' GET ' , ' POST ' ] )
 def website _ setting ( ) :
 return Response ( ' Setting page . ' )
 : param roles : List , each name of roles . Please note that ,
 ` anonymous ` is refered to anonymous .
 If you add ` anonymous ` to the rule ,
 everyone can access the resource ,
 unless you deny other roles .
 : param methods : List , each name of methods .
 methods is valid in [ ' GET ' , ' POST ' , ' PUT ' , ' DELETE ' ]
 : param with _ children : Whether allow children of roles as well .
 True by default ."""

implementation:
def decorator ( view_func ) :
    _methods = [ m . upper ( ) for m in methods ]
    for r , m , v in itertools . product ( roles , _methods , [ view_func . __name__ ] ) :
        self . before_acl [ 'allow' ] . append ( ( r , m , v , with_children ) )
    return view_func
return decorator


signature:
def remove_binaries ( ) :
    """Remove all binary files in the adslib directory ."""

implementation:
patterns = ( "adslib/*.a" , "adslib/*.o" , "adslib/obj/*.o" , "adslib/*.bin" , "adslib/*.so" , )
for f in functools . reduce ( operator . iconcat , [ glob . glob ( p ) for p in patterns ] ) :
    os . remove ( f )


signature:
def router_function ( fn ) : # type : ( Callable ) - > Callable
    """Raise a runtime error if on Win32 systems .
 Decorator .
 Decorator for functions that interact with the router for the Linux
 implementation of the ADS library .
 Unlike the Windows implementation which uses a separate router daemon ,
 the Linux library manages AMS routing in - process . As such , routing must be
 configured programatically via . the provided API . These endpoints are
 invalid on Win32 systems , so an exception will be raised ."""

implementation:
@ wraps ( fn )
def wrapper ( * args , ** kwargs ) : # type : ( Any , Any ) - > Callable
    if platform_is_windows ( ) : # pragma : no cover , skipt Windows test
        raise RuntimeError ( "Router interface is not available on Win32 systems.\n" "Configure AMS routes using the TwinCAT router service." )
    return fn ( * args , ** kwargs )
return wrapper


signature:
def adsAddRoute ( net_id , ip_address ) : # type : ( SAmsNetId , str ) - > None
    """Establish a new route in the AMS Router .
 : param pyads . structs . SAmsNetId net _ id : net id of routing endpoint
 : param str ip _ address : ip address of the routing endpoint"""

implementation:
add_route = _adsDLL . AdsAddRoute
add_route . restype = ctypes . c_long
# Convert ip address to bytes ( PY3 ) and get pointer .
ip_address_p = ctypes . c_char_p ( ip_address . encode ( "utf-8" ) )
error_code = add_route ( net_id , ip_address_p )
if error_code :
    raise ADSError ( error_code )


signature:
def adsPortOpenEx ( ) : # type : ( ) - > int
    """Connect to the TwinCAT message router .
 : rtype : int
 : return : port number"""

implementation:
port_open_ex = _adsDLL . AdsPortOpenEx
port_open_ex . restype = ctypes . c_long
port = port_open_ex ( )
if port == 0 :
    raise RuntimeError ( "Failed to open port on AMS router." )
return port


signature:
def adsPortCloseEx ( port ) : # type : ( int ) - > None
    """Close the connection to the TwinCAT message router ."""

implementation:
port_close_ex = _adsDLL . AdsPortCloseEx
port_close_ex . restype = ctypes . c_long
error_code = port_close_ex ( port )
if error_code :
    raise ADSError ( error_code )


signature:
def adsGetLocalAddressEx ( port ) : # type : ( int ) - > AmsAddr
    """Return the local AMS - address and the port number .
 : rtype : pyads . structs . AmsAddr
 : return : AMS - address"""

implementation:
get_local_address_ex = _adsDLL . AdsGetLocalAddressEx
ams_address_struct = SAmsAddr ( )
error_code = get_local_address_ex ( port , ctypes . pointer ( ams_address_struct ) )
if error_code :
    raise ADSError ( error_code )
local_ams_address = AmsAddr ( )
local_ams_address . _ams_addr = ams_address_struct
return local_ams_address


signature:
def adsSyncReadStateReqEx ( port , address ) : # type : ( int , AmsAddr ) - > Tuple [ int , int ]
    """Read the current ADS - state and the machine - state .
 Read the current ADS - state and the machine - state from the
 ADS - server .
 : param pyads . structs . AmsAddr address : local or remote AmsAddr
 : rtype : ( int , int )
 : return : ads _ state , device _ state"""

implementation:
sync_read_state_request = _adsDLL . AdsSyncReadStateReqEx
# C pointer to ams address struct
ams_address_pointer = ctypes . pointer ( address . amsAddrStruct ( ) )
# Current ADS status and corresponding pointer
ads_state = ctypes . c_int ( )
ads_state_pointer = ctypes . pointer ( ads_state )
# Current device status and corresponding pointer
device_state = ctypes . c_int ( )
device_state_pointer = ctypes . pointer ( device_state )
error_code = sync_read_state_request ( port , ams_address_pointer , ads_state_pointer , device_state_pointer )
if error_code :
    raise ADSError ( error_code )
return ( ads_state . value , device_state . value )


signature:
def adsSyncReadDeviceInfoReqEx ( port , address ) : # type : ( int , AmsAddr ) - > Tuple [ str , AdsVersion ]
    """Read the name and the version number of the ADS - server .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr address : local or remote AmsAddr
 : rtype : string , AdsVersion
 : return : device name , version"""

implementation:
sync_read_device_info_request = _adsDLL . AdsSyncReadDeviceInfoReqEx
# Get pointer to the target AMS address
ams_address_pointer = ctypes . pointer ( address . amsAddrStruct ( ) )
# Create buffer to be filled with device name , get pointer to said buffer
device_name_buffer = ctypes . create_string_buffer ( 20 )
device_name_pointer = ctypes . pointer ( device_name_buffer )
# Create ADS Version struct and get pointer .
ads_version = SAdsVersion ( )
ads_version_pointer = ctypes . pointer ( ads_version )
error_code = sync_read_device_info_request ( port , ams_address_pointer , device_name_pointer , ads_version_pointer )
if error_code :
    raise ADSError ( error_code )
return ( device_name_buffer . value . decode ( ) , AdsVersion ( ads_version ) )


signature:
def adsSyncWriteControlReqEx ( port , address , ads_state , device_state , data , plc_data_type ) : # type : ( int , AmsAddr , int , int , Any , Type ) - > None
    """Change the ADS state and the machine - state of the ADS - server .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr adr : local or remote AmsAddr
 : param int ads _ state : new ADS - state , according to ADSTATE constants
 : param int device _ state : new machine - state
 : param data : additional data
 : param int plc _ data _ type : plc datatype , according to PLCTYPE constants"""

implementation:
sync_write_control_request = _adsDLL . AdsSyncWriteControlReqEx
ams_address_pointer = ctypes . pointer ( address . amsAddrStruct ( ) )
ads_state_c = ctypes . c_ulong ( ads_state )
device_state_c = ctypes . c_ulong ( device_state )
if plc_data_type == PLCTYPE_STRING :
    data = ctypes . c_char_p ( data . encode ( "utf-8" ) )
    data_pointer = data
    data_length = len ( data_pointer . value ) + 1
else :
    data = plc_data_type ( data )
    data_pointer = ctypes . pointer ( data )
    data_length = ctypes . sizeof ( data )
error_code = sync_write_control_request ( port , ams_address_pointer , ads_state_c , device_state_c , data_length , data_pointer , )
if error_code :
    raise ADSError ( error_code )


signature:
def adsSyncWriteReqEx ( port , address , index_group , index_offset , value , plc_data_type ) : # type : ( int , AmsAddr , int , int , Any , Type ) - > None
    """Send data synchronous to an ADS - device .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr address : local or remote AmsAddr
 : param int indexGroup : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param value : value to write to the storage address of the PLC
 : param int plc _ data _ type : type of the data given to the PLC ,
 according to PLCTYPE constants"""

implementation:
sync_write_request = _adsDLL . AdsSyncWriteReqEx
ams_address_pointer = ctypes . pointer ( address . amsAddrStruct ( ) )
index_group_c = ctypes . c_ulong ( index_group )
index_offset_c = ctypes . c_ulong ( index_offset )
if plc_data_type == PLCTYPE_STRING :
    data = ctypes . c_char_p ( value . encode ( "utf-8" ) )
    data_pointer = data
    # type : Union [ ctypes . c _ char _ p , ctypes . pointer ]
    data_length = len ( data_pointer . value ) + 1
# type : ignore
else :
    if type ( plc_data_type ) . __name__ == "PyCArrayType" :
        data = plc_data_type ( * value )
    else :
        data = plc_data_type ( value )
    data_pointer = ctypes . pointer ( data )
    data_length = ctypes . sizeof ( data )
error_code = sync_write_request ( port , ams_address_pointer , index_group_c , index_offset_c , data_length , data_pointer , )
if error_code :
    raise ADSError ( error_code )


signature:
def adsSyncReadWriteReqEx2 ( port , address , index_group , index_offset , read_data_type , value , write_data_type , return_ctypes = False , ) : # type : ( int , AmsAddr , int , int , Type , Any , Type , bool ) - > Any
    """Read and write data synchronous from / to an ADS - device .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr address : local or remote AmsAddr
 : param int index _ group : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param Type read _ data _ type : type of the data given to the PLC to respond to ,
 according to PLCTYPE constants
 : param value : value to write to the storage address of the PLC
 : param Type write _ data _ type : type of the data given to the PLC , according to
 PLCTYPE constants
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )
 : rtype : read _ data _ type
 : return : value : value read from PLC"""

implementation:
sync_read_write_request = _adsDLL . AdsSyncReadWriteReqEx2
ams_address_pointer = ctypes . pointer ( address . amsAddrStruct ( ) )
index_group_c = ctypes . c_ulong ( index_group )
index_offset_c = ctypes . c_ulong ( index_offset )
if read_data_type == PLCTYPE_STRING :
    read_data = ( STRING_BUFFER * PLCTYPE_STRING ) ( )
else :
    read_data = read_data_type ( )
read_data_pointer = ctypes . pointer ( read_data )
read_length = ctypes . c_ulong ( ctypes . sizeof ( read_data ) )
bytes_read = ctypes . c_ulong ( )
bytes_read_pointer = ctypes . pointer ( bytes_read )
if write_data_type == PLCTYPE_STRING : # Get pointer to string
    write_data_pointer = ctypes . c_char_p ( value . encode ( "utf-8" ) )
    # type : Union [ ctypes . c _ char _ p , ctypes . pointer ] # noqa : E501
    # Add an extra byte to the data length for the null terminator
    write_length = len ( value ) + 1
else :
    if type ( write_data_type ) . __name__ == "PyCArrayType" :
        write_data = write_data_type ( * value )
    else :
        write_data = write_data_type ( value )
    write_data_pointer = ctypes . pointer ( write_data )
    write_length = ctypes . sizeof ( write_data )
err_code = sync_read_write_request ( port , ams_address_pointer , index_group_c , index_offset_c , read_length , read_data_pointer , write_length , write_data_pointer , bytes_read_pointer , )
if err_code :
    raise ADSError ( err_code )
# If we ' re reading a value of predetermined size ( anything but a string ) ,
# validate that the correct number of bytes were read
if read_data_type != PLCTYPE_STRING and bytes_read . value != read_length . value :
    raise RuntimeError ( "Insufficient data (expected {0} bytes, {1} were read)." . format ( read_length . value , bytes_read . value ) )
if return_ctypes :
    return read_data
if read_data_type == PLCTYPE_STRING :
    return read_data . value . decode ( "utf-8" )
if type ( read_data_type ) . __name__ == "PyCArrayType" :
    return [ i for i in read_data ]
if hasattr ( read_data , "value" ) :
    return read_data . value
return read_data


signature:
def adsSyncReadReqEx2 ( port , address , index_group , index_offset , data_type , return_ctypes = False ) : # type : ( int , AmsAddr , int , int , Type , bool ) - > Any
    """Read data synchronous from an ADS - device .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr address : local or remote AmsAddr
 : param int index _ group : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param Type data _ type : type of the data given to the PLC , according to
 PLCTYPE constants
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )
 : rtype : data _ type
 : return : value : * * value * *"""

implementation:
sync_read_request = _adsDLL . AdsSyncReadReqEx2
ams_address_pointer = ctypes . pointer ( address . amsAddrStruct ( ) )
index_group_c = ctypes . c_ulong ( index_group )
index_offset_c = ctypes . c_ulong ( index_offset )
if data_type == PLCTYPE_STRING :
    data = ( STRING_BUFFER * PLCTYPE_STRING ) ( )
else :
    data = data_type ( )
data_pointer = ctypes . pointer ( data )
data_length = ctypes . c_ulong ( ctypes . sizeof ( data ) )
bytes_read = ctypes . c_ulong ( )
bytes_read_pointer = ctypes . pointer ( bytes_read )
error_code = sync_read_request ( port , ams_address_pointer , index_group_c , index_offset_c , data_length , data_pointer , bytes_read_pointer , )
if error_code :
    raise ADSError ( error_code )
# If we ' re reading a value of predetermined size ( anything but a string ) ,
# validate that the correct number of bytes were read
if data_type != PLCTYPE_STRING and bytes_read . value != data_length . value :
    raise RuntimeError ( "Insufficient data (expected {0} bytes, {1} were read)." . format ( data_length . value , bytes_read . value ) )
if return_ctypes :
    return data
if data_type == PLCTYPE_STRING :
    return data . value . decode ( "utf-8" )
if type ( data_type ) . __name__ == "PyCArrayType" :
    return [ i for i in data ]
if hasattr ( data , "value" ) :
    return data . value
return data


signature:
def adsSyncReadByNameEx ( port , address , data_name , data_type , return_ctypes = False ) : # type : ( int , AmsAddr , str , Type , bool ) - > Any
    """Read data synchronous from an ADS - device from data name .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr address : local or remote AmsAddr
 : param string data _ name : data name
 : param Type data _ type : type of the data given to the PLC , according to
 PLCTYPE constants
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )
 : rtype : data _ type
 : return : value : * * value * *"""

implementation:
# Get the handle of the PLC - variable
handle = adsSyncReadWriteReqEx2 ( port , address , ADSIGRP_SYM_HNDBYNAME , 0x0 , PLCTYPE_UDINT , data_name , PLCTYPE_STRING , )
# Read the value of a PLC - variable , via handle
value = adsSyncReadReqEx2 ( port , address , ADSIGRP_SYM_VALBYHND , handle , data_type , return_ctypes )
# Release the handle of the PLC - variable
adsSyncWriteReqEx ( port , address , ADSIGRP_SYM_RELEASEHND , 0 , handle , PLCTYPE_UDINT )
return value


signature:
def adsSyncWriteByNameEx ( port , address , data_name , value , data_type ) : # type : ( int , AmsAddr , str , Any , Type ) - > None
    """Send data synchronous to an ADS - device from data name .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr address : local or remote AmsAddr
 : param string data _ name : PLC storage address
 : param value : value to write to the storage address of the PLC
 : param Type data _ type : type of the data given to the PLC ,
 according to PLCTYPE constants"""

implementation:
# Get the handle of the PLC - variable
handle = adsSyncReadWriteReqEx2 ( port , address , ADSIGRP_SYM_HNDBYNAME , 0x0 , PLCTYPE_UDINT , data_name , PLCTYPE_STRING , )
# Write the value of a PLC - variable , via handle
adsSyncWriteReqEx ( port , address , ADSIGRP_SYM_VALBYHND , handle , value , data_type )
# Release the handle of the PLC - variable
adsSyncWriteReqEx ( port , address , ADSIGRP_SYM_RELEASEHND , 0 , handle , PLCTYPE_UDINT )


signature:
def adsSyncAddDeviceNotificationReqEx ( port , adr , data_name , pNoteAttrib , callback , user_handle = None ) : # type : ( int , AmsAddr , str , NotificationAttrib , Callable , int ) - > Tuple [ int , int ]
    """Add a device notification .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr adr : local or remote AmsAddr
 : param string data _ name : PLC storage address
 : param pyads . structs . NotificationAttrib pNoteAttrib : notification attributes
 : param callback : Callback function to handle notification
 : param user _ handle : User Handle
 : rtype : ( int , int )
 : returns : notification handle , user handle"""

implementation:
global callback_store
if NOTEFUNC is None :
    raise TypeError ( "Callback function type can't be None" )
adsSyncAddDeviceNotificationReqFct = _adsDLL . AdsSyncAddDeviceNotificationReqEx
pAmsAddr = ctypes . pointer ( adr . amsAddrStruct ( ) )
hnl = adsSyncReadWriteReqEx2 ( port , adr , ADSIGRP_SYM_HNDBYNAME , 0x0 , PLCTYPE_UDINT , data_name , PLCTYPE_STRING )
nIndexGroup = ctypes . c_ulong ( ADSIGRP_SYM_VALBYHND )
nIndexOffset = ctypes . c_ulong ( hnl )
attrib = pNoteAttrib . notificationAttribStruct ( )
pNotification = ctypes . c_ulong ( )
nHUser = ctypes . c_ulong ( hnl )
if user_handle is not None :
    nHUser = ctypes . c_ulong ( user_handle )
adsSyncAddDeviceNotificationReqFct . argtypes = [ ctypes . c_ulong , ctypes . POINTER ( SAmsAddr ) , ctypes . c_ulong , ctypes . c_ulong , ctypes . POINTER ( SAdsNotificationAttrib ) , NOTEFUNC , ctypes . c_ulong , ctypes . POINTER ( ctypes . c_ulong ) , ]
adsSyncAddDeviceNotificationReqFct . restype = ctypes . c_long
def wrapper ( addr , notification , user ) : # type : ( AmsAddr , SAdsNotificationHeader , int ) - > Callable [ [ SAdsNotificationHeader , str ] , None ]
    return callback ( notification , data_name )
c_callback = NOTEFUNC ( wrapper )
err_code = adsSyncAddDeviceNotificationReqFct ( port , pAmsAddr , nIndexGroup , nIndexOffset , ctypes . byref ( attrib ) , c_callback , nHUser , ctypes . byref ( pNotification ) , )
if err_code :
    raise ADSError ( err_code )
callback_store [ pNotification . value ] = c_callback
return ( pNotification . value , hnl )


signature:
def adsSyncDelDeviceNotificationReqEx ( port , adr , notification_handle , user_handle ) : # type : ( int , AmsAddr , int , int ) - > None
    """Remove a device notification .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param pyads . structs . AmsAddr adr : local or remote AmsAddr
 : param int notification _ handle : Notification Handle
 : param int user _ handle : User Handle"""

implementation:
adsSyncDelDeviceNotificationReqFct = _adsDLL . AdsSyncDelDeviceNotificationReqEx
pAmsAddr = ctypes . pointer ( adr . amsAddrStruct ( ) )
nHNotification = ctypes . c_ulong ( notification_handle )
err_code = adsSyncDelDeviceNotificationReqFct ( port , pAmsAddr , nHNotification )
callback_store . pop ( notification_handle , None )
if err_code :
    raise ADSError ( err_code )
adsSyncWriteReqEx ( port , adr , ADSIGRP_SYM_RELEASEHND , 0 , user_handle , PLCTYPE_UDINT )


signature:
def adsSyncSetTimeoutEx ( port , nMs ) : # type : ( int , int ) - > None
    """Set Timeout .
 : param int port : local AMS port as returned by adsPortOpenEx ( )
 : param int nMs : timeout in ms"""

implementation:
adsSyncSetTimeoutFct = _adsDLL . AdsSyncSetTimeoutEx
cms = ctypes . c_long ( nMs )
err_code = adsSyncSetTimeoutFct ( port , cms )
if err_code :
    raise ADSError ( err_code )


signature:
def _parse_ams_netid ( ams_netid ) : # type : ( str ) - > SAmsNetId
    """Parse an AmsNetId from * str * to * SAmsNetId * .
 : param str ams _ netid : NetId as a string
 : rtype : SAmsNetId
 : return : NetId as a struct"""

implementation:
try :
    id_numbers = list ( map ( int , ams_netid . split ( "." ) ) )
except ValueError :
    raise ValueError ( "no valid netid" )
if len ( id_numbers ) != 6 :
    raise ValueError ( "no valid netid" )
# Fill the netId struct with data
ams_netid_st = SAmsNetId ( )
ams_netid_st . b = ( c_ubyte * 6 ) ( * id_numbers )
return ams_netid_st


signature:
def set_local_address ( ams_netid ) : # type : ( Union [ str , SAmsNetId ] ) - > None
    """Set the local NetID ( * * Linux only * * ) .
 : param str : new AmsNetID
 : rtype : None
 * * Usage : * *
 > > > import pyads
 > > > pyads . open _ port ( )
 > > > pyads . set _ local _ address ( ' 0.0.0.0.1.1 ' )"""

implementation:
if isinstance ( ams_netid , str ) :
    ams_netid_st = _parse_ams_netid ( ams_netid )
else :
    ams_netid_st = ams_netid
assert isinstance ( ams_netid_st , SAmsNetId )
if linux :
    return adsSetLocalAddress ( ams_netid_st )
else :
    raise ADSError ( text = "SetLocalAddress is not supported for Windows clients." )


signature:
def write_control ( adr , ads_state , device_state , data , plc_datatype ) : # type : ( AmsAddr , int , int , Any , Type ) - > None
    """Change the ADS state and the machine - state of the ADS - server .
 : param AmsAddr adr : local or remote AmsAddr
 : param int ads _ state : new ADS - state , according to ADSTATE constants
 : param int device _ state : new machine - state
 : param data : additional data
 : param int plc _ datatype : datatype , according to PLCTYPE constants
 : note : Despite changing the ADS - state and the machine - state it is possible
 to send additional data to the ADS - server . For current ADS - devices
 additional data is not progressed .
 Every ADS - device is able to communicate its current state to other
 devices .
 There is a difference between the device - state and the state of the
 ADS - interface ( AdsState ) . The possible states of an ADS - interface
 are defined in the ADS - specification ."""

implementation:
if port is not None :
    return adsSyncWriteControlReqEx ( port , adr , ads_state , device_state , data , plc_datatype )


signature:
def write ( adr , index_group , index_offset , value , plc_datatype ) : # type : ( AmsAddr , int , int , Any , Type ) - > None
    """Send data synchronous to an ADS - device .
 : param AmsAddr adr : local or remote AmsAddr
 : param int index _ group : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param value : value to write to the storage address of the PLC
 : param Type plc _ datatype : type of the data given to the PLC ,
 according to PLCTYPE constants"""

implementation:
if port is not None :
    return adsSyncWriteReqEx ( port , adr , index_group , index_offset , value , plc_datatype )


signature:
def read_write ( adr , index_group , index_offset , plc_read_datatype , value , plc_write_datatype , return_ctypes = False , ) : # type : ( AmsAddr , int , int , Type , Any , Type , bool ) - > Any
    """Read and write data synchronous from / to an ADS - device .
 : param AmsAddr adr : local or remote AmsAddr
 : param int index _ group : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param Type plc _ read _ datatype : type of the data given to the PLC to respond
 to , according to PLCTYPE constants
 : param value : value to write to the storage address of the PLC
 : param Type plc _ write _ datatype : type of the data given to the PLC , according to
 PLCTYPE constants
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )
 : rtype : PLCTYPE
 : return : value : * * value * *"""

implementation:
if port is not None :
    return adsSyncReadWriteReqEx2 ( port , adr , index_group , index_offset , plc_read_datatype , value , plc_write_datatype , return_ctypes , )
return None


signature:
def read ( adr , index_group , index_offset , plc_datatype , return_ctypes = False ) : # type : ( AmsAddr , int , int , Type , bool ) - > Any
    """Read data synchronous from an ADS - device .
 : param AmsAddr adr : local or remote AmsAddr
 : param int index _ group : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param int plc _ datatype : type of the data given to the PLC , according to
 PLCTYPE constants
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )
 : return : value : * * value * *"""

implementation:
if port is not None :
    return adsSyncReadReqEx2 ( port , adr , index_group , index_offset , plc_datatype , return_ctypes )
return None


signature:
def read_by_name ( adr , data_name , plc_datatype , return_ctypes = False ) : # type : ( AmsAddr , str , Type , bool ) - > Any
    """Read data synchronous from an ADS - device from data name .
 : param AmsAddr adr : local or remote AmsAddr
 : param string data _ name : data name
 : param int plc _ datatype : type of the data given to the PLC , according to
 PLCTYPE constants
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )
 : return : value : * * value * *"""

implementation:
if port is not None :
    return adsSyncReadByNameEx ( port , adr , data_name , plc_datatype , return_ctypes )
return None


signature:
def write_by_name ( adr , data_name , value , plc_datatype ) : # type : ( AmsAddr , str , Any , Type ) - > None
    """Send data synchronous to an ADS - device from data name .
 : param AmsAddr adr : local or remote AmsAddr
 : param string data _ name : PLC storage address
 : param value : value to write to the storage address of the PLC
 : param int plc _ datatype : type of the data given to the PLC ,
 according to PLCTYPE constants"""

implementation:
if port is not None :
    return adsSyncWriteByNameEx ( port , adr , data_name , value , plc_datatype )


signature:
def add_device_notification ( adr , data_name , attr , callback , user_handle = None ) : # type : ( AmsAddr , str , NotificationAttrib , Callable , int ) - > Optional [ Tuple [ int , int ] ] # noqa : E501
    """Add a device notification .
 : param pyads . structs . AmsAddr adr : AMS Address associated with the routing
 entry which is to be removed from the router .
 : param str data _ name : PLC storage address
 : param pyads . structs . NotificationAttrib attr : object that contains
 all the attributes for the definition of a notification
 : param callback : callback function that gets executed on in the event
 of a notification
 : rtype : ( int , int )
 : returns : notification handle , user handle
 Save the notification handle and the user handle on creating a
 notification if you want to be able to remove the notification
 later in your code ."""

implementation:
if port is not None :
    return adsSyncAddDeviceNotificationReqEx ( port , adr , data_name , attr , callback , user_handle )
return None


signature:
def del_device_notification ( adr , notification_handle , user_handle ) : # type : ( AmsAddr , int , int ) - > None
    """Remove a device notification .
 : param pyads . structs . AmsAddr adr : AMS Address associated with the routing
 entry which is to be removed from the router .
 : param notification _ handle : address of the variable that contains
 the handle of the notification
 : param user _ handle : user handle"""

implementation:
if port is not None :
    return adsSyncDelDeviceNotificationReqEx ( port , adr , notification_handle , user_handle )


signature:
def open ( self ) : # type : ( ) - > None
    """Connect to the TwinCAT message router ."""

implementation:
if self . _open :
    return
self . _port = adsPortOpenEx ( )
if linux :
    adsAddRoute ( self . _adr . netIdStruct ( ) , self . ip_address )
self . _open = True


signature:
def close ( self ) : # type : ( ) - > None
    """: summary : Close the connection to the TwinCAT message router ."""

implementation:
if not self . _open :
    return
if linux :
    adsDelRoute ( self . _adr . netIdStruct ( ) )
if self . _port is not None :
    adsPortCloseEx ( self . _port )
    self . _port = None
self . _open = False


signature:
def write_control ( self , ads_state , device_state , data , plc_datatype ) : # type : ( int , int , Any , Type ) - > None
    """Change the ADS state and the machine - state of the ADS - server .
 : param int ads _ state : new ADS - state , according to ADSTATE constants
 : param int device _ state : new machine - state
 : param data : additional data
 : param int plc _ datatype : datatype , according to PLCTYPE constants
 : note : Despite changing the ADS - state and the machine - state it is
 possible to send additional data to the ADS - server . For current
 ADS - devices additional data is not progressed .
 Every ADS - device is able to communicate its current state to other
 devices . There is a difference between the device - state and the
 state of the ADS - interface ( AdsState ) . The possible states of an
 ADS - interface are defined in the ADS - specification ."""

implementation:
if self . _port is not None :
    return adsSyncWriteControlReqEx ( self . _port , self . _adr , ads_state , device_state , data , plc_datatype )


signature:
def write ( self , index_group , index_offset , value , plc_datatype ) : # type : ( int , int , Any , Type ) - > None
    """Send data synchronous to an ADS - device .
 : param int index _ group : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param value : value to write to the storage address of the PLC
 : param int plc _ datatype : type of the data given to the PLC ,
 according to PLCTYPE constants"""

implementation:
if self . _port is not None :
    return adsSyncWriteReqEx ( self . _port , self . _adr , index_group , index_offset , value , plc_datatype )


signature:
def read_write ( self , index_group , index_offset , plc_read_datatype , value , plc_write_datatype , return_ctypes = False , ) : # type : ( int , int , Type , Any , Type , bool ) - > Any
    """Read and write data synchronous from / to an ADS - device .
 : param int index _ group : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param int plc _ read _ datatype : type of the data given to the PLC to
 respond to , according to PLCTYPE constants
 : param value : value to write to the storage address of the PLC
 : param plc _ write _ datatype : type of the data given to the PLC ,
 according to PLCTYPE constants
 : rtype : PLCTYPE
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )
 : return : value : * * value * *"""

implementation:
if self . _port is not None :
    return adsSyncReadWriteReqEx2 ( self . _port , self . _adr , index_group , index_offset , plc_read_datatype , value , plc_write_datatype , return_ctypes , )
return None


signature:
def read ( self , index_group , index_offset , plc_datatype , return_ctypes = False ) : # type : ( int , int , Type , bool ) - > Any
    """Read data synchronous from an ADS - device .
 : param int index _ group : PLC storage area , according to the INDEXGROUP
 constants
 : param int index _ offset : PLC storage address
 : param int plc _ datatype : type of the data given to the PLC , according
 to PLCTYPE constants
 : return : value : * * value * *
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )"""

implementation:
if self . _port is not None :
    return adsSyncReadReqEx2 ( self . _port , self . _adr , index_group , index_offset , plc_datatype , return_ctypes )
return None


signature:
def read_by_name ( self , data_name , plc_datatype , return_ctypes = False ) : # type : ( str , Type , bool ) - > Any
    """Read data synchronous from an ADS - device from data name .
 : param string data _ name : data name
 : param int plc _ datatype : type of the data given to the PLC , according
 to PLCTYPE constants
 : return : value : * * value * *
 : param bool return _ ctypes : return ctypes instead of python types if True
 ( default : False )"""

implementation:
if self . _port :
    return adsSyncReadByNameEx ( self . _port , self . _adr , data_name , plc_datatype , return_ctypes )
return None


signature:
def write_by_name ( self , data_name , value , plc_datatype ) : # type : ( str , Any , Type ) - > None
    """Send data synchronous to an ADS - device from data name .
 : param string data _ name : PLC storage address
 : param value : value to write to the storage address of the PLC
 : param int plc _ datatype : type of the data given to the PLC ,
 according to PLCTYPE constants"""

implementation:
if self . _port :
    return adsSyncWriteByNameEx ( self . _port , self . _adr , data_name , value , plc_datatype )


signature:
def add_device_notification ( self , data_name , attr , callback , user_handle = None ) : # type : ( str , NotificationAttrib , Callable , int ) - > Optional [ Tuple [ int , int ] ]
    """Add a device notification .
 : param str data _ name : PLC storage address
 : param pyads . structs . NotificationAttrib attr : object that contains
 all the attributes for the definition of a notification
 : param callback : callback function that gets executed on in the event
 of a notification
 : rtype : ( int , int )
 : returns : notification handle , user handle
 Save the notification handle and the user handle on creating a
 notification if you want to be able to remove the notification
 later in your code .
 * * Usage * * :
 > > > import pyads
 > > > from ctypes import size _ of
 > > > # Connect to the local TwinCAT PLC
 > > > plc = pyads . Connection ( ' 127.0.0.1.1.1 ' , 851)
 > > > # Create callback function that prints the value
 > > > def mycallback ( adr , notification , user ) :
 > > > contents = notification . contents
 > > > value = next (
 > > > map ( int ,
 > > > bytearray ( contents . data ) [ 0 : contents . cbSampleSize ] )
 > > > print ( value )
 > > > with plc :
 > > > # Add notification with default settings
 > > > attr = pyads . NotificationAttrib ( size _ of ( pyads . PLCTYPE _ INT ) )
 > > > hnotification , huser = plc . add _ device _ notification (
 > > > adr , attr , mycallback )
 > > > # Remove notification
 > > > plc . del _ device _ notification ( hnotification , huser )"""

implementation:
if self . _port is not None :
    notification_handle , user_handle = adsSyncAddDeviceNotificationReqEx ( self . _port , self . _adr , data_name , attr , callback , user_handle )
    return notification_handle , user_handle
return None


signature:
def del_device_notification ( self , notification_handle , user_handle ) : # type : ( int , int ) - > None
    """Remove a device notification .
 : param notification _ handle : address of the variable that contains
 the handle of the notification
 : param user _ handle : user handle"""

implementation:
if self . _port is not None :
    adsSyncDelDeviceNotificationReqEx ( self . _port , self . _adr , notification_handle , user_handle )


signature:
def notification ( self , plc_datatype = None ) : # type : ( Optional [ Type [ Any ] ] ) - > Callable
    """Decorate a callback function .
 * * Decorator * * .
 A decorator that can be used for callback functions in order to
 convert the data of the NotificationHeader into the fitting
 Python type .
 : param plc _ datatype : The PLC datatype that needs to be converted . This can
 be any basic PLC datatype or a ` ctypes . Structure ` .
 The callback functions need to be of the following type :
 > > > def callback ( handle , name , timestamp , value )
 * ` handle ` : the notification handle
 * ` name ` : the variable name
 * ` timestamp ` : the timestamp as datetime value
 * ` value ` : the converted value of the variable
 * * Usage * * :
 > > > import pyads
 > > > plc = pyads . Connection ( ' 172.18.3.25.1.1 ' , 851)
 > > > @ plc . notification ( pyads . PLCTYPE _ STRING )
 > > > def callback ( handle , name , timestamp , value ) :
 > > > print ( handle , name , timestamp , value )
 > > > with plc :
 > > > attr = pyads . NotificationAttrib ( 20,
 > > > pyads . ADSTRANS _ SERVERCYCLE )
 > > > handles = plc . add _ device _ notification ( ' GVL . test ' , attr ,
 > > > callback )
 > > > while True :
 > > > pass"""

implementation:
def notification_decorator ( func ) : # type : ( Callable [ [ int , str , datetime , Any ] , None ] ) - > Callable [ [ Any , str ] , None ] # noqa : E501
    def func_wrapper ( notification , data_name ) : # type : ( Any , str ) - > None
        contents = notification . contents
        data = contents . data
        data_size = contents . cbSampleSize
        datatype_map = { PLCTYPE_BOOL : "<?" , PLCTYPE_BYTE : "<c" , PLCTYPE_DINT : "<i" , PLCTYPE_DWORD : "<I" , PLCTYPE_INT : "<h" , PLCTYPE_LREAL : "<d" , PLCTYPE_REAL : "<f" , PLCTYPE_SINT : "<b" , PLCTYPE_UDINT : "<L" , PLCTYPE_UINT : "<H" , PLCTYPE_USINT : "<B" , PLCTYPE_WORD : "<H" , }
        # type : Dict [ Type , str ]
        if plc_datatype == PLCTYPE_STRING :
            dest = ( c_ubyte * data_size ) ( )
            memmove ( addressof ( dest ) , addressof ( data ) , data_size )
            # read only until null - termination character
            value = bytearray ( dest ) . split ( b"\0" , 1 ) [ 0 ] . decode ( "utf-8" )
        elif issubclass ( plc_datatype , Structure ) :
            value = plc_datatype ( )
            fit_size = min ( data_size , sizeof ( value ) )
            memmove ( addressof ( value ) , addressof ( data ) , fit_size )
        elif plc_datatype not in datatype_map :
            value = data
        else :
            value = struct . unpack ( datatype_map [ plc_datatype ] , bytearray ( data ) [ : data_size ] ) [ 0 ]
        dt = filetime_to_dt ( contents . nTimeStamp )
        return func ( contents . hNotification , data_name , dt , value )
    return func_wrapper
return notification_decorator


signature:
def xml_extract_text ( node , xpath ) :
    """: param node : the node to be queried
 : param xpath : the path to fetch the child node that has the wanted text"""

implementation:
text = node . find ( xpath ) . text
if text is not None :
    text = text . strip ( )
return text


signature:
def xml_extract_date ( node , xpath , date_format = '%d/%m/%Y' ) :
    """: param node : the node to be queried
 : param xpath : the path to fetch the child node that has the wanted date"""

implementation: return datetime . strptime ( xml_extract_text ( node , xpath ) , date_format )


signature:
def xml_extract_datetime ( node , xpath , datetime_format = '%d/%m/%Y %H:%M:%S' ) :
    """: param node : the node to be queried
 : param xpath : the path to fetch the child node that has the wanted datetime"""

implementation: return datetime . strptime ( xml_extract_text ( node , xpath ) , datetime_format )


signature:
def translate_column ( df , column , translations ) :
    """: param df : ( pandas . Dataframe ) the dataframe to be translated
 : param column : ( str ) the column to be translated
 : param translations : ( dict ) a dictionary of the strings to be categorized and translated"""

implementation:
df [ column ] = df [ column ] . astype ( 'category' )
translations = [ translations [ cat ] for cat in df [ column ] . cat . categories ]
df [ column ] . cat . rename_categories ( translations , inplace = True )


signature:
def fetch_speeches ( data_dir , range_start , range_end ) :
    """: param data _ dir : ( str ) directory in which the output file will be saved
 : param range _ start : ( str ) date in the format dd / mm / yyyy
 : param range _ end : ( str ) date in the format dd / mm / yyyy"""

implementation:
speeches = SpeechesDataset ( )
df = speeches . fetch ( range_start , range_end )
save_to_csv ( df , data_dir , "speeches" )
return df


signature:
def fetch ( self , range_start , range_end ) :
    """Fetches speeches from the ListarDiscursosPlenario endpoint of the
 SessoesReunioes ( SessionsReunions ) API .
 The date range provided should be specified as a string using the
 format supported by the API ( % d / % m / % Y )"""

implementation:
range_dates = { 'dataIni' : range_start , 'dataFim' : range_end }
url = self . URL . format ( ** range_dates )
xml = urllib . request . urlopen ( url )
tree = ET . ElementTree ( file = xml )
records = self . _parse_speeches ( tree . getroot ( ) )
return pd . DataFrame ( records , columns = [ 'session_code' , 'session_date' , 'session_num' , 'phase_code' , 'phase_desc' , 'speech_speaker_num' , 'speech_speaker_name' , 'speech_speaker_party' , 'speech_speaker_state' , 'speech_started_at' , 'speech_room_num' , 'speech_insertion_num' ] )


signature:
def fetch_presences ( data_dir , deputies , date_start , date_end ) :
    """: param data _ dir : ( str ) directory in which the output file will be saved
 : param deputies : ( pandas . DataFrame ) a dataframe with deputies data
 : param date _ start : ( str ) a date in the format dd / mm / yyyy
 : param date _ end : ( str ) a date in the format dd / mm / yyyy"""

implementation:
presences = PresencesDataset ( )
df = presences . fetch ( deputies , date_start , date_end )
save_to_csv ( df , data_dir , "presences" )
log . info ( "Presence records:" , len ( df ) )
log . info ( "Records of deputies present on a session:" , len ( df [ df . presence == 'Present' ] ) )
log . info ( "Records of deputies absent from a session:" , len ( df [ df . presence == 'Absent' ] ) )
return df


signature:
def fetch ( self , deputies , start_date , end_date ) :
    """: param deputies : ( pandas . DataFrame ) a dataframe with deputies data
 : param date _ start : ( str ) date in the format dd / mm / yyyy
 : param date _ end : ( str ) date in the format dd / mm / yyyy"""

implementation:
log . debug ( "Fetching data for {} deputies from {} -> {}" . format ( len ( deputies ) , start_date , end_date ) )
records = self . _all_presences ( deputies , start_date , end_date )
df = pd . DataFrame ( records , columns = ( 'term' , 'congressperson_document' , 'congressperson_name' , 'party' , 'state' , 'date' , 'present_on_day' , 'justification' , 'session' , 'presence' ) )
return self . _translate ( df )


signature:
def fetch_official_missions ( data_dir , start_date , end_date ) :
    """: param data _ dir : ( str ) directory in which the output file will be saved
 : param start _ date : ( datetime ) first date of the range to be scraped
 : param end _ date : ( datetime ) last date of the range to be scraped"""

implementation:
official_missions = OfficialMissionsDataset ( )
df = official_missions . fetch ( start_date , end_date )
save_to_csv ( df , data_dir , "official-missions" )
return df


signature:
def fetch ( self , start_date , end_date ) :
    """Fetches official missions within the given date range"""

implementation:
records = [ ]
for two_months_range in self . _generate_ranges ( start_date , end_date ) :
    log . debug ( two_months_range )
    for record in self . _fetch_missions_for_range ( two_months_range [ 0 ] , two_months_range [ 1 ] ) :
        records . append ( record )
df = pd . DataFrame ( records , columns = [ 'participant' , 'destination' , 'subject' , 'start' , 'end' , 'canceled' , 'report_status' , 'report_details_link' ] )
translate_column ( df , 'report_status' , { 'Disponível' : 'Available' , 'Pendente' : 'Pending' , 'Em análise' : 'Analysing' , 'Não se aplica' : 'Does not apply' } )
translate_column ( df , 'canceled' , { 'Não' : 'No' , 'Sim' : 'Yes' } )
return df . drop_duplicates ( )


signature:
def _generate_ranges ( start_date , end_date ) :
    """Generate a list of 2 month ranges for the range requested with an
 intersection between months . This is necessary because we can ' t search
 for ranges longer than 3 months and the period searched has to encompass
 the whole period of the mission ."""

implementation:
range_start = start_date
while range_start < end_date :
    range_end = range_start + timedelta ( days = 60 )
    yield ( range_start . strftime ( "%d/%m/%Y" ) , range_end . strftime ( "%d/%m/%Y" ) )
    range_start += timedelta ( days = 30 )


signature:
def fetch_session_start_times ( data_dir , pivot , session_dates ) :
    """: param data _ dir : ( str ) directory in which the output file will be saved
 : param pivot : ( int ) congressperson document to use as a pivot for scraping the data
 : param session _ dates : ( list ) datetime objects to fetch the start times for"""

implementation:
session_start_times = SessionStartTimesDataset ( )
df = session_start_times . fetch ( pivot , session_dates )
save_to_csv ( df , data_dir , "session-start-times" )
log . info ( "Dates requested:" , len ( session_dates ) )
found = pd . to_datetime ( df [ 'date' ] , format = "%Y-%m-%d %H:%M:%S" ) . dt . date . unique ( )
log . info ( "Dates found:" , len ( found ) )
return df


signature:
def fetch ( self , pivot , session_dates ) :
    """: param pivot : ( int ) a congressperson document to use as a pivot for scraping the data
 : param session _ dates : ( list ) datetime objects to fetch the start times for"""

implementation:
records = self . _all_start_times ( pivot , session_dates )
return pd . DataFrame ( records , columns = ( 'date' , 'session' , 'started_at' ) )


signature:
def fetch_deputies ( data_dir ) :
    """: param data _ dir : ( str ) directory in which the output file will be saved"""

implementation:
deputies = DeputiesDataset ( )
df = deputies . fetch ( )
save_to_csv ( df , data_dir , "deputies" )
holders = df . condition == 'Holder'
substitutes = df . condition == 'Substitute'
log . info ( "Total deputies:" , len ( df ) )
log . info ( "Holder deputies:" , len ( df [ holders ] ) )
log . info ( "Substitute deputies:" , len ( df [ substitutes ] ) )
return df


signature:
def fetch ( self ) :
    """Fetches the list of deputies for the current term ."""

implementation:
xml = urllib . request . urlopen ( self . URL )
tree = ET . ElementTree ( file = xml )
records = self . _parse_deputies ( tree . getroot ( ) )
df = pd . DataFrame ( records , columns = ( 'congressperson_id' , 'budget_id' , 'condition' , 'congressperson_document' , 'civil_name' , 'congressperson_name' , 'picture_url' , 'gender' , 'state' , 'party' , 'phone_number' , 'email' ) )
return self . _translate ( df )


signature:
def add_node ( self , node ) :
    """Adds a ` node ` to the hash ring ( including a number of replicas ) ."""

implementation:
self . nodes . append ( node )
for x in xrange ( self . replicas ) :
    ring_key = self . hash_method ( b ( "%s:%d" % ( node , x ) ) )
    self . ring [ ring_key ] = node
    self . sorted_keys . append ( ring_key )
self . sorted_keys . sort ( )


signature:
def remove_node ( self , node ) :
    """Removes ` node ` from the hash ring and its replicas ."""

implementation:
self . nodes . remove ( node )
for x in xrange ( self . replicas ) :
    ring_key = self . hash_method ( b ( "%s:%d" % ( node , x ) ) )
    self . ring . pop ( ring_key )
    self . sorted_keys . remove ( ring_key )


signature:
def get_node_pos ( self , key ) :
    """Given a string key a corresponding node in the hash ring is returned
 along with it ' s position in the ring .
 If the hash ring is empty , ( ` None ` , ` None ` ) is returned ."""

implementation:
if len ( self . ring ) == 0 :
    return [ None , None ]
crc = self . hash_method ( b ( key ) )
idx = bisect . bisect ( self . sorted_keys , crc )
# prevents out of range index
idx = min ( idx , ( self . replicas * len ( self . nodes ) ) - 1 )
return [ self . ring [ self . sorted_keys [ idx ] ] , idx ]


signature:
def iter_nodes ( self , key ) :
    """Given a string key it returns the nodes as a generator that can hold the key ."""

implementation:
if len ( self . ring ) == 0 :
    yield None , None
node , pos = self . get_node_pos ( key )
for k in self . sorted_keys [ pos : ] :
    yield k , self . ring [ k ]


implementation:
configs = [ ]
if not isinstance ( servers , list ) :
    raise ValueError ( "server's config must be list" )
_type = type ( servers [ 0 ] )
if _type == dict :
    return servers
if ( sys . version_info [ 0 ] == 3 and _type in [ str , bytes ] ) or ( sys . version_info [ 0 ] == 2 and _type in [ str , unicode ] ) :
    for config in servers :
        configs . append ( parse_url ( config ) )
else :
    raise ValueError ( "invalid server config" )
return configs


signature:
def mget ( self , keys , * args ) :
    """Returns a list of values ordered identically to ` ` keys ` `"""

implementation:
args = list_or_args ( keys , args )
server_keys = { }
ret_dict = { }
for key in args :
    server_name = self . get_server_name ( key )
    server_keys [ server_name ] = server_keys . get ( server_name , [ ] )
    server_keys [ server_name ] . append ( key )
for server_name , sub_keys in iteritems ( server_keys ) :
    values = self . connections [ server_name ] . mget ( sub_keys )
    ret_dict . update ( dict ( zip ( sub_keys , values ) ) )
result = [ ]
for key in args :
    result . append ( ret_dict . get ( key , None ) )
return result


signature:
def mset ( self , mapping ) :
    """Sets each key in the ` ` mapping ` ` dict to its corresponding value"""

implementation:
servers = { }
for key , value in mapping . items ( ) :
    server_name = self . get_server_name ( key )
    servers . setdefault ( server_name , [ ] )
    servers [ server_name ] . append ( ( key , value ) )
for name , items in servers . items ( ) :
    self . connections [ name ] . mset ( dict ( items ) )
return True


signature:
def lock ( self , name , timeout = None , sleep = 0.1 ) :
    """Return a new Lock object using key ` ` name ` ` that mimics
 the behavior of threading . Lock .
 If specified , ` ` timeout ` ` indicates a maximum life for the lock .
 By default , it will remain locked until release ( ) is called .
 ` ` sleep ` ` indicates the amount of time to sleep per loop iteration
 when the lock is in blocking mode and another client is currently
 holding the lock ."""

implementation: return Lock ( self , name , timeout = timeout , sleep = sleep )


signature:
def parse_line ( line , document = None ) :
    '''Return a language - server diagnostic from a line of the Mypy error report ;
 optionally , use the whole document to provide more context on it .'''

implementation:
result = re . match ( line_pattern , line )
if result :
    _ , lineno , offset , severity , msg = result . groups ( )
    lineno = int ( lineno or 1 )
    offset = int ( offset or 0 )
    errno = 2
    if severity == 'error' :
        errno = 1
    diag = { 'source' : 'mypy' , 'range' : { 'start' : { 'line' : lineno - 1 , 'character' : offset } , # There may be a better solution , but mypy does not provide end
    'end' : { 'line' : lineno - 1 , 'character' : offset + 1 } } , 'message' : msg , 'severity' : errno }
    if document : # although mypy does not provide the end of the affected range , we
    # can make a good guess by highlighting the word that Mypy flagged
        word = document . word_at_position ( diag [ 'range' ] [ 'start' ] )
        if word :
            diag [ 'range' ] [ 'end' ] [ 'character' ] = ( diag [ 'range' ] [ 'start' ] [ 'character' ] + len ( word ) )
    return diag


signature:
async def connect ( self ) :
    """Establishes a connection to the Lavalink server ."""

implementation:
await self . _lavalink . bot . wait_until_ready ( )
if self . _ws and self . _ws . open :
    log . debug ( 'WebSocket still open, closing...' )
    await self . _ws . close ( )
user_id = self . _lavalink . bot . user . id
shard_count = self . _lavalink . bot . shard_count or self . _shards
headers = { 'Authorization' : self . _password , 'Num-Shards' : shard_count , 'User-Id' : str ( user_id ) }
log . debug ( 'Preparing to connect to Lavalink' )
log . debug ( '    with URI: {}' . format ( self . _uri ) )
log . debug ( '    with headers: {}' . format ( str ( headers ) ) )
log . info ( 'Connecting to Lavalink...' )
try :
    self . _ws = await websockets . connect ( self . _uri , loop = self . _loop , extra_headers = headers )
except OSError as error :
    log . exception ( 'Failed to connect to Lavalink: {}' . format ( str ( error ) ) )
else :
    log . info ( 'Connected to Lavalink!' )
    self . _loop . create_task ( self . listen ( ) )
    version = self . _ws . response_headers . get ( 'Lavalink-Major-Version' , 2 )
    try :
        self . _lavalink . _server_version = int ( version )
    except ValueError :
        self . _lavalink . _server_version = 2
    log . info ( 'Lavalink server version is {}' . format ( version ) )
    if self . _queue :
        log . info ( 'Replaying {} queued events...' . format ( len ( self . _queue ) ) )
        for task in self . _queue :
            await self . send ( ** task )


signature:
async def _attempt_reconnect ( self ) :
    """Attempts to reconnect to the Lavalink server .
 Returns
 bool
 ` ` True ` ` if the reconnection attempt was successful ."""

implementation:
log . info ( 'Connection closed; attempting to reconnect in 30 seconds' )
for a in range ( 0 , self . _ws_retry ) :
    await asyncio . sleep ( 30 )
    log . info ( 'Reconnecting... (Attempt {})' . format ( a + 1 ) )
    await self . connect ( )
    if self . _ws . open :
        return True
return False


signature:
async def listen ( self ) :
    """Waits to receive a payload from the Lavalink server and processes it ."""

implementation:
while not self . _shutdown :
    try :
        data = json . loads ( await self . _ws . recv ( ) )
    except websockets . ConnectionClosed as error :
        log . warning ( 'Disconnected from Lavalink: {}' . format ( str ( error ) ) )
        for g in self . _lavalink . players . _players . copy ( ) . keys ( ) :
            ws = self . _lavalink . bot . _connection . _get_websocket ( int ( g ) )
            await ws . voice_state ( int ( g ) , None )
        self . _lavalink . players . clear ( )
        if self . _shutdown :
            break
        if await self . _attempt_reconnect ( ) :
            return
        log . warning ( 'Unable to reconnect to Lavalink!' )
        break
    op = data . get ( 'op' , None )
    log . debug ( 'Received WebSocket data {}' . format ( str ( data ) ) )
    if not op :
        return log . debug ( 'Received WebSocket message without op {}' . format ( str ( data ) ) )
    if op == 'event' :
        log . debug ( 'Received event of type {}' . format ( data [ 'type' ] ) )
        player = self . _lavalink . players [ int ( data [ 'guildId' ] ) ]
        event = None
        if data [ 'type' ] == 'TrackEndEvent' :
            event = TrackEndEvent ( player , data [ 'track' ] , data [ 'reason' ] )
        elif data [ 'type' ] == 'TrackExceptionEvent' :
            event = TrackExceptionEvent ( player , data [ 'track' ] , data [ 'error' ] )
        elif data [ 'type' ] == 'TrackStuckEvent' :
            event = TrackStuckEvent ( player , data [ 'track' ] , data [ 'thresholdMs' ] )
        if event :
            await self . _lavalink . dispatch_event ( event )
    elif op == 'playerUpdate' :
        await self . _lavalink . update_state ( data )
    elif op == 'stats' :
        self . _lavalink . stats . _update ( data )
        await self . _lavalink . dispatch_event ( StatsUpdateEvent ( self . _lavalink . stats ) )
log . debug ( 'Closing WebSocket...' )
await self . _ws . close ( )


signature:
def connected_channel ( self ) :
    """Returns the voice channel the player is connected to ."""

implementation:
if not self . channel_id :
    return None
return self . _lavalink . bot . get_channel ( int ( self . channel_id ) )


signature:
async def connect ( self , channel_id : int ) :
    """Connects to a voice channel ."""

implementation:
ws = self . _lavalink . bot . _connection . _get_websocket ( int ( self . guild_id ) )
await ws . voice_state ( self . guild_id , str ( channel_id ) )


signature:
async def disconnect ( self ) :
    """Disconnects from the voice channel , if any ."""

implementation:
if not self . is_connected :
    return
await self . stop ( )
ws = self . _lavalink . bot . _connection . _get_websocket ( int ( self . guild_id ) )
await ws . voice_state ( self . guild_id , None )


signature:
def store ( self , key : object , value : object ) :
    """Stores custom user data ."""

implementation: self . _user_data . update ( { key : value } )


signature:
def fetch ( self , key : object , default = None ) :
    """Retrieves the related value from the stored user data ."""

implementation: return self . _user_data . get ( key , default )


signature:
def add ( self , requester : int , track : dict ) :
    """Adds a track to the queue ."""

implementation: self . queue . append ( AudioTrack ( ) . build ( track , requester ) )


signature:
def add_next ( self , requester : int , track : dict ) :
    """Adds a track to beginning of the queue"""

implementation: self . queue . insert ( 0 , AudioTrack ( ) . build ( track , requester ) )


signature:
def add_at ( self , index : int , requester : int , track : dict ) :
    """Adds a track at a specific index in the queue ."""

implementation: self . queue . insert ( min ( index , len ( self . queue ) - 1 ) , AudioTrack ( ) . build ( track , requester ) )


signature:
async def play ( self , track_index : int = 0 , ignore_shuffle : bool = False ) :
    """Plays the first track in the queue , if any or plays a track from the specified index in the queue ."""

implementation:
if self . repeat and self . current :
    self . queue . append ( self . current )
self . previous = self . current
self . current = None
self . position = 0
self . paused = False
if not self . queue :
    await self . stop ( )
    await self . _lavalink . dispatch_event ( QueueEndEvent ( self ) )
else :
    if self . shuffle and not ignore_shuffle :
        track = self . queue . pop ( randrange ( len ( self . queue ) ) )
    else :
        track = self . queue . pop ( min ( track_index , len ( self . queue ) - 1 ) )
    self . current = track
    await self . _lavalink . ws . send ( op = 'play' , guildId = self . guild_id , track = track . track )
    await self . _lavalink . dispatch_event ( TrackStartEvent ( self , track ) )


signature:
async def play_now ( self , requester : int , track : dict ) :
    """Add track and play it ."""

implementation:
self . add_next ( requester , track )
await self . play ( ignore_shuffle = True )


signature:
async def play_at ( self , index : int ) :
    """Play the queue from a specific point . Disregards tracks before the index ."""

implementation:
self . queue = self . queue [ min ( index , len ( self . queue ) - 1 ) : len ( self . queue ) ]
await self . play ( ignore_shuffle = True )


signature:
async def play_previous ( self ) :
    """Plays previous track if it exist , if it doesn ' t raises a NoPreviousTrack error ."""

implementation:
if not self . previous :
    raise NoPreviousTrack
self . queue . insert ( 0 , self . previous )
await self . play ( ignore_shuffle = True )


signature:
async def stop ( self ) :
    """Stops the player , if playing ."""

implementation:
await self . _lavalink . ws . send ( op = 'stop' , guildId = self . guild_id )
self . current = None


signature:
async def set_pause ( self , pause : bool ) :
    """Sets the player ' s paused state ."""

implementation:
await self . _lavalink . ws . send ( op = 'pause' , guildId = self . guild_id , pause = pause )
self . paused = pause


signature:
async def set_volume ( self , vol : int ) :
    """Sets the player ' s volume ( 150 % or 1000 % limit imposed by lavalink depending on the version ) ."""

implementation:
if self . _lavalink . _server_version <= 2 :
    self . volume = max ( min ( vol , 150 ) , 0 )
else :
    self . volume = max ( min ( vol , 1000 ) , 0 )
await self . _lavalink . ws . send ( op = 'volume' , guildId = self . guild_id , volume = self . volume )


signature:
async def seek ( self , pos : int ) :
    """Seeks to a given position in the track ."""

implementation: await self . _lavalink . ws . send ( op = 'seek' , guildId = self . guild_id , position = pos )


signature:
async def handle_event ( self , event ) :
    """Makes the player play the next song from the queue if a song has finished or an issue occurred ."""

implementation:
if isinstance ( event , ( TrackStuckEvent , TrackExceptionEvent ) ) or isinstance ( event , TrackEndEvent ) and event . reason == 'FINISHED' :
    await self . play ( )


signature:
def get ( self , guild_id ) :
    """Returns a player from the cache , or creates one if it does not exist ."""

implementation:
if guild_id not in self . _players :
    p = self . _player ( lavalink = self . lavalink , guild_id = guild_id )
    self . _players [ guild_id ] = p
return self . _players [ guild_id ]


signature:
def remove ( self , guild_id ) :
    """Removes a player from the current players ."""

implementation:
if guild_id in self . _players :
    self . _players [ guild_id ] . cleanup ( )
    del self . _players [ guild_id ]


signature:
async def _play ( self , ctx , * , query : str ) :
    """Searches and plays a song from a given query ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
query = query . strip ( '<>' )
if not url_rx . match ( query ) :
    query = f'ytsearch:{query}'
tracks = await self . bot . lavalink . get_tracks ( query )
if not tracks :
    return await ctx . send ( 'Nothing found!' )
embed = discord . Embed ( color = discord . Color . blurple ( ) )
if 'list' in query and 'ytsearch:' not in query :
    for track in tracks :
        player . add ( requester = ctx . author . id , track = track )
    embed . title = 'Playlist enqueued!'
    embed . description = f'Imported {len(tracks)} tracks from the playlist!'
    await ctx . send ( embed = embed )
else :
    track_title = tracks [ 0 ] [ "info" ] [ "title" ]
    track_uri = tracks [ 0 ] [ "info" ] [ "uri" ]
    embed . title = "Track enqueued!"
    embed . description = f'[{track_title}]({track_uri})'
    player . add ( requester = ctx . author . id , track = tracks [ 0 ] )
if not player . is_playing :
    await player . play ( )


signature:
async def _seek ( self , ctx , * , time : str ) :
    """Seeks to a given position in a track ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . is_playing :
    return await ctx . send ( 'Not playing.' )
seconds = time_rx . search ( time )
if not seconds :
    return await ctx . send ( 'You need to specify the amount of seconds to skip!' )
seconds = int ( seconds . group ( ) ) * 1000
if time . startswith ( '-' ) :
    seconds *= - 1
track_time = player . position + seconds
await player . seek ( track_time )
await ctx . send ( f'Moved track to **{lavalink.Utils.format_time(track_time)}**' )


signature:
async def _skip ( self , ctx ) :
    """Skips the current track ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . is_playing :
    return await ctx . send ( 'Not playing.' )
await player . skip ( )
await ctx . send ( '⏭ | Skipped.' )


signature:
async def _stop ( self , ctx ) :
    """Stops the player and clears its queue ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . is_playing :
    return await ctx . send ( 'Not playing.' )
player . queue . clear ( )
await player . stop ( )
await ctx . send ( '⏹ | Stopped.' )


signature:
async def _now ( self , ctx ) :
    """Shows some stats about the currently playing song ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
song = 'Nothing'
if player . current :
    position = lavalink . Utils . format_time ( player . position )
    if player . current . stream :
        duration = '🔴 LIVE'
    else :
        duration = lavalink . Utils . format_time ( player . current . duration )
    song = f'**[{player.current.title}]({player.current.uri})**\n({position}/{duration})'
embed = discord . Embed ( color = discord . Color . blurple ( ) , title = 'Now Playing' , description = song )
await ctx . send ( embed = embed )


signature:
async def _queue ( self , ctx , page : int = 1 ) :
    """Shows the player ' s queue ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . queue :
    return await ctx . send ( 'There\'s nothing in the queue! Why not queue something?' )
items_per_page = 10
pages = math . ceil ( len ( player . queue ) / items_per_page )
start = ( page - 1 ) * items_per_page
end = start + items_per_page
queue_list = ''
for index , track in enumerate ( player . queue [ start : end ] , start = start ) :
    queue_list += f'`{index + 1}.` [**{track.title}**]({track.uri})\n'
embed = discord . Embed ( colour = discord . Color . blurple ( ) , description = f'**{len(player.queue)} tracks**\n\n{queue_list}' )
embed . set_footer ( text = f'Viewing page {page}/{pages}' )
await ctx . send ( embed = embed )


signature:
async def _pause ( self , ctx ) :
    """Pauses / Resumes the current track ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . is_playing :
    return await ctx . send ( 'Not playing.' )
if player . paused :
    await player . set_pause ( False )
    await ctx . send ( '⏯ | Resumed' )
else :
    await player . set_pause ( True )
    await ctx . send ( '⏯ | Paused' )


signature:
async def _volume ( self , ctx , volume : int = None ) :
    """Changes the player ' s volume . Must be between 0 and 150 . Error Handling for that is done by Lavalink ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not volume :
    return await ctx . send ( f'🔈 | {player.volume}%' )
await player . set_volume ( volume )
await ctx . send ( f'🔈 | Set to {player.volume}%' )


signature:
async def _shuffle ( self , ctx ) :
    """Shuffles the player ' s queue ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . is_playing :
    return await ctx . send ( 'Nothing playing.' )
player . shuffle = not player . shuffle
await ctx . send ( '🔀 | Shuffle ' + ( 'enabled' if player . shuffle else 'disabled' ) )


signature:
async def _repeat ( self , ctx ) :
    """Repeats the current song until the command is invoked again ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . is_playing :
    return await ctx . send ( 'Nothing playing.' )
player . repeat = not player . repeat
await ctx . send ( '🔁 | Repeat ' + ( 'enabled' if player . repeat else 'disabled' ) )


signature:
async def _remove ( self , ctx , index : int ) :
    """Removes an item from the player ' s queue with the given index ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . queue :
    return await ctx . send ( 'Nothing queued.' )
if index > len ( player . queue ) or index < 1 :
    return await ctx . send ( f'Index has to be **between** 1 and {len(player.queue)}' )
index -= 1
removed = player . queue . pop ( index )
await ctx . send ( f'Removed **{removed.title}** from the queue.' )


signature:
async def _disconnect ( self , ctx ) :
    """Disconnects the player from the voice channel and clears its queue ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . is_connected :
    return await ctx . send ( 'Not connected.' )
if not ctx . author . voice or ( player . is_connected and ctx . author . voice . channel . id != int ( player . channel_id ) ) :
    return await ctx . send ( 'You\'re not in my voicechannel!' )
player . queue . clear ( )
await player . disconnect ( )
await ctx . send ( '*⃣ | Disconnected.' )


signature:
async def ensure_voice ( self , ctx ) :
    """A few checks to make sure the bot can join a voice channel ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . is_connected :
    if not ctx . author . voice or not ctx . author . voice . channel :
        await ctx . send ( 'You aren\'t connected to any voice channel.' )
        raise commands . CommandInvokeError ( 'Author not connected to voice channel.' )
    permissions = ctx . author . voice . channel . permissions_for ( ctx . me )
    if not permissions . connect or not permissions . speak :
        await ctx . send ( 'Missing permissions `CONNECT` and/or `SPEAK`.' )
        raise commands . CommandInvokeError ( 'Bot has no permissions CONNECT and/or SPEAK' )
    player . store ( 'channel' , ctx . channel . id )
    await player . connect ( ctx . author . voice . channel . id )
else :
    if player . connected_channel . id != ctx . author . voice . channel . id :
        return await ctx . send ( 'Join my voice channel!' )


signature:
def register_hook ( self , func ) :
    """Registers a hook . Since this probably is a bit difficult , I ' ll explain it in detail .
 A hook basically is an object of a function you pass . This will append that object to a list and whenever
 an event from the Lavalink server is dispatched , the function will be called internally . For declaring the
 function that should become a hook , pass ` ` event ` as its sole parameter .
 Can be a function but also a coroutine .
 Example for a method declaration inside a class :
 self . bot . lavalink . register _ hook ( my _ hook )
 async def my _ hook ( self , event ) :
 channel = self . bot . get _ channel ( event . player . fetch ( ' channel ' ) )
 if not channel :
 return
 if isinstance ( event , lavalink . Events . TrackStartEvent ) :
 await channel . send ( embed = discord . Embed ( title = ' Now playing : ' ,
 description = event . track . title ,
 color = discord . Color . blurple ( ) ) )
 : param func :
 The function that should be registered as a hook ."""

implementation:
if func not in self . hooks :
    self . hooks . append ( func )


signature:
def unregister_hook ( self , func ) :
    """Unregisters a hook . For further explanation , please have a look at ` ` register _ hook ` ` ."""

implementation:
if func in self . hooks :
    self . hooks . remove ( func )


signature:
async def dispatch_event ( self , event ) :
    """Dispatches an event to all registered hooks ."""

implementation:
log . debug ( 'Dispatching event of type {} to {} hooks' . format ( event . __class__ . __name__ , len ( self . hooks ) ) )
for hook in self . hooks :
    try :
        if asyncio . iscoroutinefunction ( hook ) :
            await hook ( event )
        else :
            hook ( event )
    except Exception as e : # pylint : disable = broad - except
    # Catch generic exception thrown by user hooks
        log . warning ( 'Encountered exception while dispatching an event to hook `{}` ({})' . format ( hook . __name__ , str ( e ) ) )
if isinstance ( event , ( TrackEndEvent , TrackExceptionEvent , TrackStuckEvent ) ) and event . player :
    await event . player . handle_event ( event )


signature:
async def update_state ( self , data ) :
    """Updates a player ' s state when a payload with opcode ` ` playerUpdate ` ` is received ."""

implementation:
guild_id = int ( data [ 'guildId' ] )
if guild_id in self . players :
    player = self . players . get ( guild_id )
    player . position = data [ 'state' ] . get ( 'position' , 0 )
    player . position_timestamp = data [ 'state' ] [ 'time' ]


signature:
async def get_tracks ( self , query ) :
    """Returns a Dictionary containing search results for a given query ."""

implementation:
log . debug ( 'Requesting tracks for query {}' . format ( query ) )
async with self . http . get ( self . rest_uri + quote ( query ) , headers = { 'Authorization' : self . password } ) as res :
    return await res . json ( content_type = None )


signature:
async def on_socket_response ( self , data ) :
    """This coroutine will be called every time an event from Discord is received .
 It is used to update a player ' s voice state through forwarding a payload via the WebSocket connection to Lavalink .
 : param data :
 The payload received from Discord ."""

implementation:
# INTERCEPT VOICE UPDATES
if not data or data . get ( 't' , '' ) not in [ 'VOICE_STATE_UPDATE' , 'VOICE_SERVER_UPDATE' ] :
    return
if data [ 't' ] == 'VOICE_SERVER_UPDATE' :
    self . voice_state . update ( { 'op' : 'voiceUpdate' , 'guildId' : data [ 'd' ] [ 'guild_id' ] , 'event' : data [ 'd' ] } )
else :
    if int ( data [ 'd' ] [ 'user_id' ] ) != self . bot . user . id :
        return
    self . voice_state . update ( { 'sessionId' : data [ 'd' ] [ 'session_id' ] } )
    guild_id = int ( data [ 'd' ] [ 'guild_id' ] )
    if self . players [ guild_id ] :
        self . players [ guild_id ] . channel_id = data [ 'd' ] [ 'channel_id' ]
if { 'op' , 'guildId' , 'sessionId' , 'event' } == self . voice_state . keys ( ) :
    await self . ws . send ( ** self . voice_state )
    self . voice_state . clear ( )


signature:
def destroy ( self ) :
    """Destroys the Lavalink client ."""

implementation:
self . ws . destroy ( )
self . bot . remove_listener ( self . on_socket_response )
self . hooks . clear ( )


signature:
def format_time ( time ) :
    """Formats the given time into HH : MM : SS ."""

implementation:
hours , remainder = divmod ( time / 1000 , 3600 )
minutes , seconds = divmod ( remainder , 60 )
return '%02d:%02d:%02d' % ( hours , minutes , seconds )


signature:
def build ( self , track , requester ) :
    """Returns an optional AudioTrack ."""

implementation:
try :
    self . track = track [ 'track' ]
    self . identifier = track [ 'info' ] [ 'identifier' ]
    self . can_seek = track [ 'info' ] [ 'isSeekable' ]
    self . author = track [ 'info' ] [ 'author' ]
    self . duration = track [ 'info' ] [ 'length' ]
    self . stream = track [ 'info' ] [ 'isStream' ]
    self . title = track [ 'info' ] [ 'title' ]
    self . uri = track [ 'info' ] [ 'uri' ]
    self . requester = requester
    return self
except KeyError :
    raise InvalidTrack ( 'An invalid track was passed.' )


signature:
async def _previous ( self , ctx ) :
    """Plays the previous song ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
try :
    await player . play_previous ( )
except lavalink . NoPreviousTrack :
    await ctx . send ( 'There is no previous song to play.' )


signature:
async def _playnow ( self , ctx , * , query : str ) :
    """Plays immediately a song ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if not player . queue and not player . is_playing :
    return await ctx . invoke ( self . _play , query = query )
query = query . strip ( '<>' )
if not url_rx . match ( query ) :
    query = f'ytsearch:{query}'
results = await self . bot . lavalink . get_tracks ( query )
if not results or not results [ 'tracks' ] :
    return await ctx . send ( 'Nothing found!' )
tracks = results [ 'tracks' ]
track = tracks . pop ( 0 )
if results [ 'loadType' ] == 'PLAYLIST_LOADED' :
    for _track in tracks :
        player . add ( requester = ctx . author . id , track = _track )
await player . play_now ( requester = ctx . author . id , track = track )


signature:
async def _playat ( self , ctx , index : int ) :
    """Plays the queue from a specific point . Disregards tracks before the index ."""

implementation:
player = self . bot . lavalink . players . get ( ctx . guild . id )
if index < 1 :
    return await ctx . send ( 'Invalid specified index.' )
if len ( player . queue ) < index :
    return await ctx . send ( 'This index exceeds the queue\'s length.' )
await player . play_at ( index - 1 )


signature:
async def _find ( self , ctx , * , query ) :
    """Lists the first 10 search results from a given query ."""

implementation:
if not query . startswith ( 'ytsearch:' ) and not query . startswith ( 'scsearch:' ) :
    query = 'ytsearch:' + query
results = await self . bot . lavalink . get_tracks ( query )
if not results or not results [ 'tracks' ] :
    return await ctx . send ( 'Nothing found' )
tracks = results [ 'tracks' ] [ : 10 ]
# First 10 results
o = ''
for index , track in enumerate ( tracks , start = 1 ) :
    track_title = track [ "info" ] [ "title" ]
    track_uri = track [ "info" ] [ "uri" ]
    o += f'`{index}.` [{track_title}]({track_uri})\n'
embed = discord . Embed ( color = discord . Color . blurple ( ) , description = o )
await ctx . send ( embed = embed )


signature:
def add_suggestions ( self , * suggestions , ** kwargs ) :
    """Add suggestion terms to the AutoCompleter engine . Each suggestion has a score and string .
 If kwargs [ ' increment ' ] is true and the terms are already in the server ' s dictionary , we increment their scores"""

implementation:
pipe = self . redis . pipeline ( )
for sug in suggestions :
    args = [ AutoCompleter . SUGADD_COMMAND , self . key , sug . string , sug . score ]
    if kwargs . get ( 'increment' ) :
        args . append ( AutoCompleter . INCR )
    if sug . payload :
        args . append ( 'PAYLOAD' )
        args . append ( sug . payload )
    pipe . execute_command ( * args )
return pipe . execute ( ) [ - 1 ]


signature:
def delete ( self , string ) :
    """Delete a string from the AutoCompleter index .
 Returns 1 if the string was found and deleted , 0 otherwise"""

implementation: return self . redis . execute_command ( AutoCompleter . SUGDEL_COMMAND , self . key , string )


signature:
def get_suggestions ( self , prefix , fuzzy = False , num = 10 , with_scores = False , with_payloads = False ) :
    """Get a list of suggestions from the AutoCompleter , for a given prefix
 # # # Parameters :
 - * * prefix * * : the prefix we are searching . * * Must be valid ascii or utf - 8 * *
 - * * fuzzy * * : If set to true , the prefix search is done in fuzzy mode .
 * * NOTE * * : Running fuzzy searches on short ( < 3 letters ) prefixes can be very slow , and even scan the entire index .
 - * * with _ scores * * : if set to true , we also return the ( refactored ) score of each suggestion .
 This is normally not needed , and is NOT the original score inserted into the index
 - * * with _ payloads * * : Return suggestion payloads
 - * * num * * : The maximum number of results we return . Note that we might return less . The algorithm trims irrelevant suggestions .
 Returns a list of Suggestion objects . If with _ scores was False , the score of all suggestions is 1."""

implementation:
args = [ AutoCompleter . SUGGET_COMMAND , self . key , prefix , 'MAX' , num ]
if fuzzy :
    args . append ( AutoCompleter . FUZZY )
if with_scores :
    args . append ( AutoCompleter . WITHSCORES )
if with_payloads :
    args . append ( AutoCompleter . WITHPAYLOADS )
ret = self . redis . execute_command ( * args )
results = [ ]
if not ret :
    return results
parser = SuggestionParser ( with_scores , with_payloads , ret )
return [ s for s in parser ]


signature:
def create_index ( self , fields , no_term_offsets = False , no_field_flags = False , stopwords = None ) :
    """Create the search index . The index must not already exist .
 # # # Parameters :
 - * * fields * * : a list of TextField or NumericField objects
 - * * no _ term _ offsets * * : If true , we will not save term offsets in the index
 - * * no _ field _ flags * * : If true , we will not save field flags that allow searching in specific fields
 - * * stopwords * * : If not None , we create the index with this custom stopword list . The list can be empty"""

implementation:
args = [ self . CREATE_CMD , self . index_name ]
if no_term_offsets :
    args . append ( self . NOOFFSETS )
if no_field_flags :
    args . append ( self . NOFIELDS )
if stopwords is not None and isinstance ( stopwords , ( list , tuple , set ) ) :
    args += [ self . STOPWORDS , len ( stopwords ) ]
    if len ( stopwords ) > 0 :
        args += list ( stopwords )
args . append ( 'SCHEMA' )
args += list ( itertools . chain ( * ( f . redis_args ( ) for f in fields ) ) )
return self . redis . execute_command ( * args )


signature:
def _add_document ( self , doc_id , conn = None , nosave = False , score = 1.0 , payload = None , replace = False , partial = False , language = None , ** fields ) :
    """Internal add _ document used for both batch and single doc indexing"""

implementation:
if conn is None :
    conn = self . redis
if partial :
    replace = True
args = [ self . ADD_CMD , self . index_name , doc_id , score ]
if nosave :
    args . append ( 'NOSAVE' )
if payload is not None :
    args . append ( 'PAYLOAD' )
    args . append ( payload )
if replace :
    args . append ( 'REPLACE' )
    if partial :
        args . append ( 'PARTIAL' )
if language :
    args += [ 'LANGUAGE' , language ]
args . append ( 'FIELDS' )
args += list ( itertools . chain ( * fields . items ( ) ) )
return conn . execute_command ( * args )


signature:
def add_document ( self , doc_id , nosave = False , score = 1.0 , payload = None , replace = False , partial = False , language = None , ** fields ) :
    """Add a single document to the index .
 # # # Parameters
 - * * doc _ id * * : the id of the saved document .
 - * * nosave * * : if set to true , we just index the document , and don ' t save a copy of it . This means that searches will just return ids .
 - * * score * * : the document ranking , between 0.0 and 1.0
 - * * payload * * : optional inner - index payload we can save for fast access in scoring functions
 - * * replace * * : if True , and the document already is in the index , we perform an update and reindex the document
 - * * partial * * : if True , the fields specified will be added to the existing document .
 This has the added benefit that any fields specified with ` no _ index `
 will not be reindexed again . Implies ` replace `
 - * * language * * : Specify the language used for document tokenization .
 - * * fields * * kwargs dictionary of the document fields to be saved and / or indexed .
 NOTE : Geo points shoule be encoded as strings of " lon , lat " """

implementation: return self . _add_document ( doc_id , conn = None , nosave = nosave , score = score , payload = payload , replace = replace , partial = partial , language = language , ** fields )


signature:
def delete_document ( self , doc_id , conn = None ) :
    """Delete a document from index
 Returns 1 if the document was deleted , 0 if not"""

implementation:
if conn is None :
    conn = self . redis
return conn . execute_command ( self . DEL_CMD , self . index_name , doc_id )


signature:
def load_document ( self , id ) :
    """Load a single document by id"""

implementation:
fields = self . redis . hgetall ( id )
if six . PY3 :
    f2 = { to_string ( k ) : to_string ( v ) for k , v in fields . items ( ) }
    fields = f2
try :
    del fields [ 'id' ]
except KeyError :
    pass
return Document ( id = id , ** fields )


signature:
def info ( self ) :
    """Get info an stats about the the current index , including the number of documents , memory consumption , etc"""

implementation:
res = self . redis . execute_command ( 'FT.INFO' , self . index_name )
it = six . moves . map ( to_string , res )
return dict ( six . moves . zip ( it , it ) )


signature:
def search ( self , query ) :
    """Search the index for a given query , and return a result of documents
 # # # Parameters
 - * * query * * : the search query . Either a text for simple queries with default parameters , or a Query object for complex queries .
 See RediSearch ' s documentation on query format
 - * * snippet _ sizes * * : A dictionary of { field : snippet _ size } used to trim and format the result . e . g . e { ' body ' : 500}"""

implementation:
args , query = self . _mk_query_args ( query )
st = time . time ( )
res = self . redis . execute_command ( self . SEARCH_CMD , * args )
return Result ( res , not query . _no_content , duration = ( time . time ( ) - st ) * 1000.0 , has_payload = query . _with_payloads )


signature:
def aggregate ( self , query ) :
    """Issue an aggregation query
 # # # Parameters
 * * query * * : This can be either an ` AggeregateRequest ` , or a ` Cursor `
 An ` AggregateResult ` object is returned . You can access the rows from its
 ` rows ` property , which will always yield the rows of the result"""

implementation:
if isinstance ( query , AggregateRequest ) :
    has_schema = query . _with_schema
    has_cursor = bool ( query . _cursor )
    cmd = [ self . AGGREGATE_CMD , self . index_name ] + query . build_args ( )
elif isinstance ( query , Cursor ) :
    has_schema = False
    has_cursor = True
    cmd = [ self . CURSOR_CMD , 'READ' , self . index_name ] + query . build_args ( )
else :
    raise ValueError ( 'Bad query' , query )
raw = self . redis . execute_command ( * cmd )
if has_cursor :
    if isinstance ( query , Cursor ) :
        query . cid = raw [ 1 ]
        cursor = query
    else :
        cursor = Cursor ( raw [ 1 ] )
    raw = raw [ 0 ]
else :
    cursor = None
if query . _with_schema :
    schema = raw [ 0 ]
    rows = raw [ 2 : ]
else :
    schema = None
    rows = raw [ 1 : ]
res = AggregateResult ( rows , cursor , schema )
return res


signature:
def alias ( self , alias ) :
    """Set the alias for this reducer .
 # # # Parameters
 - * * alias * * : The value of the alias for this reducer . If this is the
 special value ` aggregation . FIELDNAME ` then this reducer will be
 aliased using the same name as the field upon which it operates .
 Note that using ` FIELDNAME ` is only possible on reducers which
 operate on a single field value .
 This method returns the ` Reducer ` object making it suitable for
 chaining ."""

implementation:
if alias is FIELDNAME :
    if not self . _field :
        raise ValueError ( "Cannot use FIELDNAME alias with no field" )
    # Chop off initial ' @ '
    alias = self . _field [ 1 : ]
self . _alias = alias
return self


signature:
def group_by ( self , fields , * reducers ) :
    """Specify by which fields to group the aggregation .
 # # # Parameters
 - * * fields * * : Fields to group by . This can either be a single string ,
 or a list of strings . both cases , the field should be specified as
 ` @ field ` .
 - * * reducers * * : One or more reducers . Reducers may be found in the
 ` aggregation ` module ."""

implementation:
group = Group ( fields , reducers )
self . _groups . append ( group )
return self


signature:
def apply ( self , ** kwexpr ) :
    """Specify one or more projection expressions to add to each result
 # # # Parameters
 - * * kwexpr * * : One or more key - value pairs for a projection . The key is
 the alias for the projection , and the value is the projection
 expression itself , for example ` apply ( square _ root = " sqrt ( @ foo ) " ) `"""

implementation:
for alias , expr in kwexpr . items ( ) :
    self . _projections . append ( [ alias , expr ] )
return self


signature:
def limit ( self , offset , num ) :
    """Sets the limit for the most recent group or query .
 If no group has been defined yet ( via ` group _ by ( ) ` ) then this sets
 the limit for the initial pool of results from the query . Otherwise ,
 this limits the number of items operated on from the previous group .
 Setting a limit on the initial search results may be useful when
 attempting to execute an aggregation on a sample of a large data set .
 # # # Parameters
 - * * offset * * : Result offset from which to begin paging
 - * * num * * : Number of results to return
 Example of sorting the initial results :
 AggregateRequest ( ' @ sale _ amount : [ 10000 , inf ] ' ) . limit ( 0 , 10 ) . group _ by ( ' @ state ' , r . count ( ) )
 Will only group by the states found in the first 10 results of the
 query ` @ sale _ amount : [ 10000 , inf ] ` . On the other hand ,
 AggregateRequest ( ' @ sale _ amount : [ 10000 , inf ] ' ) . limit ( 0 , 1000 ) . group _ by ( ' @ state ' , r . count ( ) . limit ( 0 , 10)
 Will group all the results matching the query , but only return the
 first 10 groups .
 If you only wish to return a * top - N * style query , consider using
 ` sort _ by ( ) ` instead ."""

implementation:
limit = Limit ( offset , num )
if self . _groups :
    self . _groups [ - 1 ] . limit = limit
else :
    self . _limit = limit
return self


signature:
def sort_by ( self , * fields , ** kwargs ) :
    """Indicate how the results should be sorted . This can also be used for
 * top - N * style queries
 # # # Parameters
 - * * fields * * : The fields by which to sort . This can be either a single
 field or a list of fields . If you wish to specify order , you can
 use the ` Asc ` or ` Desc ` wrapper classes .
 - * * max * * : Maximum number of results to return . This can be used instead
 of ` LIMIT ` and is also faster .
 Example of sorting by ` foo ` ascending and ` bar ` descending :
 sort _ by ( Asc ( ' @ foo ' ) , Desc ( ' @ bar ' ) )
 Return the top 10 customers :
 AggregateRequest ( ) . group _ by ( ' @ customer ' , r . sum ( ' @ paid ' ) . alias ( FIELDNAME ) ) . sort _ by ( Desc ( ' @ paid ' ) , max = 10)"""

implementation:
self . _max = kwargs . get ( 'max' , 0 )
if isinstance ( fields , ( string_types , SortDirection ) ) :
    fields = [ fields ]
for f in fields :
    if isinstance ( f , SortDirection ) :
        self . _sortby += [ f . field , f . DIRSTRING ]
    else :
        self . _sortby . append ( f )
return self


signature:
def summarize ( self , fields = None , context_len = None , num_frags = None , sep = None ) :
    """Return an abridged format of the field , containing only the segments of
 the field which contain the matching term ( s ) .
 If ` fields ` is specified , then only the mentioned fields are
 summarized ; otherwise all results are summarized .
 Server side defaults are used for each option ( except ` fields ` ) if not specified
 - * * fields * * List of fields to summarize . All fields are summarized if not specified
 - * * context _ len * * Amount of context to include with each fragment
 - * * num _ frags * * Number of fragments per document
 - * * sep * * Separator string to separate fragments"""

implementation:
args = [ 'SUMMARIZE' ]
fields = self . _mk_field_list ( fields )
if fields :
    args += [ 'FIELDS' , str ( len ( fields ) ) ] + fields
if context_len is not None :
    args += [ 'LEN' , str ( context_len ) ]
if num_frags is not None :
    args += [ 'FRAGS' , str ( num_frags ) ]
if sep is not None :
    args += [ 'SEPARATOR' , sep ]
self . _summarize_fields = args
return self


signature:
def highlight ( self , fields = None , tags = None ) :
    """Apply specified markup to matched term ( s ) within the returned field ( s )
 - * * fields * * If specified then only those mentioned fields are highlighted , otherwise all fields are highlighted
 - * * tags * * A list of two strings to surround the match ."""

implementation:
args = [ 'HIGHLIGHT' ]
fields = self . _mk_field_list ( fields )
if fields :
    args += [ 'FIELDS' , str ( len ( fields ) ) ] + fields
if tags :
    args += [ 'TAGS' ] + list ( tags )
self . _highlight_fields = args
return self


signature:
def get_args ( self ) :
    """Format the redis arguments for this query and return them"""

implementation:
args = [ self . _query_string ]
if self . _no_content :
    args . append ( 'NOCONTENT' )
if self . _fields :
    args . append ( 'INFIELDS' )
    args . append ( len ( self . _fields ) )
    args += self . _fields
if self . _verbatim :
    args . append ( 'VERBATIM' )
if self . _no_stopwords :
    args . append ( 'NOSTOPWORDS' )
if self . _filters :
    for flt in self . _filters :
        assert isinstance ( flt , Filter )
        args += flt . args
if self . _with_payloads :
    args . append ( 'WITHPAYLOADS' )
if self . _ids :
    args . append ( 'INKEYS' )
    args . append ( len ( self . _ids ) )
    args += self . _ids
if self . _slop >= 0 :
    args += [ 'SLOP' , self . _slop ]
if self . _in_order :
    args . append ( 'INORDER' )
if self . _return_fields :
    args . append ( 'RETURN' )
    args . append ( len ( self . _return_fields ) )
    args += self . _return_fields
if self . _sortby :
    assert isinstance ( self . _sortby , SortbyField )
    args . append ( 'SORTBY' )
    args += self . _sortby . args
if self . _language :
    args += [ 'LANGUAGE' , self . _language ]
args += self . _summarize_fields + self . _highlight_fields
args += [ "LIMIT" , self . _offset , self . _num ]
return args


signature:
def paging ( self , offset , num ) :
    """Set the paging for the query ( defaults to 0 . . 10 ) .
 - * * offset * * : Paging offset for the results . Defaults to 0
 - * * num * * : How many results do we want"""

implementation:
self . _offset = offset
self . _num = num
return self


signature:
def sort_by ( self , field , asc = True ) :
    """Add a sortby field to the query
 - * * field * * - the name of the field to sort by
 - * * asc * * - when ` True ` , sorting will be done in asceding order"""

implementation:
self . _sortby = SortbyField ( field , asc )
return self


signature:
def between ( a , b , inclusive_min = True , inclusive_max = True ) :
    """Indicate that value is a numeric range"""

implementation: return RangeValue ( a , b , inclusive_min = inclusive_min , inclusive_max = inclusive_max )


signature:
def geo ( lat , lon , radius , unit = 'km' ) :
    """Indicate that value is a geo region"""

implementation: return GeoValue ( lat , lon , radius , unit )