signature:
def delete_folder ( self , token , folder_id ) :
    """Delete the folder with the passed in folder _ id .
 : param token : A valid token for the user in question .
 : type token : string
 : param folder _ id : The id of the folder to be deleted .
 : type folder _ id : int | long
 : returns : None .
 : rtype : None"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'id' ] = folder_id
response = self . request ( 'midas.folder.delete' , parameters )
return response


signature:
def move_folder ( self , token , folder_id , dest_folder_id ) :
    """Move a folder to the destination folder .
 : param token : A valid token for the user in question .
 : type token : string
 : param folder _ id : The id of the folder to be moved .
 : type folder _ id : int | long
 : param dest _ folder _ id : The id of destination ( new parent ) folder .
 : type dest _ folder _ id : int | long
 : returns : Dictionary containing the details of the moved folder .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'id' ] = folder_id
parameters [ 'dstfolderid' ] = dest_folder_id
response = self . request ( 'midas.folder.move' , parameters )
return response


signature:
def create_item ( self , token , name , parent_id , ** kwargs ) :
    """Create an item to the server .
 : param token : A valid token for the user in question .
 : type token : string
 : param name : The name of the item to be created .
 : type name : string
 : param parent _ id : The id of the destination folder .
 : type parent _ id : int | long
 : param description : ( optional ) The description text of the item .
 : type description : string
 : param uuid : ( optional ) The UUID for the item . It will be generated if
 not given .
 : type uuid : string
 : param privacy : ( optional ) The privacy state of the item
 ( ' Public ' or ' Private ' ) .
 : type privacy : string
 : returns : Dictionary containing the details of the created item .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'name' ] = name
parameters [ 'parentid' ] = parent_id
optional_keys = [ 'description' , 'uuid' , 'privacy' ]
for key in optional_keys :
    if key in kwargs :
        parameters [ key ] = kwargs [ key ]
response = self . request ( 'midas.item.create' , parameters )
return response


signature:
def item_get ( self , token , item_id ) :
    """Get the attributes of the specified item .
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : The id of the requested item .
 : type item _ id : int | string
 : returns : Dictionary of the item attributes .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'id' ] = item_id
response = self . request ( 'midas.item.get' , parameters )
return response


signature:
def download_item ( self , item_id , token = None , revision = None ) :
    """Download an item to disk .
 : param item _ id : The id of the item to be downloaded .
 : type item _ id : int | long
 : param token : ( optional ) The authentication token of the user
 requesting the download .
 : type token : None | string
 : param revision : ( optional ) The revision of the item to download , this
 defaults to HEAD .
 : type revision : None | int | long
 : returns : A tuple of the filename and the content iterator .
 : rtype : ( string , unknown )"""

implementation:
parameters = dict ( )
parameters [ 'id' ] = item_id
if token :
    parameters [ 'token' ] = token
if revision :
    parameters [ 'revision' ] = revision
method_url = self . full_url + 'midas.item.download'
request = requests . get ( method_url , params = parameters , stream = True , verify = self . _verify_ssl_certificate )
filename = request . headers [ 'content-disposition' ] [ 21 : ] . strip ( '"' )
return filename , request . iter_content ( chunk_size = 10 * 1024 )


signature:
def delete_item ( self , token , item_id ) :
    """Delete the item with the passed in item _ id .
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : The id of the item to be deleted .
 : type item _ id : int | long
 : returns : None .
 : rtype : None"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'id' ] = item_id
response = self . request ( 'midas.item.delete' , parameters )
return response


signature:
def get_item_metadata ( self , item_id , token = None , revision = None ) :
    """Get the metadata associated with an item .
 : param item _ id : The id of the item for which metadata will be returned
 : type item _ id : int | long
 : param token : ( optional ) A valid token for the user in question .
 : type token : None | string
 : param revision : ( optional ) Revision of the item . Defaults to latest
 revision .
 : type revision : int | long
 : returns : List of dictionaries containing item metadata .
 : rtype : list [ dict ]"""

implementation:
parameters = dict ( )
parameters [ 'id' ] = item_id
if token :
    parameters [ 'token' ] = token
if revision :
    parameters [ 'revision' ] = revision
response = self . request ( 'midas.item.getmetadata' , parameters )
return response


signature:
def set_item_metadata ( self , token , item_id , element , value , qualifier = None ) :
    """Set the metadata associated with an item .
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : The id of the item for which metadata will be set .
 : type item _ id : int | long
 : param element : The metadata element name .
 : type element : string
 : param value : The metadata value for the field .
 : type value : string
 : param qualifier : ( optional ) The metadata qualifier . Defaults to empty
 string .
 : type qualifier : None | string
 : returns : None .
 : rtype : None"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'itemId' ] = item_id
parameters [ 'element' ] = element
parameters [ 'value' ] = value
if qualifier :
    parameters [ 'qualifier' ] = qualifier
response = self . request ( 'midas.item.setmetadata' , parameters )
return response


signature:
def share_item ( self , token , item_id , dest_folder_id ) :
    """Share an item to the destination folder .
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : The id of the item to be shared .
 : type item _ id : int | long
 : param dest _ folder _ id : The id of destination folder where the item is
 shared to .
 : type dest _ folder _ id : int | long
 : returns : Dictionary containing the details of the shared item .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'id' ] = item_id
parameters [ 'dstfolderid' ] = dest_folder_id
response = self . request ( 'midas.item.share' , parameters )
return response


signature:
def move_item ( self , token , item_id , src_folder_id , dest_folder_id ) :
    """Move an item from the source folder to the destination folder .
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : The id of the item to be moved
 : type item _ id : int | long
 : param src _ folder _ id : The id of source folder where the item is located
 : type src _ folder _ id : int | long
 : param dest _ folder _ id : The id of destination folder where the item is
 moved to
 : type dest _ folder _ id : int | long
 : returns : Dictionary containing the details of the moved item
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'id' ] = item_id
parameters [ 'srcfolderid' ] = src_folder_id
parameters [ 'dstfolderid' ] = dest_folder_id
response = self . request ( 'midas.item.move' , parameters )
return response


signature:
def search_item_by_name ( self , name , token = None ) :
    """Return all items .
 : param name : The name of the item to search by .
 : type name : string
 : param token : ( optional ) A valid token for the user in question .
 : type token : None | string
 : returns : A list of all items with the given name .
 : rtype : list [ dict ]"""

implementation:
parameters = dict ( )
parameters [ 'name' ] = name
if token :
    parameters [ 'token' ] = token
response = self . request ( 'midas.item.searchbyname' , parameters )
return response [ 'items' ]


signature:
def search_item_by_name_and_folder ( self , name , folder_id , token = None ) :
    """Return all items with a given name and parent folder id .
 : param name : The name of the item to search by .
 : type name : string
 : param folder _ id : The id of the parent folder to search by .
 : type folder _ id : int | long
 : param token : ( optional ) A valid token for the user in question .
 : type token : None | string
 : returns : A list of all items with the given name and parent folder id .
 : rtype : list [ dict ]"""

implementation:
parameters = dict ( )
parameters [ 'name' ] = name
parameters [ 'folderId' ] = folder_id
if token :
    parameters [ 'token' ] = token
response = self . request ( 'midas.item.searchbynameandfolder' , parameters )
return response [ 'items' ]


signature:
def search_item_by_name_and_folder_name ( self , name , folder_name , token = None ) :
    """Return all items with a given name and parent folder name .
 : param name : The name of the item to search by .
 : type name : string
 : param folder _ name : The name of the parent folder to search by .
 : type folder _ name : string
 : param token : ( optional ) A valid token for the user in question .
 : type token : None | string
 : returns : A list of all items with the given name and parent folder
 name .
 : rtype : list [ dict ]"""

implementation:
parameters = dict ( )
parameters [ 'name' ] = name
parameters [ 'folderName' ] = folder_name
if token :
    parameters [ 'token' ] = token
response = self . request ( 'midas.item.searchbynameandfoldername' , parameters )
return response [ 'items' ]


signature:
def create_link ( self , token , folder_id , url , ** kwargs ) :
    """Create a link bitstream .
 : param token : A valid token for the user in question .
 : type token : string
 : param folder _ id : The id of the folder in which to create a new item
 that will contain the link . The new item will have the same name as
 the URL unless an item name is supplied .
 : type folder _ id : int | long
 : param url : The URL of the link you will create , will be used as the
 name of the bitstream and of the item unless an item name is
 supplied .
 : type url : string
 : param item _ name : ( optional ) The name of the newly created item , if
 not supplied , the item will have the same name as the URL .
 : type item _ name : string
 : param length : ( optional ) The length in bytes of the file to which the
 link points .
 : type length : int | long
 : param checksum : ( optional ) The MD5 checksum of the file to which the
 link points .
 : type checksum : string
 : returns : The item information of the item created .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'folderid' ] = folder_id
parameters [ 'url' ] = url
optional_keys = [ 'item_name' , 'length' , 'checksum' ]
for key in optional_keys :
    if key in kwargs :
        if key == 'item_name' :
            parameters [ 'itemname' ] = kwargs [ key ]
            continue
        parameters [ key ] = kwargs [ key ]
response = self . request ( 'midas.link.create' , parameters )
return response


signature:
def generate_upload_token ( self , token , item_id , filename , checksum = None ) :
    """Generate a token to use for upload .
 Midas Server uses a individual token for each upload . The token
 corresponds to the file specified and that file only . Passing the MD5
 checksum allows the server to determine if the file is already in the
 asset store .
 If : param : ` checksum ` is passed and the token returned is blank , the
 server already has this file and there is no need to follow this
 call with a call to ` perform _ upload ` , as the passed in file will have
 been added as a bitstream to the item ' s latest revision , creating a
 new revision if one doesn ' t exist .
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : The id of the item in which to upload the file as a
 bitstream .
 : type item _ id : int | long
 : param filename : The name of the file to generate the upload token for .
 : type filename : string
 : param checksum : ( optional ) The checksum of the file to upload .
 : type checksum : None | string
 : returns : String of the upload token .
 : rtype : string"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'itemid' ] = item_id
parameters [ 'filename' ] = filename
if checksum is not None :
    parameters [ 'checksum' ] = checksum
response = self . request ( 'midas.upload.generatetoken' , parameters )
return response [ 'token' ]


signature:
def perform_upload ( self , upload_token , filename , ** kwargs ) :
    """Upload a file into a given item ( or just to the public folder if the
 item is not specified .
 : param upload _ token : The upload token ( returned by
 generate _ upload _ token )
 : type upload _ token : string
 : param filename : The upload filename . Also used as the path to the
 file , if ' filepath ' is not set .
 : type filename : string
 : param mode : ( optional ) Stream or multipart . Default is stream .
 : type mode : string
 : param folder _ id : ( optional ) The id of the folder to upload into .
 : type folder _ id : int | long
 : param item _ id : ( optional ) If set , will append item ` ` bitstreams ` ` to
 the latest revision ( or the one set using : param : ` revision ` ) of
 the existing item .
 : type item _ id : int | long
 : param revision : ( optional ) If set , will add a new file into an
 existing revision . Set this to ' head ' to add to the most recent
 revision .
 : type revision : string | int | long
 : param filepath : ( optional ) The path to the file .
 : type filepath : string
 : param create _ additional _ revision : ( optional ) If set , will create a
 new revision in the existing item .
 : type create _ additional _ revision : bool
 : returns : Dictionary containing the details of the item created or
 changed .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'uploadtoken' ] = upload_token
parameters [ 'filename' ] = filename
try :
    create_additional_revision = kwargs [ 'create_additional_revision' ]
except KeyError :
    create_additional_revision = False
if not create_additional_revision :
    parameters [ 'revision' ] = 'head'
optional_keys = [ 'mode' , 'folderid' , 'item_id' , 'itemid' , 'revision' ]
for key in optional_keys :
    if key in kwargs :
        if key == 'item_id' :
            parameters [ 'itemid' ] = kwargs [ key ]
            continue
        if key == 'folder_id' :
            parameters [ 'folderid' ] = kwargs [ key ]
            continue
        parameters [ key ] = kwargs [ key ]
# We may want a different name than path
file_payload = open ( kwargs . get ( 'filepath' , filename ) , 'rb' )
# Arcane getting of the file size using fstat . More details can be
# found in the python library docs
parameters [ 'length' ] = os . fstat ( file_payload . fileno ( ) ) . st_size
response = self . request ( 'midas.upload.perform' , parameters , file_payload )
return response


signature:
def search ( self , search , token = None ) :
    """Get the resources corresponding to a given query .
 : param search : The search criterion .
 : type search : string
 : param token : ( optional ) The credentials to use when searching .
 : type token : None | string
 : returns : Dictionary containing the search result . Notable is the
 dictionary item ' results ' , which is a list of item details .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'search' ] = search
if token :
    parameters [ 'token' ] = token
response = self . request ( 'midas.resource.search' , parameters )
return response


signature:
def add_condor_dag ( self , token , batchmaketaskid , dagfilename , dagmanoutfilename ) :
    """Add a Condor DAG to the given Batchmake task .
 : param token : A valid token for the user in question .
 : type token : string
 : param batchmaketaskid : id of the Batchmake task for this DAG
 : type batchmaketaskid : int | long
 : param dagfilename : Filename of the DAG file
 : type dagfilename : string
 : param dagmanoutfilename : Filename of the DAG processing output
 : type dagmanoutfilename : string
 : returns : The created Condor DAG DAO
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'batchmaketaskid' ] = batchmaketaskid
parameters [ 'dagfilename' ] = dagfilename
parameters [ 'outfilename' ] = dagmanoutfilename
response = self . request ( 'midas.batchmake.add.condor.dag' , parameters )
return response


signature:
def add_condor_job ( self , token , batchmaketaskid , jobdefinitionfilename , outputfilename , errorfilename , logfilename , postfilename ) :
    """Add a Condor DAG job to the Condor DAG associated with this
 Batchmake task
 : param token : A valid token for the user in question .
 : type token : string
 : param batchmaketaskid : id of the Batchmake task for this DAG
 : type batchmaketaskid : int | long
 : param jobdefinitionfilename : Filename of the definition file for the
 job
 : type jobdefinitionfilename : string
 : param outputfilename : Filename of the output file for the job
 : type outputfilename : string
 : param errorfilename : Filename of the error file for the job
 : type errorfilename : string
 : param logfilename : Filename of the log file for the job
 : type logfilename : string
 : param postfilename : Filename of the post script log file for the job
 : type postfilename : string
 : return : The created Condor job DAO .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'batchmaketaskid' ] = batchmaketaskid
parameters [ 'jobdefinitionfilename' ] = jobdefinitionfilename
parameters [ 'outputfilename' ] = outputfilename
parameters [ 'errorfilename' ] = errorfilename
parameters [ 'logfilename' ] = logfilename
parameters [ 'postfilename' ] = postfilename
response = self . request ( 'midas.batchmake.add.condor.job' , parameters )
return response


signature:
def extract_dicommetadata ( self , token , item_id ) :
    """Extract DICOM metadata from the given item
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : id of the item to be extracted
 : type item _ id : int | long
 : return : the item revision DAO
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'item' ] = item_id
response = self . request ( 'midas.dicomextractor.extract' , parameters )
return response


signature:
def mfa_otp_login ( self , temp_token , one_time_pass ) :
    """Log in to get the real token using the temporary token and otp .
 : param temp _ token : The temporary token or id returned from normal login
 : type temp _ token : string
 : param one _ time _ pass : The one - time pass to be sent to the underlying
 multi - factor engine .
 : type one _ time _ pass : string
 : returns : A standard token for interacting with the web api .
 : rtype : string"""

implementation:
parameters = dict ( )
parameters [ 'mfaTokenId' ] = temp_token
parameters [ 'otp' ] = one_time_pass
response = self . request ( 'midas.mfa.otp.login' , parameters )
return response [ 'token' ]


signature:
def create_big_thumbnail ( self , token , bitstream_id , item_id , width = 575 ) :
    """Create a big thumbnail for the given bitstream with the given width .
 It is used as the main image of the given item and shown in the item
 view page .
 : param token : A valid token for the user in question .
 : type token : string
 : param bitstream _ id : The bitstream from which to create the thumbnail .
 : type bitstream _ id : int | long
 : param item _ id : The item on which to set the thumbnail .
 : type item _ id : int | long
 : param width : ( optional ) The width in pixels to which to resize ( aspect
 ratio will be preserved ) . Defaults to 575.
 : type width : int | long
 : returns : The ItemthumbnailDao object that was created .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'bitstreamId' ] = bitstream_id
parameters [ 'itemId' ] = item_id
parameters [ 'width' ] = width
response = self . request ( 'midas.thumbnailcreator.create.big.thumbnail' , parameters )
return response


signature:
def create_small_thumbnail ( self , token , item_id ) :
    """Create a 100x100 small thumbnail for the given item . It is used for
 preview purpose and displayed in the ' preview ' and ' thumbnails '
 sidebar sections .
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : The item on which to set the thumbnail .
 : type item _ id : int | long
 : returns : The item object ( with the new thumbnail id ) and the path
 where the newly created thumbnail is stored .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'itemId' ] = item_id
response = self . request ( 'midas.thumbnailcreator.create.small.thumbnail' , parameters )
return response


signature:
def solr_advanced_search ( self , query , token = None , limit = 20 ) :
    """Search item metadata using Apache Solr .
 : param query : The Apache Lucene search query .
 : type query : string
 : param token : ( optional ) A valid token for the user in question .
 : type token : None | string
 : param limit : ( optional ) The limit of the search .
 : type limit : int | long
 : returns : The list of items that match the search query .
 : rtype : list [ dict ]"""

implementation:
parameters = dict ( )
parameters [ 'query' ] = query
parameters [ 'limit' ] = limit
if token :
    parameters [ 'token' ] = token
response = self . request ( 'midas.solr.search.advanced' , parameters )
return response


signature:
def associate_item_with_scalar_data ( self , token , item_id , scalar_id , label ) :
    """Associate a result item with a particular scalar value .
 : param token : A valid token for the user in question .
 : type token : string
 : param item _ id : The id of the item to associate with the scalar .
 : type item _ id : int | long
 : param scalar _ id : Scalar id with which to associate the item .
 : type scalar _ id : int | long
 : param label : The label describing the nature of the association .
 : type label : string"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'scalarIds' ] = scalar_id
parameters [ 'itemId' ] = item_id
parameters [ 'label' ] = label
self . request ( 'midas.tracker.item.associate' , parameters )


signature:
def create_submission ( self , token , ** kwargs ) :
    """Associate a result item with a particular scalar value .
 : param token : A valid token for the user in question .
 : type token : string
 : param uuid ( optional ) The uuid of the submission ( must be unique )
 : type uuid : string
 : param name ( optional ) The name of the submission
 : type name : string
 : returns : The submission object that was created .
 : rtype : dict"""

implementation:
parameters = { }
parameters [ 'token' ] = token
optional_keys = [ 'uuid' , 'name' ]
for key in optional_keys :
    if key in kwargs :
        parameters [ key ] = kwargs [ key ]
return self . request ( 'midas.tracker.submission.create' , parameters )


signature:
def add_scalar_data ( self , token , community_id , producer_display_name , metric_name , producer_revision , submit_time , value , ** kwargs ) :
    """Create a new scalar data point .
 : param token : A valid token for the user in question .
 : type token : string
 : param community _ id : The id of the community that owns the producer .
 : type community _ id : int | long
 : param producer _ display _ name : The display name of the producer .
 : type producer _ display _ name : string
 : param metric _ name : The metric name that identifies which trend this
 point belongs to .
 : type metric _ name : string
 : param producer _ revision : The repository revision of the producer that
 produced this value .
 : type producer _ revision : int | long | string
 : param submit _ time : The submit timestamp . Must be parsable with PHP
 strtotime ( ) .
 : type submit _ time : string
 : param value : The value of the scalar .
 : type value : float
 : param config _ item _ id : ( optional ) If this value pertains to a specific
 configuration item , pass its id here .
 : type config _ item _ id : int | long
 : param test _ dataset _ id : ( optional ) If this value pertains to a
 specific test dataset , pass its id here .
 : type test _ dataset _ id : int | long
 : param truth _ dataset _ id : ( optional ) If this value pertains to a
 specific ground truth dataset , pass its id here .
 : type truth _ dataset _ id : int | long
 : param silent : ( optional ) If true , do not perform threshold - based email
 notifications for this scalar .
 : type silent : bool
 : param unofficial : ( optional ) If true , creates an unofficial scalar
 visible only to the user performing the submission .
 : type unofficial : bool
 : param build _ results _ url : ( optional ) A URL for linking to build results
 for this submission .
 : type build _ results _ url : string
 : param branch : ( optional ) The branch name in the source repository for
 this submission .
 : type branch : string
 : param submission _ id : ( optional ) The id of the submission .
 : type submission _ id : int | long
 : param submission _ uuid : ( optional ) The uuid of the submission . If one
 does not exist , it will be created .
 : type submission _ uuid : string
 : type branch : string
 : param params : ( optional ) Any key / value pairs that should be displayed
 with this scalar result .
 : type params : dict
 : param extra _ urls : ( optional ) Other URL ' s that should be displayed with
 with this scalar result . Each element of the list should be a dict
 with the following keys : label , text , href
 : type extra _ urls : list [ dict ]
 : param unit : ( optional ) The unit of the scalar value .
 : type unit : string
 : param reproduction _ command : ( optional ) The command to reproduce this
 scalar .
 : type reproduction _ command : string
 : returns : The scalar object that was created .
 : rtype : dict"""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'communityId' ] = community_id
parameters [ 'producerDisplayName' ] = producer_display_name
parameters [ 'metricName' ] = metric_name
parameters [ 'producerRevision' ] = producer_revision
parameters [ 'submitTime' ] = submit_time
parameters [ 'value' ] = value
optional_keys = [ 'config_item_id' , 'test_dataset_id' , 'truth_dataset_id' , 'silent' , 'unofficial' , 'build_results_url' , 'branch' , 'extra_urls' , 'params' , 'submission_id' , 'submission_uuid' , 'unit' , 'reproduction_command' ]
for key in optional_keys :
    if key in kwargs :
        if key == 'config_item_id' :
            parameters [ 'configItemId' ] = kwargs [ key ]
        elif key == 'test_dataset_id' :
            parameters [ 'testDatasetId' ] = kwargs [ key ]
        elif key == 'truth_dataset_id' :
            parameters [ 'truthDatasetId' ] = kwargs [ key ]
        elif key == 'build_results_url' :
            parameters [ 'buildResultsUrl' ] = kwargs [ key ]
        elif key == 'extra_urls' :
            parameters [ 'extraUrls' ] = json . dumps ( kwargs [ key ] )
        elif key == 'params' :
            parameters [ key ] = json . dumps ( kwargs [ key ] )
        elif key == 'silent' :
            if kwargs [ key ] :
                parameters [ key ] = kwargs [ key ]
        elif key == 'unofficial' :
            if kwargs [ key ] :
                parameters [ key ] = kwargs [ key ]
        elif key == 'submission_id' :
            parameters [ 'submissionId' ] = kwargs [ key ]
        elif key == 'submission_uuid' :
            parameters [ 'submissionUuid' ] = kwargs [ key ]
        elif key == 'unit' :
            parameters [ 'unit' ] = kwargs [ key ]
        elif key == 'reproduction_command' :
            parameters [ 'reproductionCommand' ] = kwargs [ key ]
        else :
            parameters [ key ] = kwargs [ key ]
response = self . request ( 'midas.tracker.scalar.add' , parameters )
return response


signature:
def upload_json_results ( self , token , filepath , community_id , producer_display_name , metric_name , producer_revision , submit_time , ** kwargs ) :
    """Upload a JSON file containing numeric scoring results to be added as
 scalars . File is parsed and then deleted from the server .
 : param token : A valid token for the user in question .
 : param filepath : The path to the JSON file .
 : param community _ id : The id of the community that owns the producer .
 : param producer _ display _ name : The display name of the producer .
 : param producer _ revision : The repository revision of the producer
 that produced this value .
 : param submit _ time : The submit timestamp . Must be parsable with PHP
 strtotime ( ) .
 : param config _ item _ id : ( optional ) If this value pertains to a specific
 configuration item , pass its id here .
 : param test _ dataset _ id : ( optional ) If this value pertains to a
 specific test dataset , pass its id here .
 : param truth _ dataset _ id : ( optional ) If this value pertains to a
 specific ground truth dataset , pass its id here .
 : param parent _ keys : ( optional ) Semicolon - separated list of parent keys
 to look for numeric results under . Use ' . ' to denote nesting , like
 in normal javascript syntax .
 : param silent : ( optional ) If true , do not perform threshold - based email
 notifications for this scalar .
 : param unofficial : ( optional ) If true , creates an unofficial scalar
 visible only to the user performing the submission .
 : param build _ results _ url : ( optional ) A URL for linking to build results
 for this submission .
 : param branch : ( optional ) The branch name in the source repository for
 this submission .
 : param params : ( optional ) Any key / value pairs that should be displayed
 with this scalar result .
 : type params : dict
 : param extra _ urls : ( optional ) Other URL ' s that should be displayed with
 with this scalar result . Each element of the list should be a dict
 with the following keys : label , text , href
 : type extra _ urls : list of dicts
 : returns : The list of scalars that were created ."""

implementation:
parameters = dict ( )
parameters [ 'token' ] = token
parameters [ 'communityId' ] = community_id
parameters [ 'producerDisplayName' ] = producer_display_name
parameters [ 'metricName' ] = metric_name
parameters [ 'producerRevision' ] = producer_revision
parameters [ 'submitTime' ] = submit_time
optional_keys = [ 'config_item_id' , 'test_dataset_id' , 'truth_dataset_id' , 'silent' , 'unofficial' , 'build_results_url' , 'branch' , 'extra_urls' , 'params' ]
for key in optional_keys :
    if key in kwargs :
        if key == 'config_item_id' :
            parameters [ 'configItemId' ] = kwargs [ key ]
        elif key == 'test_dataset_id' :
            parameters [ 'testDatasetId' ] = kwargs [ key ]
        elif key == 'truth_dataset_id' :
            parameters [ 'truthDatasetId' ] = kwargs [ key ]
        elif key == 'parent_keys' :
            parameters [ 'parentKeys' ] = kwargs [ key ]
        elif key == 'build_results_url' :
            parameters [ 'buildResultsUrl' ] = kwargs [ key ]
        elif key == 'extra_urls' :
            parameters [ 'extraUrls' ] = json . dumps ( kwargs [ key ] )
        elif key == 'params' :
            parameters [ key ] = json . dumps ( kwargs [ key ] )
        elif key == 'silent' :
            if kwargs [ key ] :
                parameters [ key ] = kwargs [ key ]
        elif key == 'unofficial' :
            if kwargs [ key ] :
                parameters [ key ] = kwargs [ key ]
        else :
            parameters [ key ] = kwargs [ key ]
file_payload = open ( filepath , 'rb' )
response = self . request ( 'midas.tracker.results.upload.json' , parameters , file_payload )
return response


signature:
def __get_rev ( self , key , version , ** kwa ) :
    '''Obtain particular version of the doc at key .'''

implementation:
if '_doc' in kwa :
    doc = kwa [ '_doc' ]
else :
    if type ( version ) is int :
        if version == 0 :
            order = pymongo . ASCENDING
        elif version == - 1 :
            order = pymongo . DESCENDING
        doc = self . _collection . find_one ( { 'k' : key } , sort = [ [ 'd' , order ] ] )
    elif type ( version ) is datetime :
        ver = self . __round_time ( version )
        doc = self . _collection . find_one ( { 'k' : key , 'd' : ver } )
if doc is None :
    raise KeyError ( 'Supplied key `{0}` or version `{1}` does not exist' . format ( key , str ( version ) ) )
coded_val = doc [ 'v' ]
return pickle . loads ( coded_val )


signature:
def __round_time ( self , dt ) :
    """Round a datetime object to a multiple of a timedelta
 dt : datetime . datetime object , default now ."""

implementation:
round_to = self . _resolution . total_seconds ( )
seconds = ( dt - dt . min ) . seconds
rounding = ( seconds + round_to / 2 ) // round_to * round_to
return dt + timedelta ( 0 , rounding - seconds , - dt . microsecond )


signature:
def _hashkey ( self , method , url , ** kwa ) :
    '''Find a hash value for the linear combination of invocation methods .'''

implementation:
to_hash = '' . join ( [ str ( method ) , str ( url ) , str ( kwa . get ( 'data' , '' ) ) , str ( kwa . get ( 'params' , '' ) ) ] )
return hashlib . md5 ( to_hash . encode ( ) ) . hexdigest ( )


signature:
def setup ( self , address , rack = 0 , slot = 1 , port = 102 ) :
    """Connects to a Siemens S7 PLC .
 Connects to a Siemens S7 using the Snap7 library .
 See [ the snap7 documentation ] ( http : / / snap7 . sourceforge . net / ) for
 supported models and more details .
 It ' s not currently possible to query the device for available pins ,
 so ` available _ pins ( ) ` returns an empty list . Instead , you should use
 ` map _ pin ( ) ` to map to a Merker , Input or Output in the PLC . The
 internal id you should use is a string following this format :
 ' [ DMQI ] [ XBWD ] [ 0-9 ] + . ? [ 0-9 ] * ' where :
 * [ DMQI ] : D for DB , M for Merker , Q for Output , I for Input
 * [ XBWD ] : X for bit , B for byte , W for word , D for dword
 * [ 0-9 ] + : Address of the resource
 * [ 0-9 ] * : Bit of the address ( type X only , ignored in others )
 For example : ' IB100 ' will read a byte from an input at address 100 and
 ' MX50.2 ' will read / write bit 2 of the Merker at address 50 . It ' s not
 allowed to write to inputs ( I ) , but you can read / write Outpus , DBs and
 Merkers . If it ' s disallowed by the PLC , an exception will be thrown by
 python - snap7 library .
 For this library to work , it might be needed to change some settings
 in the PLC itself . See
 [ the snap7 documentation ] ( http : / / snap7 . sourceforge . net / ) for more
 information . You also need to put the PLC in RUN mode . Not however that
 having a Ladder program downloaded , running and modifying variables
 will probably interfere with inputs and outputs , so put it in RUN mode ,
 but preferably without a downloaded program .
 @ arg address IP address of the module .
 @ arg rack rack where the module is installed .
 @ arg slot slot in the rack where the module is installed .
 @ arg port port the PLC is listenning to .
 @ throw RuntimeError if something went wrong
 @ throw any exception thrown by ` snap7 ` ' s methods ."""

implementation:
rack = int ( rack )
slot = int ( slot )
port = int ( port )
address = str ( address )
self . _client = snap7 . client . Client ( )
self . _client . connect ( address , rack , slot , port )


signature:
def setup ( self , port ) :
    """Connects to an Arduino UNO on serial port ` port ` .
 @ throw RuntimeError can ' t connect to Arduino"""

implementation:
port = str ( port )
# timeout is used by all I / O operations
self . _serial = serial . Serial ( port , 115200 , timeout = 2 )
time . sleep ( 2 )
# time to Arduino reset
if not self . _serial . is_open :
    raise RuntimeError ( 'Could not connect to Arduino' )
self . _serial . write ( b'\x01' )
if self . _serial . read ( ) != b'\x06' :
    raise RuntimeError ( 'Could not connect to Arduino' )
ps = [ p for p in self . available_pins ( ) if p [ 'digital' ] [ 'output' ] ]
for pin in ps :
    self . _set_pin_direction ( pin [ 'id' ] , ahio . Direction . Output )


signature:
def block_resource_fitnesses ( self , block : block . Block ) :
    """Returns a map of nodename to average fitness value for this block .
 Assumes that required resources have been checked on all nodes ."""

implementation:
# Short - circuit ! My algorithm is terrible , so it doesn ' t work well for the edge case where
# the block has no requirements
if not block . resources :
    return { n : 1 for n in self . config . nodes . keys ( ) }
node_fitnesses = { }
for resource in block . resources :
    resource_fitnesses = self . resource_fitnesses ( resource )
    if not resource_fitnesses :
        raise UnassignableBlock ( block . name )
    max_fit = max ( resource_fitnesses . values ( ) )
    min_fit = min ( resource_fitnesses . values ( ) )
    for node , fitness in resource_fitnesses . items ( ) :
        if node not in node_fitnesses :
            node_fitnesses [ node ] = { }
        if not fitness : # Since we ' re rescaling , 0 is now an OK value . . .
        # We will check for ` is False ` after this
            node_fitnesses [ node ] [ resource . describe ( ) ] = False
        else :
            if max_fit - min_fit :
                node_fitnesses [ node ] [ resource . describe ( ) ] = ( fitness - min_fit ) / ( max_fit - min_fit )
            else : # All the values are the same , default to 1
                node_fitnesses [ node ] [ resource . describe ( ) ] = 1.0
res = { }
for node , res_fits in node_fitnesses . items ( ) :
    fit_sum = 0
    for res_desc , fit in res_fits . items ( ) :
        if fit is False :
            fit_sum = False
            break
        fit_sum += fit
    if fit_sum is False : # Skip this node entirely
        res [ node ] = False
        continue
    res [ node ] = fit_sum
return res


signature:
def available_drivers ( ) :
    """Returns a list of available drivers names ."""

implementation:
global __modules
global __available
if type ( __modules ) is not list :
    __modules = list ( __modules )
if not __available :
    __available = [ d . ahioDriverInfo . NAME for d in __modules if d . ahioDriverInfo . AVAILABLE ]
return __available


signature:
def __load_driver ( name ) :
    """Tries to load the driver named @ arg name .
 A driver is considered valid if it has a ahioDriverInfo object . It should
 however implement all APIs described in ` ahio . abstract _ driver ` , as they ' ll
 be needed to use the driver .
 @ returns the driver package , or False if it failed ."""

implementation:
global __count
try :
    dname = os . path . basename ( name ) . replace ( '.py' , '' )
    mod_name = 'ahio.drivers.%s%d' % ( dname , __count )
    loader = importlib . machinery . SourceFileLoader ( mod_name , name )
    driver = loader . load_module ( )
    __count += 1
    return driver if hasattr ( driver , 'ahioDriverInfo' ) else False
except Exception :
    return False


signature:
def __locate_driver_named ( name ) :
    """Searchs _ _ modules for a driver named @ arg name .
 @ returns the package for driver @ arg name or None if one can ' t be found ."""

implementation:
global __modules
if type ( __modules ) is not list :
    __modules = list ( __modules )
ms = [ d for d in __modules if d . ahioDriverInfo . NAME == name ]
if not ms :
    return None
return ms [ 0 ]


signature:
def map_pin ( self , abstract_pin_id , physical_pin_id ) :
    """Maps a pin number to a physical device pin .
 To make it easy to change drivers without having to refactor a lot of
 code , this library does not use the names set by the driver to identify
 a pin . This function will map a number , that will be used by other
 functions , to a physical pin represented by the drivers pin id . That
 way , if you need to use another pin or change the underlying driver
 completly , you only need to redo the mapping .
 If you ' re developing a driver , keep in mind that your driver will not
 know about this . The other functions will translate the mapped pin to
 your id before calling your function .
 @ arg abstract _ pin _ id the id that will identify this pin in the
 other function calls . You can choose what you want .
 @ arg physical _ pin _ id the id returned in the driver .
 See ` AbstractDriver . available _ pins ` . Setting it to None removes the
 mapping ."""

implementation:
if physical_pin_id :
    self . _pin_mapping [ abstract_pin_id ] = physical_pin_id
else :
    self . _pin_mapping . pop ( abstract_pin_id , None )


signature:
def set_pin_interpolation ( self , pin , read_min , read_max , write_min , write_max ) :
    """Interpolates input and output values for ` pin ` .
 Changes the output and input of ` AbstractDriver . read ` and
 ` AbstractDriver . write ` functions to use a value in range
 ( ` read _ min ` , ` read _ max ` ) or ( ` write _ min ` , ` write _ max ` ) instead of the
 values returned by ` available _ pins ` ( analog only ) . The conversion is
 done using linear interpolation . If ` read _ min ` , ` read _ max ` , ` write _ min `
 and ` write _ max ` are all None or don ' t form valid pairs ( like , read _ min
 has a value but read _ max is None ) , the pin is deregistered . If you pass
 a pair but leave the other with None values , only one direction is
 registered .
 @ arg pin pin id you ' ve set using ` AbstractDriver . map _ pin `
 @ arg read _ min the min value for the linear interpolation of
 ` AbstractDriver . read ` .
 @ arg read _ max the max value for the linear interpolation of
 ` AbstractDriver . read ` .
 @ arg write _ min the min value for the linear interpolation of
 ` AbstractDriver . write ` .
 @ arg write _ max the max value for the linear interpolation of
 ` AbstractDriver . write ` ."""

implementation:
if type ( pin ) is list : # I don ' t like breaking calls in multiple lines
    args = ( read_min , read_max , write_min , write_max )
    for p in pin :
        self . set_pin_interpolation ( p , * args )
    return
valid_read = ( read_min is not None and read_max is not None )
valid_write = ( write_min is not None and write_max is not None )
if not valid_read and not valid_write :
    self . _pin_lin . pop ( pin , None )
    return
pin_id = self . _pin_mapping . get ( pin , None )
pins = [ pin for pin in self . available_pins ( ) if pin_id == pin [ 'id' ] ]
read = pins [ 0 ] [ 'analog' ] [ 'read_range' ]
write = pins [ 0 ] [ 'analog' ] [ 'write_range' ]
valid_read = valid_read and read
valid_write = valid_write and write
self . _pin_lin [ pin ] = { 'read' : ( * read , read_min , read_max ) if valid_read else None , 'write' : ( write_min , write_max , * write ) if valid_write else None }


signature:
def set_pin_direction ( self , pin , direction ) :
    """Sets pin ` pin ` to ` direction ` .
 The pin should support the requested mode . Calling this function
 on a unmapped pin does nothing . Calling it with a unsupported direction
 throws RuntimeError .
 If you ' re developing a driver , you should implement
 _ set _ pin _ direction ( self , pin , direction ) where ` pin ` will be one of
 your internal IDs . If a pin is set to OUTPUT , put it on LOW state .
 @ arg pin pin id you ' ve set using ` AbstractDriver . map _ pin `
 @ arg mode a value from ` AbstractDriver . Direction `
 @ throw KeyError if pin isn ' t mapped .
 @ throw RuntimeError if direction is not supported by pin ."""

implementation:
if type ( pin ) is list :
    for p in pin :
        self . set_pin_direction ( p , direction )
    return
pin_id = self . _pin_mapping . get ( pin , None )
if pin_id and type ( direction ) is ahio . Direction :
    self . _set_pin_direction ( pin_id , direction )
else :
    raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def pin_direction ( self , pin ) :
    """Gets the ` ahio . Direction ` this pin was set to .
 If you ' re developing a driver , implement _ pin _ direction ( self , pin )
 @ arg pin the pin you want to see the mode
 @ returns the ` ahio . Direction ` the pin is set to
 @ throw KeyError if pin isn ' t mapped ."""

implementation:
if type ( pin ) is list :
    return [ self . pin_direction ( p ) for p in pin ]
pin_id = self . _pin_mapping . get ( pin , None )
if pin_id :
    return self . _pin_direction ( pin_id )
else :
    raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def set_pin_type ( self , pin , ptype ) :
    """Sets pin ` pin ` to ` type ` .
 The pin should support the requested mode . Calling this function
 on a unmapped pin does nothing . Calling it with a unsupported mode
 throws RuntimeError .
 If you ' re developing a driver , you should implement
 _ set _ pin _ type ( self , pin , ptype ) where ` pin ` will be one of your
 internal IDs . If a pin is set to OUTPUT , put it on LOW state .
 @ arg pin pin id you ' ve set using ` AbstractDriver . map _ pin `
 @ arg mode a value from ` AbstractDriver . PortType `
 @ throw KeyError if pin isn ' t mapped .
 @ throw RuntimeError if type is not supported by pin ."""

implementation:
if type ( pin ) is list :
    for p in pin :
        self . set_pin_type ( p , ptype )
    return
pin_id = self . _pin_mapping . get ( pin , None )
if type ( ptype ) is not ahio . PortType :
    raise KeyError ( 'ptype must be of type ahio.PortType' )
elif pin_id :
    self . _set_pin_type ( pin_id , ptype )
else :
    raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def pin_type ( self , pin ) :
    """Gets the ` ahio . PortType ` this pin was set to .
 If you ' re developing a driver , implement _ pin _ type ( self , pin )
 @ arg pin the pin you want to see the mode
 @ returns the ` ahio . PortType ` the pin is set to
 @ throw KeyError if pin isn ' t mapped ."""

implementation:
if type ( pin ) is list :
    return [ self . pin_type ( p ) for p in pin ]
pin_id = self . _pin_mapping . get ( pin , None )
if pin_id :
    return self . _pin_type ( pin_id )
else :
    raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def write ( self , pin , value , pwm = False ) :
    """Sets the output to the given value .
 Sets ` pin ` output to given value . If the pin is in INPUT mode , do
 nothing . If it ' s an analog pin , value should be in write _ range .
 If it ' s not in the allowed range , it will be clamped . If pin is in
 digital mode , value can be ` ahio . LogicValue ` if ` pwm ` = False , or a
 number between 0 and 1 if ` pwm ` = True . If PWM is False , the pin will
 be set to HIGH or LOW , if ` pwm ` is True , a PWM wave with the given
 cycle will be created . If the pin does not support PWM and ` pwm ` is
 True , raise RuntimeError . The ` pwm ` argument should be ignored in case
 the pin is analog . If value is not valid for the given
 pwm / analog | digital combination , raise TypeError .
 If you ' re developing a driver , implement _ write ( self , pin , value , pwm )
 @ arg pin the pin to write to
 @ arg value the value to write on the pin
 @ arg pwm wether the output should be a pwm wave
 @ throw RuntimeError if the pin does not support PWM and ` pwm ` is True .
 @ throw TypeError if value is not valid for this pin ' s mode and pwm
 value .
 @ throw KeyError if pin isn ' t mapped ."""

implementation:
if type ( pin ) is list :
    for p in pin :
        self . write ( p , value , pwm )
    return
if pwm and type ( value ) is not int and type ( value ) is not float :
    raise TypeError ( 'pwm is set, but value is not a float or int' )
pin_id = self . _pin_mapping . get ( pin , None )
if pin_id :
    lpin = self . _pin_lin . get ( pin , None )
    if lpin and type ( lpin [ 'write' ] ) is tuple :
        write_range = lpin [ 'write' ]
        value = self . _linear_interpolation ( value , * write_range )
    self . _write ( pin_id , value , pwm )
else :
    raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def read ( self , pin ) :
    """Reads value from pin ` pin ` .
 Returns the value read from pin ` pin ` . If it ' s an analog pin , returns
 a number in analog . input _ range . If it ' s digital , returns
 ` ahio . LogicValue ` .
 If you ' re developing a driver , implement _ read ( self , pin )
 @ arg pin the pin to read from
 @ returns the value read from the pin
 @ throw KeyError if pin isn ' t mapped ."""

implementation:
if type ( pin ) is list :
    return [ self . read ( p ) for p in pin ]
pin_id = self . _pin_mapping . get ( pin , None )
if pin_id :
    value = self . _read ( pin_id )
    lpin = self . _pin_lin . get ( pin , None )
    if lpin and type ( lpin [ 'read' ] ) is tuple :
        read_range = lpin [ 'read' ]
        value = self . _linear_interpolation ( value , * read_range )
    return value
else :
    raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def set_analog_reference ( self , reference , pin = None ) :
    """Sets the analog reference to ` reference `
 If the driver supports per pin reference setting , set pin to the
 desired reference . If not , passing None means set to all , which is the
 default in most hardware . If only per pin reference is supported and
 pin is None , raise RuntimeError .
 If you ' re developing a driver , implement
 _ set _ analog _ reference ( self , reference , pin ) . Raise RuntimeError if pin
 was set but is not supported by the platform .
 @ arg reference the value that describes the analog reference . See
 ` AbstractDriver . analog _ references `
 @ arg pin if the the driver supports it , the pin that will use
 ` reference ` as reference . None for all .
 @ throw RuntimeError if pin is None on a per pin only hardware , or if
 it ' s a valid pin on a global only analog reference hardware .
 @ throw KeyError if pin isn ' t mapped ."""

implementation:
if pin is None :
    self . _set_analog_reference ( reference , None )
else :
    pin_id = self . _pin_mapping . get ( pin , None )
    if pin_id :
        self . _set_analog_reference ( reference , pin_id )
    else :
        raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def analog_reference ( self , pin = None ) :
    """Returns the analog reference .
 If the driver supports per pin analog reference setting , returns the
 reference for pin ` pin ` . If pin is None , returns the global analog
 reference . If only per pin reference is supported and pin is None ,
 raise RuntimeError .
 If you ' re developing a driver , implement _ analog _ reference ( self , pin )
 @ arg pin if the the driver supports it , the pin that will use
 ` reference ` as reference . None for all .
 @ returns the reference used for pin
 @ throw RuntimeError if pin is None on a per pin only hardware , or if
 it ' s a valid pin on a global only analog reference hardware .
 @ throw KeyError if pin isn ' t mapped ."""

implementation:
if pin is None :
    return self . _analog_reference ( None )
else :
    pin_id = self . _pin_mapping . get ( pin , None )
    if pin_id :
        return self . _analog_reference ( pin_id )
    else :
        raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def set_pwm_frequency ( self , frequency , pin = None ) :
    """Sets PWM frequency , if supported by hardware
 If the driver supports per pin frequency setting , set pin to the
 desired frequency . If not , passing None means set to all . If only per
 pin frequency is supported and pin is None , raise RuntimeError .
 If you ' re developing a driver , implement
 _ set _ pwm _ frequency ( self , frequency , pin ) . Raise RuntimeError if pin
 was set but is not supported by the platform .
 @ arg frequency pwm frequency to be set , in Hz
 @ arg pin if the the driver supports it , the pin that will use
 ` frequency ` as pwm frequency . None for all / global .
 @ throw RuntimeError if pin is None on a per pin only hardware , or if
 it ' s a valid pin on a global only hardware .
 @ throw KeyError if pin isn ' t mapped ."""

implementation:
if pin is None :
    self . _set_pwm_frequency ( frequency , None )
else :
    pin_id = self . _pin_mapping . get ( pin , None )
    if pin_id :
        self . _set_pwm_frequency ( frequency , pin_id )
    else :
        raise KeyError ( 'Requested pin is not mapped: %s' % pin )


signature:
def notify ( self , message , title = "" , ** kwargs ) :
    """priority ( - 2 - 1 0 1 2)
 sound ( bike , bugle , cashregister , classical , cosmic , falling , gamelan ,
 incoming , intermission , magic , mechanical , pianobar , siren , spacealarm ,
 tugboat , alien , climb , persistent , echo , updown , none )"""

implementation:
logger . info ( "pushover notify title:{0} message:{1}" . format ( title , message ) )
try :
    data = { 'token' : self . _token , 'user' : self . _user , 'title' : title , 'message' : message , }
    data . update ( kwargs )
    payload = [ ]
    for ( k , v ) in data . items ( ) :
        if isinstance ( v , str_type ) :
            payload . append ( ( k , v . encode ( "utf-8" ) ) )
        else :
            payload . append ( ( k , v ) )
    headers = { "Content-type" : "application/x-www-form-urlencoded" , }
    conn = HTTPSConnection ( "api.pushover.net" )
    params = urlencode ( payload )
    conn . request ( "POST" , "/1/messages.json" , params , headers )
    rsp = conn . getresponse ( )
    if rsp . status != 200 :
        raise PushoverException ( "pushover:{0}" . format ( rsp . status ) )
    conn . close ( )
except Exception as e :
    raise PushoverException ( "exception:{0!r}" . format ( e ) )


signature:
def SIRode ( y0 , time , beta , gamma ) :
    """Integrate SIR epidemic model
 Simulate a very basic deterministic SIR system .
 : param 2x1 numpy array y0 : initial conditions
 : param Ntimestep length numpy array time : Vector of time points that solution is returned at
 : param float beta : transmission rate
 : param float gamma : recovery rate
 : returns : ( 2 ) x ( Ntimestep ) numpy array Xsim : first row S ( t ) , second row I ( t )"""

implementation:
Xsim = rk4 ( SIR_D , y0 , time , args = ( beta , gamma , ) )
Xsim = Xsim . transpose ( )
return Xsim


signature:
def url ( self ) :
    """Return the URL of the server .
 : returns : URL of the server
 : rtype : string"""

implementation:
if len ( self . drivers ) > 0 :
    return self . drivers [ 0 ] . url
else :
    return self . _url


signature:
def guess_array_memory_usage ( bam_readers , dtype , use_strand = False ) :
    """Returns an estimate for the maximum amount of memory to be consumed by numpy arrays ."""

implementation:
ARRAY_COUNT = 5
if not isinstance ( bam_readers , list ) :
    bam_readers = [ bam_readers ]
if isinstance ( dtype , basestring ) :
    dtype = NUMPY_DTYPES . get ( dtype , None )
use_strand = use_strand + 1
# if false , factor of 1 , if true , factor of 2
dtypes = guess_numpy_dtypes_from_idxstats ( bam_readers , default = None , force_dtype = False )
if not [ dt for dt in dtypes if dt is not None ] : # found no info from idx
    dtypes = guess_numpy_dtypes_from_idxstats ( bam_readers , default = dtype or numpy . uint64 , force_dtype = True )
elif dtype :
    dtypes = [ dtype if dt else None for dt in dtypes ]
read_groups = [ ]
no_read_group = False
for bam in bam_readers :
    rgs = bam . get_read_groups ( )
    if rgs :
        for rg in rgs :
            if rg not in read_groups :
                read_groups . append ( rg )
    else :
        no_read_group = True
read_groups = len ( read_groups ) + no_read_group
max_ref_size = 0
array_byte_overhead = sys . getsizeof ( numpy . zeros ( ( 0 ) , dtype = numpy . uint64 ) )
array_count = ARRAY_COUNT * use_strand * read_groups
for bam in bam_readers :
    for i , ( name , length ) in enumerate ( bam . get_references ( ) ) :
        if dtypes [ i ] is not None :
            max_ref_size = max ( max_ref_size , ( length + length * dtypes [ i ] ( ) . nbytes * array_count + ( array_byte_overhead * ( array_count + 1 ) ) ) )
return max_ref_size


signature:
def pi_version ( ) :
    """Detect the version of the Raspberry Pi . Returns either 1 , 2 or
 None depending on if it ' s a Raspberry Pi 1 ( model A , B , A + , B + ) ,
 Raspberry Pi 2 ( model B + ) , or not a Raspberry Pi .
 https : / / github . com / adafruit / Adafruit _ Python _ GPIO / blob / master / Adafruit _ GPIO / Platform . py"""

implementation:
if not os . path . isfile ( '/proc/cpuinfo' ) :
    return None
# Check / proc / cpuinfo for the Hardware field value .
# 2708 is pi 1
# 2709 is pi 2
# Anything else is not a pi .
with open ( '/proc/cpuinfo' , 'r' ) as infile :
    cpuinfo = infile . read ( )
# Match a line like ' Hardware : BCM2709'
match = re . search ( '^Hardware\s+:\s+(\w+)$' , cpuinfo , flags = re . MULTILINE | re . IGNORECASE )
if not match : # Couldn ' t find the hardware , assume it isn ' t a pi .
    return None
if match . group ( 1 ) == 'BCM2708' : # Pi 1
    return 1
elif match . group ( 1 ) == 'BCM2709' : # Pi 2
    return 2
elif match . group ( 1 ) == 'BCM2835' : # Pi 3
    return 3
else : # Something else , not a pi .
    return None


signature:
def system ( command , input = None ) :
    """commands . getoutput ( ) replacement that also works on windows .
 Code mostly copied from zc . buildout ."""

implementation:
logger . debug ( "Executing command: %s" , command )
p = subprocess . Popen ( command , shell = True , stdin = subprocess . PIPE , stdout = subprocess . PIPE , stderr = subprocess . PIPE , close_fds = MUST_CLOSE_FDS )
stdoutdata , stderrdata = p . communicate ( input = input )
result = stdoutdata + stderrdata
if p . returncode :
    logger . error ( "Something went wrong when executing '%s'" , command )
    logger . error ( "Returncode: %s" , p . returncode )
    logger . error ( "Output:" )
    logger . error ( result )
    sys . exit ( 1 )
logger . info ( result )


signature:
def main ( ) :
    """Create coverage reports and open them in the browser ."""

implementation:
usage = "Usage: %prog PATH_TO_PACKAGE"
parser = optparse . OptionParser ( usage = usage )
parser . add_option ( "-v" , "--verbose" , action = "store_true" , dest = "verbose" , default = False , help = "Show debug output" )
parser . add_option ( "-d" , "--output-dir" , action = "store" , type = "string" , dest = "output_dir" , default = '' , help = "" )
parser . add_option ( "-t" , "--test-args" , action = "store" , type = "string" , dest = "test_args" , default = '' , help = ( "Pass argument on to bin/test. Quote the argument, " + "for instance \"-t '-m somemodule'\"." ) )
( options , args ) = parser . parse_args ( )
if options . verbose :
    log_level = logging . DEBUG
else :
    log_level = logging . INFO
logging . basicConfig ( level = log_level , format = "%(levelname)s: %(message)s" )
curdir = os . getcwd ( )
testbinary = os . path . join ( curdir , 'bin' , 'test' )
if not os . path . exists ( testbinary ) :
    raise RuntimeError ( "Test command doesn't exist: %s" % testbinary )
coveragebinary = os . path . join ( curdir , 'bin' , 'coverage' )
if not os . path . exists ( coveragebinary ) :
    logger . debug ( "Trying globally installed coverage command." )
    coveragebinary = 'coverage'
logger . info ( "Running tests in coverage mode (can take a long time)" )
parts = [ coveragebinary , 'run' , testbinary ]
if options . test_args :
    parts . append ( options . test_args )
system ( " " . join ( parts ) )
logger . debug ( "Creating coverage reports..." )
if options . output_dir :
    coverage_dir = options . output_dir
    open_in_browser = False
else :
    coverage_dir = 'htmlcov'
    # The default
    open_in_browser = True
system ( "%s html --directory=%s" % ( coveragebinary , coverage_dir ) )
logger . info ( "Wrote coverage files to %s" , coverage_dir )
if open_in_browser :
    index_file = os . path . abspath ( os . path . join ( coverage_dir , 'index.html' ) )
    logger . debug ( "About to open %s in your webbrowser." , index_file )
    webbrowser . open ( 'file://' + index_file )
    logger . info ( "Opened reports in your browser." )


signature:
def setup ( self , configuration = "ModbusSerialClient(method='rtu',port='/dev/cu.usbmodem14101',baudrate=9600)" ) :
    """Start a Modbus server .
 The following classes are available with their respective named
 parameters :
 ModbusTcpClient
 host : The host to connect to ( default 127.0.0.1)
 port : The modbus port to connect to ( default 502)
 source _ address : The source address tuple to bind to ( default ( ' ' , 0 ) )
 timeout : The timeout to use for this socket ( default Defaults . Timeout )
 ModbusUdpClient
 host : The host to connect to ( default 127.0.0.1)
 port : The modbus port to connect to ( default 502)
 timeout : The timeout to use for this socket ( default None )
 ModbusSerialClient
 method : The method to use for connection ( asii , rtu , binary )
 port : The serial port to attach to
 stopbits : The number of stop bits to use ( default 1)
 bytesize : The bytesize of the serial messages ( default 8 bits )
 parity : Which kind of parity to use ( default None )
 baudrate : The baud rate to use for the serial device
 timeout : The timeout between serial requests ( default 3s )
 When configuring the ports , the following convention should be
 respected :
 portname : C1:13 - > Coil on device 1 , address 13
 The letters can be :
 C = Coil
 I = Input
 R = Register
 H = Holding
 @ arg configuration a string that instantiates one of those classes .
 @ throw RuntimeError can ' t connect to Arduino"""

implementation:
from pymodbus3 . client . sync import ModbusSerialClient , ModbusUdpClient , ModbusTcpClient
self . _client = eval ( configuration )
self . _client . connect ( )


signature:
def setup ( self , address , port ) :
    """Connects to server at ` address ` : ` port ` .
 Connects to a TCP server listening at ` address ` : ` port ` that implements
 the protocol described in the file " Generic TCP I : O Protocol . md "
 @ arg address IP or address to connect to .
 @ arg port port to connect to .
 @ throw RuntimeError if connection was successiful but protocol isn ' t
 supported .
 @ throw any exception thrown by ` socket . socket ` ' s methods ."""

implementation:
address = str ( address )
port = int ( port )
self . _socket = socket . socket ( )
self . _socket . connect ( ( address , port ) )
self . _socket . send ( b'HELLO 1.0\n' )
with self . _socket . makefile ( ) as f :
    if f . readline ( ) . strip ( ) != 'OK' :
        raise RuntimeError ( 'Protocol not supported' )


signature:
def get_exception_from_status_and_error_codes ( status_code , error_code , value ) :
    """Return an exception given status and error codes .
 : param status _ code : HTTP status code .
 : type status _ code : None | int
 : param error _ code : Midas Server error code .
 : type error _ code : None | int
 : param value : Message to display .
 : type value : string
 : returns : Exception .
 : rtype : pydas . exceptions . ResponseError"""

implementation:
if status_code == requests . codes . bad_request :
    exception = BadRequest ( value )
elif status_code == requests . codes . unauthorized :
    exception = Unauthorized ( value )
elif status_code == requests . codes . forbidden :
    exception = Unauthorized ( value )
elif status_code in [ requests . codes . not_found , requests . codes . gone ] :
    exception = NotFound ( value )
elif status_code == requests . codes . method_not_allowed :
    exception = MethodNotAllowed ( value )
elif status_code >= requests . codes . bad_request :
    exception = HTTPError ( value )
else :
    exception = ResponseError ( value )
if error_code == - 100 : # MIDAS _ INTERNAL _ ERROR
    exception = InternalError ( value )
elif error_code == - 101 : # MIDAS _ INVALID _ TOKEN
    exception = InvalidToken ( value )
elif error_code == - 105 : # MIDAS _ UPLOAD _ FAILED
    exception = UploadFailed ( value )
elif error_code == - 140 : # MIDAS _ UPLOAD _ TOKEN _ GENERATION _ FAILED
    exception = UploadTokenGenerationFailed ( value )
elif error_code == - 141 : # MIDAS _ INVALID _ UPLOAD _ TOKEN
    exception = InvalidUploadToken ( value )
elif error_code == - 150 : # MIDAS _ INVALID _ PARAMETER
    exception = InvalidParameter ( value )
elif error_code == - 151 : # MIDAS _ INVALID _ POLICY
    exception = InvalidPolicy ( value )
return exception


signature:
def analog_read ( self , pin ) :
    """Retrieve the last analog data value received for the specified pin .
 : param pin : Selected pin
 : return : The last value entered into the analog response table ."""

implementation:
with self . data_lock :
    data = self . _command_handler . analog_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_PIN_DATA_VALUE ]
return data


signature:
def analog_write ( self , pin , value ) :
    """Set the specified pin to the specified value .
 : param pin : Pin number
 : param value : Pin value
 : return : No return value"""

implementation:
if self . _command_handler . ANALOG_MESSAGE + pin < 0xf0 :
    command = [ self . _command_handler . ANALOG_MESSAGE + pin , value & 0x7f , ( value >> 7 ) & 0x7f ]
    self . _command_handler . send_command ( command )
else :
    self . extended_analog ( pin , value )


signature:
def close ( self ) :
    """This method will close the transport ( serial port ) and exit
 : return : No return value , but sys . exit ( 0 ) is called ."""

implementation:
self . _command_handler . system_reset ( )
self . _command_handler . stop ( )
self . transport . stop ( )
self . transport . close ( )
if self . verbose :
    print ( "PyMata close(): Calling sys.exit(0): Hope to see you soon!" )
sys . exit ( 0 )


signature:
def digital_read ( self , pin ) :
    """Retrieve the last digital data value received for the specified pin .
 NOTE : This command will return values for digital , pwm , etc , pin types
 : param pin : Selected pin
 : return : The last value entered into the digital response table ."""

implementation:
with self . data_lock :
    data = self . _command_handler . digital_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_PIN_DATA_VALUE ]
return data


signature:
def digital_write ( self , pin , value ) :
    """Set the specified pin to the specified value .
 : param pin : pin number
 : param value : pin value
 : return : No return value"""

implementation:
# The command value is not a fixed value , but needs to be calculated using the
# pin ' s port number
port = pin // 8
calculated_command = self . _command_handler . DIGITAL_MESSAGE + port
mask = 1 << ( pin % 8 )
# Calculate the value for the pin ' s position in the port mask
if value == 1 :
    self . digital_output_port_pins [ port ] |= mask
else :
    self . digital_output_port_pins [ port ] &= ~ mask
# Assemble the command
command = ( calculated_command , self . digital_output_port_pins [ port ] & 0x7f , ( self . digital_output_port_pins [ port ] >> 7 ) & 0x7f )
self . _command_handler . send_command ( command )


signature:
def disable_analog_reporting ( self , pin ) :
    """Disables analog reporting for a single analog pin .
 : param pin : Analog pin number . For example for A0 , the number is 0.
 : return : No return value"""

implementation:
command = [ self . _command_handler . REPORT_ANALOG + pin , self . REPORTING_DISABLE ]
self . _command_handler . send_command ( command )


signature:
def disable_digital_reporting ( self , pin ) :
    """Disables digital reporting . By turning reporting off for this pin , reporting
 is disabled for all 8 bits in the " port " -
 : param pin : Pin and all pins for this port
 : return : No return value"""

implementation:
port = pin // 8
command = [ self . _command_handler . REPORT_DIGITAL + port , self . REPORTING_DISABLE ]
self . _command_handler . send_command ( command )


signature:
def enable_analog_reporting ( self , pin ) :
    """Enables analog reporting . By turning reporting on for a single pin .
 : param pin : Analog pin number . For example for A0 , the number is 0.
 : return : No return value"""

implementation:
command = [ self . _command_handler . REPORT_ANALOG + pin , self . REPORTING_ENABLE ]
self . _command_handler . send_command ( command )


signature:
def enable_digital_reporting ( self , pin ) :
    """Enables digital reporting . By turning reporting on for all 8 bits in the " port " -
 this is part of Firmata ' s protocol specification .
 : param pin : Pin and all pins for this port
 : return : No return value"""

implementation:
port = pin // 8
command = [ self . _command_handler . REPORT_DIGITAL + port , self . REPORTING_ENABLE ]
self . _command_handler . send_command ( command )


signature:
def encoder_config ( self , pin_a , pin_b , cb = None ) :
    """This command enables the rotary encoder ( 2 pin + ground ) and will
 enable encoder reporting .
 NOTE : This command is not currently part of standard arduino firmata , but is provided for legacy
 support of CodeShield on an Arduino UNO .
 Encoder data is retrieved by performing a digital _ read from pin a ( encoder pin 1)
 : param pin _ a : Encoder pin 1.
 : param pin _ b : Encoder pin 2.
 : param cb : callback function to report encoder changes
 : return : No return value"""

implementation:
data = [ pin_a , pin_b ]
self . _command_handler . digital_response_table [ pin_a ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . ENCODER
self . _command_handler . digital_response_table [ pin_a ] [ self . _command_handler . RESPONSE_TABLE_CALLBACK ] = cb
self . enable_digital_reporting ( pin_a )
self . _command_handler . digital_response_table [ pin_b ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . ENCODER
self . _command_handler . digital_response_table [ pin_b ] [ self . _command_handler . RESPONSE_TABLE_CALLBACK ] = cb
self . enable_digital_reporting ( pin_b )
self . _command_handler . send_sysex ( self . _command_handler . ENCODER_CONFIG , data )


signature:
def extended_analog ( self , pin , data ) :
    """This method will send an extended data analog output command to the selected pin
 : param pin : 0 - 127
 : param data : 0 - 0xfffff"""

implementation:
analog_data = [ pin , data & 0x7f , ( data >> 7 ) & 0x7f , ( data >> 14 ) & 0x7f ]
self . _command_handler . send_sysex ( self . _command_handler . EXTENDED_ANALOG , analog_data )


signature:
def get_pin_state_query_results ( self ) :
    """This method returns the results of a previous call to pin _ state _ query ( ) and then resets
 the pin state query data to None
 : return : Raw pin state query data"""

implementation:
r_data = self . _command_handler . last_pin_query_results
self . _command_handler . last_pin_query_results = [ ]
return r_data


signature:
def get_stepper_version ( self , timeout = 20 ) :
    """Get the stepper library version number .
 : param timeout : specify a time to allow arduino to process and return a version
 : return : the stepper version number if it was set ."""

implementation:
# get current time
start_time = time . time ( )
# wait for up to 20 seconds for a successful capability query to occur
while self . _command_handler . stepper_library_version <= 0 :
    if time . time ( ) - start_time > timeout :
        if self . verbose is True :
            print ( "Stepper Library Version Request timed-out. " "Did you send a stepper_request_library_version command?" )
        return
    else :
        pass
return self . _command_handler . stepper_library_version


signature:
def i2c_config ( self , read_delay_time = 0 , pin_type = None , clk_pin = 0 , data_pin = 0 ) :
    """NOTE : THIS METHOD MUST BE CALLED BEFORE ANY I2C REQUEST IS MADE
 This method initializes Firmata for I2c operations .
 It allows setting of a read time delay amount , and to optionally track
 the pins as I2C in the appropriate response table .
 To track pins : Set the pin _ type to ANALOG or DIGITAL and provide the pin numbers .
 If using ANALOG , pin numbers use the analog number , for example A4 : use 4.
 : param read _ delay _ time : an optional parameter , default is 0
 : param pin _ type : ANALOG or DIGITAL to select response table type to track pin numbers
 : param clk _ pin : pin number ( see comment above ) .
 : param data _ pin : pin number ( see comment above ) .
 : return : No Return Value"""

implementation:
data = [ read_delay_time & 0x7f , ( read_delay_time >> 7 ) & 0x7f ]
self . _command_handler . send_sysex ( self . _command_handler . I2C_CONFIG , data )
# If pin type is set , set pin mode in appropriate response table for these pins
if pin_type :
    if pin_type == self . DIGITAL :
        self . _command_handler . digital_response_table [ clk_pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . I2C
        self . _command_handler . digital_response_table [ data_pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . I2C
    else :
        self . _command_handler . analog_response_table [ clk_pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . I2C
        self . _command_handler . analog_response_table [ data_pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . I2C


signature:
def i2c_read ( self , address , register , number_of_bytes , read_type , cb = None ) :
    """This method requests the read of an i2c device . Results are retrieved by a call to
 i2c _ get _ read _ data ( ) .
 If a callback method is provided , when data is received from the device it will be sent to the callback method
 : param address : i2c device address
 : param register : register number ( can be set to zero )
 : param number _ of _ bytes : number of bytes expected to be returned
 : param read _ type : I2C _ READ or I2C _ READ _ CONTINUOUSLY
 : param cb : Optional callback function to report i2c data as result of read command"""

implementation:
data = [ address , read_type , register & 0x7f , ( register >> 7 ) & 0x7f , number_of_bytes & 0x7f , ( number_of_bytes >> 7 ) & 0x7f ]
# add or update entry in i2c _ map for reply
self . _command_handler . i2c_map [ address ] = [ cb , None ]
self . _command_handler . send_sysex ( self . _command_handler . I2C_REQUEST , data )


signature:
def i2c_write ( self , address , * args ) :
    """Write data to an i2c device .
 : param address : i2c device address
 : param args : A variable number of bytes to be sent to the device"""

implementation:
data = [ address , self . I2C_WRITE ]
for item in args :
    data . append ( item & 0x7f )
    data . append ( ( item >> 7 ) & 0x7f )
self . _command_handler . send_sysex ( self . _command_handler . I2C_REQUEST , data )


signature:
def i2c_stop_reading ( self , address ) :
    """This method stops an I2C _ READ _ CONTINUOUSLY operation for the i2c device address specified .
 : param address : address of i2c device"""

implementation:
data = [ address , self . I2C_STOP_READING ]
self . _command_handler . send_sysex ( self . _command_handler . I2C_REQUEST , data )


signature:
def i2c_get_read_data ( self , address ) :
    """This method retrieves the i2c read data as the result of an i2c _ read ( ) command .
 : param address : i2c device address
 : return : raw data read from device"""

implementation:
if address in self . _command_handler . i2c_map :
    map_entry = self . _command_handler . i2c_map [ address ]
    return map_entry [ 1 ]


signature:
def pin_state_query ( self , pin ) :
    """This method issues a pin state query command . Data returned is retrieved via
 a call to get _ pin _ state _ query _ results ( )
 : param pin : pin number"""

implementation: self . _command_handler . send_sysex ( self . _command_handler . PIN_STATE_QUERY , [ pin ] )


signature:
def play_tone ( self , pin , tone_command , frequency , duration ) :
    """This method will call the Tone library for the selected pin .
 If the tone command is set to TONE _ TONE , then the specified tone will be played .
 Else , if the tone command is TONE _ NO _ TONE , then any currently playing tone will be disabled .
 It is intended for a future release of Arduino Firmata
 : param pin : Pin number
 : param tone _ command : Either TONE _ TONE , or TONE _ NO _ TONE
 : param frequency : Frequency of tone in hz
 : param duration : Duration of tone in milliseconds
 : return : No return value"""

implementation:
# convert the integer values to bytes
if tone_command == self . TONE_TONE : # duration is specified
    if duration :
        data = [ tone_command , pin , frequency & 0x7f , ( frequency >> 7 ) & 0x7f , duration & 0x7f , ( duration >> 7 ) & 0x7f ]
    else :
        data = [ tone_command , pin , frequency & 0x7f , ( frequency >> 7 ) & 0x7f , 0 , 0 ]
    self . _command_handler . digital_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . TONE
# turn off tone
else :
    data = [ tone_command , pin ]
self . _command_handler . send_sysex ( self . _command_handler . TONE_PLAY , data )


signature:
def refresh_report_version ( self ) :
    """This method will query firmata for the report version .
 Retrieve the report version via a call to get _ firmata _ version ( )"""

implementation:
command = [ self . _command_handler . REPORT_VERSION ]
self . _command_handler . send_command ( command )


signature:
def reset ( self ) :
    """This command sends a reset message to the Arduino . The response tables will be reinitialized
 : return : No return value ."""

implementation:
# set all output pins to a value of 0
for pin in range ( 0 , self . _command_handler . total_pins_discovered ) :
    if self . _command_handler . digital_response_table [ self . _command_handler . RESPONSE_TABLE_MODE ] == self . PWM :
        self . analog_write ( pin , 0 )
    elif self . _command_handler . digital_response_table [ self . _command_handler . RESPONSE_TABLE_MODE ] == self . SERVO :
        self . analog_write ( pin , 0 )
    elif self . _command_handler . digital_response_table [ self . _command_handler . RESPONSE_TABLE_MODE ] == self . TONE :
        data = [ self . TONE_NO_TONE , pin ]
        self . _command_handler . send_sysex ( self . _command_handler . TONE_PLAY , data )
    else :
        self . digital_write ( pin , 0 )
self . _command_handler . system_reset ( )


signature:
def set_analog_latch ( self , pin , threshold_type , threshold_value , cb = None ) :
    """This method " arms " an analog pin for its data to be latched and saved in the latching table
 If a callback method is provided , when latching criteria is achieved , the callback function is called
 with latching data notification . In that case , the latching table is not updated .
 : param pin : Analog pin number ( value following an ' A ' designator , i . e . A5 = 5
 : param threshold _ type : ANALOG _ LATCH _ GT | ANALOG _ LATCH _ LT | ANALOG _ LATCH _ GTE | ANALOG _ LATCH _ LTE
 : param threshold _ value : numerical value - between 0 and 1023
 : param cb : callback method
 : return : True if successful , False if parameter data is invalid"""

implementation:
if self . ANALOG_LATCH_GT <= threshold_type <= self . ANALOG_LATCH_LTE :
    if 0 <= threshold_value <= 1023 :
        self . _command_handler . set_analog_latch ( pin , threshold_type , threshold_value , cb )
        return True
else :
    return False


signature:
def set_digital_latch ( self , pin , threshold_type , cb = None ) :
    """This method " arms " a digital pin for its data to be latched and saved in the latching table
 If a callback method is provided , when latching criteria is achieved , the callback function is called
 with latching data notification . In that case , the latching table is not updated .
 : param pin : Digital pin number
 : param threshold _ type : DIGITAL _ LATCH _ HIGH | DIGITAL _ LATCH _ LOW
 : param cb : callback function
 : return : True if successful , False if parameter data is invalid"""

implementation:
if 0 <= threshold_type <= 1 :
    self . _command_handler . set_digital_latch ( pin , threshold_type , cb )
    return True
else :
    return False


signature:
def set_pin_mode ( self , pin , mode , pin_type , cb = None ) :
    """This method sets a pin to the desired pin mode for the pin _ type .
 It automatically enables data reporting .
 NOTE : DO NOT CALL THIS METHOD FOR I2C . See i2c _ config ( ) .
 : param pin : Pin number ( for analog use the analog number , for example A4 : use 4)
 : param mode : INPUT , OUTPUT , PWM , PULLUP
 : param pin _ type : ANALOG or DIGITAL
 : param cb : This is an optional callback function to report data changes to the user
 : return : No return value"""

implementation:
command = [ self . _command_handler . SET_PIN_MODE , pin , mode ]
self . _command_handler . send_command ( command )
# enable reporting for input pins
if mode == self . INPUT or mode == self . PULLUP :
    if pin_type == self . ANALOG : # set analog response table to show this pin is an input pin
        self . _command_handler . analog_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . INPUT
        self . _command_handler . analog_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_CALLBACK ] = cb
        self . enable_analog_reporting ( pin )
    # if not analog it has to be digital
    else :
        self . _command_handler . digital_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = self . INPUT
        self . _command_handler . digital_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_CALLBACK ] = cb
        self . enable_digital_reporting ( pin )
else : # must be output - so set the tables accordingly
    if pin_type == self . ANALOG :
        self . _command_handler . analog_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = mode
    else :
        self . _command_handler . digital_response_table [ pin ] [ self . _command_handler . RESPONSE_TABLE_MODE ] = mode


signature:
def set_sampling_interval ( self , interval ) :
    """This method sends the desired sampling interval to Firmata .
 Note : Standard Firmata will ignore any interval less than 10 milliseconds
 : param interval : Integer value for desired sampling interval in milliseconds
 : return : No return value ."""

implementation:
data = [ interval & 0x7f , ( interval >> 7 ) & 0x7f ]
self . _command_handler . send_sysex ( self . _command_handler . SAMPLING_INTERVAL , data )


signature:
def servo_config ( self , pin , min_pulse = 544 , max_pulse = 2400 ) :
    """Configure a pin as a servo pin . Set pulse min , max in ms .
 : param pin : Servo Pin .
 : param min _ pulse : Min pulse width in ms .
 : param max _ pulse : Max pulse width in ms .
 : return : No return value"""

implementation:
self . set_pin_mode ( pin , self . SERVO , self . OUTPUT )
command = [ pin , min_pulse & 0x7f , ( min_pulse >> 7 ) & 0x7f , max_pulse & 0x7f , ( max_pulse >> 7 ) & 0x7f ]
self . _command_handler . send_sysex ( self . _command_handler . SERVO_CONFIG , command )


signature:
def sonar_config ( self , trigger_pin , echo_pin , cb = None , ping_interval = 50 , max_distance = 200 ) :
    """Configure the pins , ping interval and maximum distance for an HC - SR04 type device .
 Single pin configuration may be used . To do so , set both the trigger and echo pins to the same value .
 Up to a maximum of 6 SONAR devices is supported
 If the maximum is exceeded a message is sent to the console and the request is ignored .
 NOTE : data is measured in centimeters
 : param trigger _ pin : The pin number of for the trigger ( transmitter ) .
 : param echo _ pin : The pin number for the received echo .
 : param ping _ interval : Minimum interval between pings . Lowest number to use is 33 ms . Max is 127
 : param max _ distance : Maximum distance in cm . Max is 200.
 : param cb : optional callback function to report sonar data changes"""

implementation:
if max_distance > 200 :
    max_distance = 200
max_distance_lsb = max_distance & 0x7f
max_distance_msb = ( max_distance >> 7 ) & 0x7f
data = [ trigger_pin , echo_pin , ping_interval , max_distance_lsb , max_distance_msb ]
self . set_pin_mode ( trigger_pin , self . SONAR , self . INPUT )
self . set_pin_mode ( echo_pin , self . SONAR , self . INPUT )
# update the ping data map for this pin
if len ( self . _command_handler . active_sonar_map ) > 6 :
    if self . verbose :
        print ( "sonar_config: maximum number of devices assigned - ignoring request" )
    return
else :
    with self . data_lock : # self . _ command _ handler . active _ sonar _ map [ trigger _ pin ] = self . IGNORE
        self . _command_handler . active_sonar_map [ trigger_pin ] = [ cb , [ self . IGNORE ] ]
self . _command_handler . send_sysex ( self . _command_handler . SONAR_CONFIG , data )


signature:
def stepper_config ( self , steps_per_revolution , stepper_pins ) :
    """Configure stepper motor prior to operation .
 : param steps _ per _ revolution : number of steps per motor revolution
 : param stepper _ pins : a list of control pin numbers - either 4 or 2"""

implementation:
data = [ self . STEPPER_CONFIGURE , steps_per_revolution & 0x7f , ( steps_per_revolution >> 7 ) & 0x7f ]
for pin in range ( len ( stepper_pins ) ) :
    data . append ( stepper_pins [ pin ] )
self . _command_handler . send_sysex ( self . _command_handler . STEPPER_DATA , data )


signature:
def stepper_step ( self , motor_speed , number_of_steps ) :
    """Move a stepper motor for the number of steps at the specified speed
 : param motor _ speed : 21 bits of data to set motor speed
 : param number _ of _ steps : 14 bits for number of steps & direction
 positive is forward , negative is reverse"""

implementation:
if number_of_steps > 0 :
    direction = 1
else :
    direction = 0
abs_number_of_steps = abs ( number_of_steps )
data = [ self . STEPPER_STEP , motor_speed & 0x7f , ( motor_speed >> 7 ) & 0x7f , ( motor_speed >> 14 ) & 0x7f , abs_number_of_steps & 0x7f , ( abs_number_of_steps >> 7 ) & 0x7f , direction ]
self . _command_handler . send_sysex ( self . _command_handler . STEPPER_DATA , data )


signature:
def stepper_request_library_version ( self ) :
    """Request the stepper library version from the Arduino .
 To retrieve the version after this command is called , call
 get _ stepper _ version"""

implementation:
data = [ self . STEPPER_LIBRARY_VERSION ]
self . _command_handler . send_sysex ( self . _command_handler . STEPPER_DATA , data )


signature:
def open ( self , verbose ) :
    """open the serial port using the configuration data
 returns a reference to this instance"""

implementation:
# open a serial port
if verbose :
    print ( '\nOpening Arduino Serial port %s ' % self . port_id )
try : # in case the port is already open , let ' s close it and then
# reopen it
    self . arduino . close ( )
    time . sleep ( 1 )
    self . arduino . open ( )
    time . sleep ( 1 )
    return self . arduino
except Exception : # opened failed - will report back to caller
    raise


signature:
def write ( self , data ) :
    """write the data to the serial port
 return : None"""

implementation:
if sys . version_info [ 0 ] < 3 :
    self . arduino . write ( data )
else :
    self . arduino . write ( bytes ( [ ord ( data ) ] ) )


signature:
def run ( self ) :
    """This method continually runs . If an incoming character is available on the serial port
 it is read and placed on the _ command _ deque
 @ return : Never Returns"""

implementation:
while not self . is_stopped ( ) : # we can get an OSError : [ Errno9 ] Bad file descriptor when shutting down
# just ignore it
    try :
        if self . arduino . inWaiting ( ) :
            c = self . arduino . read ( )
            self . command_deque . append ( ord ( c ) )
        else :
            time . sleep ( .1 )
    except OSError :
        pass
    except IOError :
        self . stop ( )
self . close ( )


signature:
def set_blink_rate ( self , b ) :
    """Set the user ' s desired blink rate ( 0 - 3)
 @ param b : blink rate"""

implementation:
if b > 3 :
    b = 0
    # turn off if not sure
self . firmata . i2c_write ( self . board_address , ( self . HT16K33_BLINK_CMD | self . HT16K33_BLINK_DISPLAYON | ( b << 1 ) ) )


signature:
def set_brightness ( self , brightness ) :
    """Set the brightness level for the entire display
 @ param brightness : brightness level ( 0 - 15)"""

implementation:
if brightness > 15 :
    brightness = 15
brightness |= 0xE0
self . brightness = brightness
self . firmata . i2c_write ( 0x70 , brightness )


signature:
def set_bit_map ( self , shape , color ) :
    """Populate the bit map with the supplied " shape " and color
 and then write the entire bitmap to the display
 @ param shape : pattern to display
 @ param color : color for the pattern"""

implementation:
for row in range ( 0 , 8 ) :
    data = shape [ row ]
    # shift data into buffer
    bit_mask = 0x80
    for column in range ( 0 , 8 ) :
        if data & bit_mask :
            self . set_pixel ( row , column , color , True )
        bit_mask >>= 1
self . output_entire_buffer ( )


signature:
def output_entire_buffer ( self ) :
    """Write the entire buffer to the display"""

implementation:
green = 0
red = 0
for row in range ( 0 , 8 ) :
    for col in range ( 0 , 8 ) :
        if self . display_buffer [ row ] [ col ] == self . LED_GREEN :
            green |= 1 << col
        elif self . display_buffer [ row ] [ col ] == self . LED_RED :
            red |= 1 << col
        elif self . display_buffer [ row ] [ col ] == self . LED_YELLOW :
            green |= 1 << col
            red |= 1 << col
        elif self . display_buffer [ row ] [ col ] == self . LED_OFF :
            green &= ~ ( 1 << col )
            red &= ~ ( 1 << col )
    self . firmata . i2c_write ( 0x70 , row * 2 , 0 , green )
    self . firmata . i2c_write ( 0x70 , row * 2 + 1 , 0 , red )


signature:
def clear_display_buffer ( self ) :
    """Set all led ' s to off ."""

implementation:
for row in range ( 0 , 8 ) :
    self . firmata . i2c_write ( 0x70 , row * 2 , 0 , 0 )
    self . firmata . i2c_write ( 0x70 , ( row * 2 ) + 1 , 0 , 0 )
    for column in range ( 0 , 8 ) :
        self . display_buffer [ row ] [ column ] = 0


signature:
def auto_discover_board ( self , verbose ) :
    """This method will allow up to 30 seconds for discovery ( communicating with ) an Arduino board
 and then will determine a pin configuration table for the board .
 : return : True if board is successfully discovered or False upon timeout"""

implementation:
# get current time
start_time = time . time ( )
# wait for up to 30 seconds for a successful capability query to occur
while len ( self . analog_mapping_query_results ) == 0 :
    if time . time ( ) - start_time > 30 :
        return False
        # keep sending out a capability query until there is a response
    self . send_sysex ( self . ANALOG_MAPPING_QUERY )
    time . sleep ( .1 )
if verbose :
    print ( "Board initialized in %d seconds" % ( time . time ( ) - start_time ) )
for pin in self . analog_mapping_query_results :
    self . total_pins_discovered += 1
    # non analog pins will be marked as IGNORE
    if pin != self . pymata . IGNORE :
        self . number_of_analog_pins_discovered += 1
if verbose :
    print ( 'Total Number of Pins Detected = %d' % self . total_pins_discovered )
    print ( 'Total Number of Analog Pins Detected = %d' % self . number_of_analog_pins_discovered )
# response table initialization
# for each pin set the mode to input and the last read data value to zero
for pin in range ( 0 , self . total_pins_discovered ) :
    response_entry = [ self . pymata . INPUT , 0 , None ]
    self . digital_response_table . append ( response_entry )
for pin in range ( 0 , self . number_of_analog_pins_discovered ) :
    response_entry = [ self . pymata . INPUT , 0 , None ]
    self . analog_response_table . append ( response_entry )
# set up latching tables
for pin in range ( 0 , self . total_pins_discovered ) :
    digital_latch_table_entry = [ 0 , 0 , 0 , 0 , None ]
    self . digital_latch_table . append ( digital_latch_table_entry )
for pin in range ( 0 , self . number_of_analog_pins_discovered ) :
    analog_latch_table_entry = [ 0 , 0 , 0 , 0 , 0 , None ]
    self . analog_latch_table . append ( analog_latch_table_entry )
return True


signature:
def report_version ( self , data ) :
    """This method processes the report version message , sent asynchronously by Firmata when it starts up
 or after refresh _ report _ version ( ) is called
 Use the api method api _ get _ version to retrieve this information
 : param data : Message data from Firmata
 : return : No return value ."""

implementation:
self . firmata_version . append ( data [ 0 ] )
# add major
self . firmata_version . append ( data [ 1 ] )


signature:
def set_analog_latch ( self , pin , threshold_type , threshold_value , cb ) :
    """This method " arms " a pin to allow data latching for the pin .
 : param pin : Analog pin number ( value following an ' A ' designator , i . e . A5 = 5
 : param threshold _ type : ANALOG _ LATCH _ GT | ANALOG _ LATCH _ LT | ANALOG _ LATCH _ GTE | ANALOG _ LATCH _ LTE
 : param threshold _ value : numerical value
 : param cb : User provided callback function"""

implementation:
with self . pymata . data_lock :
    self . analog_latch_table [ pin ] = [ self . LATCH_ARMED , threshold_type , threshold_value , 0 , 0 , cb ]


signature:
def set_digital_latch ( self , pin , threshold_type , cb ) :
    """This method " arms " a pin to allow data latching for the pin .
 : param pin : digital pin number
 : param threshold _ type : DIGITAL _ LATCH _ HIGH | DIGITAL _ LATCH _ LOW
 : param cb : User provided callback function"""

implementation:
with self . pymata . data_lock :
    self . digital_latch_table [ pin ] = [ self . LATCH_ARMED , threshold_type , 0 , 0 , cb ]


signature:
def get_analog_latch_data ( self , pin ) :
    """This method reads the analog latch table for the specified pin and returns a list that contains :
 [ latch _ state , latched _ data , and time _ stamp ] .
 If the latch state is latched , the entry in the table is cleared
 : param pin : pin number
 : return : [ latch _ state , latched _ data , and time _ stamp ]"""

implementation:
with self . pymata . data_lock :
    pin_data = self . analog_latch_table [ pin ]
    current_latch_data = [ pin , pin_data [ self . LATCH_STATE ] , pin_data [ self . ANALOG_LATCHED_DATA ] , pin_data [ self . ANALOG_TIME_STAMP ] , pin_data [ self . ANALOG_LATCH_CALLBACK ] ]
    # if this is latched data , clear the latch table entry for this pin
    if pin_data [ self . LATCH_STATE ] == self . LATCH_LATCHED :
        self . analog_latch_table [ pin ] = [ 0 , 0 , 0 , 0 , 0 , None ]
return current_latch_data


signature:
def get_digital_latch_data ( self , pin ) :
    """This method reads the digital latch table for the specified pin and returns a list that contains :
 [ latch _ state , latched _ data , and time _ stamp ] .
 If the latch state is latched , the entry in the table is cleared
 : param pin : pin number
 : return : [ latch _ state , latched _ data , and time _ stamp ]"""

implementation:
with self . pymata . data_lock :
    pin_data = self . digital_latch_table [ pin ]
    current_latch_data = [ pin , pin_data [ self . LATCH_STATE ] , pin_data [ self . DIGITAL_LATCHED_DATA ] , pin_data [ self . DIGITAL_TIME_STAMP ] , pin_data [ self . DIGITAL_LATCH_CALLBACK ] ]
    if pin_data [ self . LATCH_STATE ] == self . LATCH_LATCHED :
        self . digital_latch_table [ pin ] = [ 0 , 0 , 0 , 0 , None ]
return current_latch_data


signature:
def report_firmware ( self , data ) :
    """This method processes the report firmware message , sent asynchronously by Firmata when it starts up
 or after refresh _ report _ firmware ( ) is called
 Use the api method api _ get _ firmware _ version to retrieve this information
 : param data : Message data from Firmata
 : return : No return value ."""

implementation:
self . firmata_firmware . append ( data [ 0 ] )
# add major
self . firmata_firmware . append ( data [ 1 ] )
# add minor
# extract the file name string from the message
# file name is in bytes 2 to the end
name_data = data [ 2 : ]
# constructed file name
file_name = [ ]
# the file name is passed in with each character as 2 bytes , the high order byte is equal to 0
# so skip over these zero bytes
for i in name_data [ : : 2 ] :
    file_name . append ( chr ( i ) )
# add filename to tuple
self . firmata_firmware . append ( "" . join ( file_name ) )