signature:
def add_file_contribution ( self , doc , value ) :
    """Raises OrderError if no package or file defined ."""

implementation:
if self . has_package ( doc ) and self . has_file ( doc ) :
    self . file ( doc ) . add_contrib ( value )
else :
    raise OrderError ( 'File::Contributor' )


signature:
def add_file_dep ( self , doc , value ) :
    """Raises OrderError if no package or file defined ."""

implementation:
if self . has_package ( doc ) and self . has_file ( doc ) :
    self . file ( doc ) . add_depend ( value )
else :
    raise OrderError ( 'File::Dependency' )


signature:
def set_file_atrificat_of_project ( self , doc , symbol , value ) :
    """Sets a file name , uri or home artificat .
 Raises OrderError if no package or file defined ."""

implementation:
if self . has_package ( doc ) and self . has_file ( doc ) :
    self . file ( doc ) . add_artifact ( symbol , value )
else :
    raise OrderError ( 'File::Artificat' )


signature:
def reset_file_stat ( self ) :
    """Resets the builder ' s state to enable building new files ."""

implementation:
# FIXME : this state does not make sense
self . file_spdx_id_set = False
self . file_comment_set = False
self . file_type_set = False
self . file_chksum_set = False
self . file_conc_lics_set = False
self . file_license_comment_set = False
self . file_notice_set = False
self . file_copytext_set = False


signature:
def set_lic_id ( self , doc , lic_id ) :
    """Adds a new extracted license to the document .
 Raises SPDXValueError if data format is incorrect ."""

implementation:
# FIXME : this state does not make sense
self . reset_extr_lics ( )
if validations . validate_extracted_lic_id ( lic_id ) :
    doc . add_extr_lic ( document . ExtractedLicense ( lic_id ) )
    return True
else :
    raise SPDXValueError ( 'ExtractedLicense::id' )


signature:
def set_lic_text ( self , doc , text ) :
    """Sets license extracted text .
 Raises SPDXValueError if text is not free form text .
 Raises OrderError if no license ID defined ."""

implementation:
if self . has_extr_lic ( doc ) :
    if not self . extr_text_set :
        self . extr_text_set = True
        if validations . validate_is_free_form_text ( text ) :
            self . extr_lic ( doc ) . text = str_from_text ( text )
            return True
        else :
            raise SPDXValueError ( 'ExtractedLicense::text' )
    else :
        raise CardinalityError ( 'ExtractedLicense::text' )
else :
    raise OrderError ( 'ExtractedLicense::text' )


signature:
def set_lic_name ( self , doc , name ) :
    """Sets license name .
 Raises SPDXValueError if name is not str or utils . NoAssert
 Raises OrderError if no license id defined ."""

implementation:
if self . has_extr_lic ( doc ) :
    if not self . extr_lic_name_set :
        self . extr_lic_name_set = True
        if validations . validate_extr_lic_name ( name ) :
            self . extr_lic ( doc ) . full_name = name
            return True
        else :
            raise SPDXValueError ( 'ExtractedLicense::Name' )
    else :
        raise CardinalityError ( 'ExtractedLicense::Name' )
else :
    raise OrderError ( 'ExtractedLicense::Name' )


signature:
def set_lic_comment ( self , doc , comment ) :
    """Sets license comment .
 Raises SPDXValueError if comment is not free form text .
 Raises OrderError if no license ID defined ."""

implementation:
if self . has_extr_lic ( doc ) :
    if not self . extr_lic_comment_set :
        self . extr_lic_comment_set = True
        if validations . validate_is_free_form_text ( comment ) :
            self . extr_lic ( doc ) . comment = str_from_text ( comment )
            return True
        else :
            raise SPDXValueError ( 'ExtractedLicense::comment' )
    else :
        raise CardinalityError ( 'ExtractedLicense::comment' )
else :
    raise OrderError ( 'ExtractedLicense::comment' )


signature:
def add_lic_xref ( self , doc , ref ) :
    """Adds a license cross reference .
 Raises OrderError if no License ID defined ."""

implementation:
if self . has_extr_lic ( doc ) :
    self . extr_lic ( doc ) . add_xref ( ref )
    return True
else :
    raise OrderError ( 'ExtractedLicense::CrossRef' )


signature:
def reset ( self ) :
    """Resets builder ' s state for building new documents .
 Must be called between usage with different documents ."""

implementation:
# FIXME : this state does not make sense
self . reset_creation_info ( )
self . reset_document ( )
self . reset_package ( )
self . reset_file_stat ( )
self . reset_reviews ( )
self . reset_annotations ( )
self . reset_extr_lics ( )


signature:
def datetime_iso_format ( date ) :
    """Return an ISO - 8601 representation of a datetime object ."""

implementation: return "{0:0>4}-{1:0>2}-{2:0>2}T{3:0>2}:{4:0>2}:{5:0>2}Z" . format ( date . year , date . month , date . day , date . hour , date . minute , date . second )


signature:
def datetime_from_iso_format ( string ) :
    """Return a datetime object from an iso 8601 representation .
 Return None if string is non conforming ."""

implementation:
match = DATE_ISO_REGEX . match ( string )
if match :
    date = datetime . datetime ( year = int ( match . group ( DATE_ISO_YEAR_GRP ) ) , month = int ( match . group ( DATE_ISO_MONTH_GRP ) ) , day = int ( match . group ( DATE_ISO_DAY_GRP ) ) , hour = int ( match . group ( DATE_ISO_HOUR_GRP ) ) , second = int ( match . group ( DATE_ISO_SEC_GRP ) ) , minute = int ( match . group ( DATE_ISO_MIN_GRP ) ) )
    return date
else :
    return None


signature:
def build ( self , ** kwargs ) :
    """Must be called before parse ."""

implementation: self . yacc = yacc . yacc ( module = self , ** kwargs )


signature:
def parse ( self , data ) :
    """Parses a license list and returns a License or None if it failed ."""

implementation:
try :
    return self . yacc . parse ( data , lexer = self . lex )
except :
    return None


signature:
def write_document ( document , out , validate = True ) :
    """Write an SPDX RDF document .
 - document - spdx . document instance .
 - out - file like object that will be written to .
 Optionally ` validate ` the document before writing and raise
 InvalidDocumentError if document . validate returns False ."""

implementation:
if validate :
    messages = [ ]
    messages = document . validate ( messages )
    if messages :
        raise InvalidDocumentError ( messages )
writer = Writer ( document , out )
writer . write ( )


signature:
def create_checksum_node ( self , chksum ) :
    """Return a node representing spdx . checksum ."""

implementation:
chksum_node = BNode ( )
type_triple = ( chksum_node , RDF . type , self . spdx_namespace . Checksum )
self . graph . add ( type_triple )
algorithm_triple = ( chksum_node , self . spdx_namespace . algorithm , Literal ( chksum . identifier ) )
self . graph . add ( algorithm_triple )
value_triple = ( chksum_node , self . spdx_namespace . checksumValue , Literal ( chksum . value ) )
self . graph . add ( value_triple )
return chksum_node


signature:
def to_special_value ( self , value ) :
    """Return proper spdx term or Literal"""

implementation:
if isinstance ( value , utils . NoAssert ) :
    return self . spdx_namespace . noassertion
elif isinstance ( value , utils . SPDXNone ) :
    return self . spdx_namespace . none
else :
    return Literal ( value )


signature:
def licenses_from_tree ( self , tree ) :
    """Traverse conjunctions and disjunctions like trees and return a
 set of all licenses in it as nodes ."""

implementation:
# FIXME : this is unordered !
licenses = set ( )
self . licenses_from_tree_helper ( tree , licenses )
return licenses


signature:
def create_conjunction_node ( self , conjunction ) :
    """Return a node representing a conjunction of licenses ."""

implementation:
node = BNode ( )
type_triple = ( node , RDF . type , self . spdx_namespace . ConjunctiveLicenseSet )
self . graph . add ( type_triple )
licenses = self . licenses_from_tree ( conjunction )
for lic in licenses :
    member_triple = ( node , self . spdx_namespace . member , lic )
    self . graph . add ( member_triple )
return node


signature:
def create_disjunction_node ( self , disjunction ) :
    """Return a node representing a disjunction of licenses ."""

implementation:
node = BNode ( )
type_triple = ( node , RDF . type , self . spdx_namespace . DisjunctiveLicenseSet )
self . graph . add ( type_triple )
licenses = self . licenses_from_tree ( disjunction )
for lic in licenses :
    member_triple = ( node , self . spdx_namespace . member , lic )
    self . graph . add ( member_triple )
return node


signature:
def create_license_helper ( self , lic ) :
    """Handle single ( no conjunction / disjunction ) licenses .
 Return the created node ."""

implementation:
if isinstance ( lic , document . ExtractedLicense ) :
    return self . create_extracted_license ( lic )
if lic . identifier . rstrip ( '+' ) in config . LICENSE_MAP :
    return URIRef ( lic . url )
else :
    matches = [ l for l in self . document . extracted_licenses if l . identifier == lic . identifier ]
    if len ( matches ) != 0 :
        return self . create_extracted_license ( matches [ 0 ] )
    else :
        raise InvalidDocumentError ( 'Missing extracted license: {0}' . format ( lic . identifier ) )


signature:
def create_extracted_license ( self , lic ) :
    """Handle extracted license .
 Return the license node ."""

implementation:
licenses = list ( self . graph . triples ( ( None , self . spdx_namespace . licenseId , lic . identifier ) ) )
if len ( licenses ) != 0 :
    return licenses [ 0 ] [ 0 ]
    # return subject in first triple
else :
    license_node = BNode ( )
    type_triple = ( license_node , RDF . type , self . spdx_namespace . ExtractedLicensingInfo )
    self . graph . add ( type_triple )
    ident_triple = ( license_node , self . spdx_namespace . licenseId , Literal ( lic . identifier ) )
    self . graph . add ( ident_triple )
    text_triple = ( license_node , self . spdx_namespace . extractedText , Literal ( lic . text ) )
    self . graph . add ( text_triple )
    if lic . full_name is not None :
        name_triple = ( license_node , self . spdx_namespace . licenseName , self . to_special_value ( lic . full_name ) )
        self . graph . add ( name_triple )
    for ref in lic . cross_ref :
        triple = ( license_node , RDFS . seeAlso , URIRef ( ref ) )
        self . graph . add ( triple )
    if lic . comment is not None :
        comment_triple = ( license_node , RDFS . comment , Literal ( lic . comment ) )
        self . graph . add ( comment_triple )
    return license_node


signature:
def create_license_node ( self , lic ) :
    """Return a node representing a license .
 Could be a single license ( extracted or part of license list . ) or
 a conjunction / disjunction of licenses ."""

implementation:
if isinstance ( lic , document . LicenseConjunction ) :
    return self . create_conjunction_node ( lic )
elif isinstance ( lic , document . LicenseDisjunction ) :
    return self . create_disjunction_node ( lic )
else :
    return self . create_license_helper ( lic )


signature:
def license_or_special ( self , lic ) :
    """Check for special values spdx : none and spdx : noassertion .
 Return the term for the special value or the result of passing
 license to create _ license _ node ."""

implementation:
if isinstance ( lic , utils . NoAssert ) :
    return self . spdx_namespace . noassertion
elif isinstance ( lic , utils . SPDXNone ) :
    return self . spdx_namespace . none
else :
    return self . create_license_node ( lic )


signature:
def create_file_node ( self , doc_file ) :
    """Create a node for spdx . file ."""

implementation:
file_node = URIRef ( 'http://www.spdx.org/files#{id}' . format ( id = str ( doc_file . spdx_id ) ) )
type_triple = ( file_node , RDF . type , self . spdx_namespace . File )
self . graph . add ( type_triple )
name_triple = ( file_node , self . spdx_namespace . fileName , Literal ( doc_file . name ) )
self . graph . add ( name_triple )
if doc_file . has_optional_field ( 'comment' ) :
    comment_triple = ( file_node , RDFS . comment , Literal ( doc_file . comment ) )
    self . graph . add ( comment_triple )
if doc_file . has_optional_field ( 'type' ) :
    ftype = self . spdx_namespace [ self . FILE_TYPES [ doc_file . type ] ]
    ftype_triple = ( file_node , self . spdx_namespace . fileType , ftype )
    self . graph . add ( ftype_triple )
self . graph . add ( ( file_node , self . spdx_namespace . checksum , self . create_checksum_node ( doc_file . chk_sum ) ) )
conc_lic_node = self . license_or_special ( doc_file . conc_lics )
conc_lic_triple = ( file_node , self . spdx_namespace . licenseConcluded , conc_lic_node )
self . graph . add ( conc_lic_triple )
license_info_nodes = map ( self . license_or_special , doc_file . licenses_in_file )
for lic in license_info_nodes :
    triple = ( file_node , self . spdx_namespace . licenseInfoInFile , lic )
    self . graph . add ( triple )
if doc_file . has_optional_field ( 'license_comment' ) :
    comment_triple = ( file_node , self . spdx_namespace . licenseComments , Literal ( doc_file . license_comment ) )
    self . graph . add ( comment_triple )
cr_text_node = self . to_special_value ( doc_file . copyright )
cr_text_triple = ( file_node , self . spdx_namespace . copyrightText , cr_text_node )
self . graph . add ( cr_text_triple )
if doc_file . has_optional_field ( 'notice' ) :
    notice_triple = ( file_node , self . spdx_namespace . noticeText , doc_file . notice )
    self . graph . add ( notice_triple )
contrib_nodes = map ( lambda c : Literal ( c ) , doc_file . contributors )
contrib_triples = [ ( file_node , self . spdx_namespace . fileContributor , node ) for node in contrib_nodes ]
for triple in contrib_triples :
    self . graph . add ( triple )
return file_node


signature:
def add_file_dependencies_helper ( self , doc_file ) :
    """Handle dependencies for a single file .
 - doc _ file - instance of spdx . file . File ."""

implementation:
subj_triples = list ( self . graph . triples ( ( None , self . spdx_namespace . fileName , Literal ( doc_file . name ) ) ) )
if len ( subj_triples ) != 1 :
    raise InvalidDocumentError ( 'Could not find dependency subject {0}' . format ( doc_file . name ) )
subject_node = subj_triples [ 0 ] [ 0 ]
for dependency in doc_file . dependencies :
    dep_triples = list ( self . graph . triples ( ( None , self . spdx_namespace . fileName , Literal ( dependency ) ) ) )
    if len ( dep_triples ) == 1 :
        dep_node = dep_triples [ 0 ] [ 0 ]
        dep_triple = ( subject_node , self . spdx_namespace . fileDependency , dep_node )
        self . graph . add ( dep_triple )
    else :
        print ( 'Warning could not resolve file dependency {0} -> {1}' . format ( doc_file . name , dependency ) )


signature:
def create_review_node ( self , review ) :
    """Return a review node ."""

implementation:
review_node = BNode ( )
type_triple = ( review_node , RDF . type , self . spdx_namespace . Review )
self . graph . add ( type_triple )
reviewer_node = Literal ( review . reviewer . to_value ( ) )
self . graph . add ( ( review_node , self . spdx_namespace . reviewer , reviewer_node ) )
reviewed_date_node = Literal ( review . review_date_iso_format )
reviewed_triple = ( review_node , self . spdx_namespace . reviewDate , reviewed_date_node )
self . graph . add ( reviewed_triple )
if review . has_comment :
    comment_node = Literal ( review . comment )
    comment_triple = ( review_node , RDFS . comment , comment_node )
    self . graph . add ( comment_triple )
return review_node


signature:
def create_annotation_node ( self , annotation ) :
    """Return an annotation node ."""

implementation:
annotation_node = URIRef ( str ( annotation . spdx_id ) )
type_triple = ( annotation_node , RDF . type , self . spdx_namespace . Annotation )
self . graph . add ( type_triple )
annotator_node = Literal ( annotation . annotator . to_value ( ) )
self . graph . add ( ( annotation_node , self . spdx_namespace . annotator , annotator_node ) )
annotation_date_node = Literal ( annotation . annotation_date_iso_format )
annotation_triple = ( annotation_node , self . spdx_namespace . annotationDate , annotation_date_node )
self . graph . add ( annotation_triple )
if annotation . has_comment :
    comment_node = Literal ( annotation . comment )
    comment_triple = ( annotation_node , RDFS . comment , comment_node )
    self . graph . add ( comment_triple )
annotation_type_node = Literal ( annotation . annotation_type )
annotation_type_triple = ( annotation_node , self . spdx_namespace . annotationType , annotation_type_node )
self . graph . add ( annotation_type_triple )
return annotation_node


signature:
def creators ( self ) :
    """Return a list of creator nodes .
 Note : Does not add anything to the graph ."""

implementation: return map ( lambda c : Literal ( c . to_value ( ) ) , self . document . creation_info . creators )


signature:
def create_creation_info ( self ) :
    """Add and return a creation info node to graph"""

implementation:
ci_node = BNode ( )
# Type property
type_triple = ( ci_node , RDF . type , self . spdx_namespace . CreationInfo )
self . graph . add ( type_triple )
created_date = Literal ( self . document . creation_info . created_iso_format )
created_triple = ( ci_node , self . spdx_namespace . created , created_date )
self . graph . add ( created_triple )
creators = self . creators ( )
for creator in creators :
    self . graph . add ( ( ci_node , self . spdx_namespace . creator , creator ) )
if self . document . creation_info . has_comment :
    comment_node = Literal ( self . document . creation_info . comment )
    comment_triple = ( ci_node , RDFS . comment , comment_node )
    self . graph . add ( comment_triple )
return ci_node


signature:
def create_external_document_ref_node ( self , ext_document_references ) :
    """Add and return a creation info node to graph"""

implementation:
ext_doc_ref_node = BNode ( )
type_triple = ( ext_doc_ref_node , RDF . type , self . spdx_namespace . ExternalDocumentRef )
self . graph . add ( type_triple )
ext_doc_id = Literal ( ext_document_references . external_document_id )
ext_doc_id_triple = ( ext_doc_ref_node , self . spdx_namespace . externalDocumentId , ext_doc_id )
self . graph . add ( ext_doc_id_triple )
doc_uri = Literal ( ext_document_references . spdx_document_uri )
doc_uri_triple = ( ext_doc_ref_node , self . spdx_namespace . spdxDocument , doc_uri )
self . graph . add ( doc_uri_triple )
checksum_node = self . create_checksum_node ( ext_document_references . check_sum )
self . graph . add ( ( ext_doc_ref_node , self . spdx_namespace . checksum , checksum_node ) )
return ext_doc_ref_node


signature:
def package_verif_node ( self , package ) :
    """Return a node representing package verification code ."""

implementation:
verif_node = BNode ( )
type_triple = ( verif_node , RDF . type , self . spdx_namespace . PackageVerificationCode )
self . graph . add ( type_triple )
value_triple = ( verif_node , self . spdx_namespace . packageVerificationCodeValue , Literal ( package . verif_code ) )
self . graph . add ( value_triple )
excl_file_nodes = map ( lambda excl : Literal ( excl ) , package . verif_exc_files )
excl_predicate = self . spdx_namespace . packageVerificationCodeExcludedFile
excl_file_triples = [ ( verif_node , excl_predicate , xcl_file ) for xcl_file in excl_file_nodes ]
for trp in excl_file_triples :
    self . graph . add ( trp )
return verif_node


signature:
def handle_package_literal_optional ( self , package , package_node , predicate , field ) :
    """Check if optional field is set .
 If so it adds the triple ( package _ node , predicate , $ ) to the graph .
 Where $ is a literal or special value term of the value of the field ."""

implementation:
if package . has_optional_field ( field ) :
    value = getattr ( package , field , None )
    value_node = self . to_special_value ( value )
    triple = ( package_node , predicate , value_node )
    self . graph . add ( triple )


signature:
def handle_pkg_optional_fields ( self , package , package_node ) :
    """Write package optional fields ."""

implementation:
self . handle_package_literal_optional ( package , package_node , self . spdx_namespace . versionInfo , 'version' )
self . handle_package_literal_optional ( package , package_node , self . spdx_namespace . packageFileName , 'file_name' )
self . handle_package_literal_optional ( package , package_node , self . spdx_namespace . supplier , 'supplier' )
self . handle_package_literal_optional ( package , package_node , self . spdx_namespace . originator , 'originator' )
self . handle_package_literal_optional ( package , package_node , self . spdx_namespace . sourceInfo , 'source_info' )
self . handle_package_literal_optional ( package , package_node , self . spdx_namespace . licenseComments , 'license_comment' )
self . handle_package_literal_optional ( package , package_node , self . spdx_namespace . summary , 'summary' )
self . handle_package_literal_optional ( package , package_node , self . spdx_namespace . description , 'description' )
if package . has_optional_field ( 'check_sum' ) :
    checksum_node = self . create_checksum_node ( package . check_sum )
    self . graph . add ( ( package_node , self . spdx_namespace . checksum , checksum_node ) )
if package . has_optional_field ( 'homepage' ) :
    homepage_node = URIRef ( self . to_special_value ( package . homepage ) )
    homepage_triple = ( package_node , self . doap_namespace . homepage , homepage_node )
    self . graph . add ( homepage_triple )


signature:
def create_package_node ( self , package ) :
    """Return a Node representing the package .
 Files must have been added to the graph before this method is called ."""

implementation:
package_node = BNode ( )
type_triple = ( package_node , RDF . type , self . spdx_namespace . Package )
self . graph . add ( type_triple )
# Handle optional fields :
self . handle_pkg_optional_fields ( package , package_node )
# package name
name_triple = ( package_node , self . spdx_namespace . name , Literal ( package . name ) )
self . graph . add ( name_triple )
# Package download location
down_loc_node = ( package_node , self . spdx_namespace . downloadLocation , self . to_special_value ( package . download_location ) )
self . graph . add ( down_loc_node )
# Handle package verification
verif_node = self . package_verif_node ( package )
verif_triple = ( package_node , self . spdx_namespace . packageVerificationCode , verif_node )
self . graph . add ( verif_triple )
# Handle concluded license
conc_lic_node = self . license_or_special ( package . conc_lics )
conc_lic_triple = ( package_node , self . spdx_namespace . licenseConcluded , conc_lic_node )
self . graph . add ( conc_lic_triple )
# Handle declared license
decl_lic_node = self . license_or_special ( package . license_declared )
decl_lic_triple = ( package_node , self . spdx_namespace . licenseDeclared , decl_lic_node )
self . graph . add ( decl_lic_triple )
# Package licenses from files
licenses_from_files_nodes = map ( lambda el : self . license_or_special ( el ) , package . licenses_from_files )
lic_from_files_predicate = self . spdx_namespace . licenseInfoFromFiles
lic_from_files_triples = [ ( package_node , lic_from_files_predicate , node ) for node in licenses_from_files_nodes ]
for triple in lic_from_files_triples :
    self . graph . add ( triple )
# Copyright Text
cr_text_node = self . to_special_value ( package . cr_text )
cr_text_triple = ( package_node , self . spdx_namespace . copyrightText , cr_text_node )
self . graph . add ( cr_text_triple )
# Handle files
self . handle_package_has_file ( package , package_node )
return package_node


signature:
def handle_package_has_file_helper ( self , pkg_file ) :
    """Return node representing pkg _ file
 pkg _ file should be instance of spdx . file ."""

implementation:
nodes = list ( self . graph . triples ( ( None , self . spdx_namespace . fileName , Literal ( pkg_file . name ) ) ) )
if len ( nodes ) == 1 :
    return nodes [ 0 ] [ 0 ]
else :
    raise InvalidDocumentError ( 'handle_package_has_file_helper could not' + ' find file node for file: {0}' . format ( pkg_file . name ) )


signature:
def handle_package_has_file ( self , package , package_node ) :
    """Add hasFile triples to graph .
 Must be called after files have been added ."""

implementation:
file_nodes = map ( self . handle_package_has_file_helper , package . files )
triples = [ ( package_node , self . spdx_namespace . hasFile , node ) for node in file_nodes ]
for triple in triples :
    self . graph . add ( triple )


signature:
def create_doc ( self ) :
    """Add and return the root document node to graph ."""

implementation:
doc_node = URIRef ( 'http://www.spdx.org/tools#SPDXRef-DOCUMENT' )
# Doc type
self . graph . add ( ( doc_node , RDF . type , self . spdx_namespace . SpdxDocument ) )
# Version
vers_literal = Literal ( str ( self . document . version ) )
self . graph . add ( ( doc_node , self . spdx_namespace . specVersion , vers_literal ) )
# Data license
data_lics = URIRef ( self . document . data_license . url )
self . graph . add ( ( doc_node , self . spdx_namespace . dataLicense , data_lics ) )
doc_name = URIRef ( self . document . name )
self . graph . add ( ( doc_node , self . spdx_namespace . name , doc_name ) )
return doc_node


signature:
def load_license_list ( file_name ) :
    """Return the licenses list version tuple and a mapping of licenses
 name - > id and id - > name loaded from a JSON file
 from https : / / github . com / spdx / license - list - data"""

implementation:
licenses_map = { }
with codecs . open ( file_name , 'rb' , encoding = 'utf-8' ) as lics :
    licenses = json . load ( lics )
    version = licenses [ 'licenseListVersion' ] . split ( '.' )
    for lic in licenses [ 'licenses' ] :
        if lic . get ( 'isDeprecatedLicenseId' ) :
            continue
        name = lic [ 'name' ]
        identifier = lic [ 'licenseId' ]
        licenses_map [ name ] = identifier
        licenses_map [ identifier ] = name
return version , licenses_map


signature:
def validate ( self , messages ) :
    """Returns True if the fields are valid according to the SPDX standard .
 Appends user friendly messages to the messages parameter ."""

implementation:
messages = self . validate_creators ( messages )
messages = self . validate_created ( messages )
return messages


signature:
def more_than_one_error ( self , field ) :
    """Logs a more than one error .
 field is the field / property that has more than one defined ."""

implementation:
msg = 'More than one {0} defined.' . format ( field )
self . logger . log ( msg )
self . error = True


signature:
def value_error ( self , key , bad_value ) :
    """Reports a value error using ERROR _ MESSAGES dict .
 key - key to use for ERROR _ MESSAGES .
 bad _ value - is passed to format which is called on what key maps to
 in ERROR _ MESSAGES ."""

implementation:
msg = ERROR_MESSAGES [ key ] . format ( bad_value )
self . logger . log ( msg )
self . error = True


signature:
def to_special_value ( self , value ) :
    """Checks if value is a special SPDX value such as
 NONE , NOASSERTION or UNKNOWN if so returns proper model .
 else returns value"""

implementation:
if value == self . spdx_namespace . none :
    return utils . SPDXNone ( )
elif value == self . spdx_namespace . noassertion :
    return utils . NoAssert ( )
elif value == self . spdx_namespace . unknown :
    return utils . UnKnown ( )
else :
    return value


signature:
def handle_lics ( self , lics ) :
    """Return a License from a ` lics ` license resource ."""

implementation:
# Handle extracted licensing info type .
if ( lics , RDF . type , self . spdx_namespace [ 'ExtractedLicensingInfo' ] ) in self . graph :
    return self . parse_only_extr_license ( lics )
# Assume resource , hence the path separator
ident_start = lics . rfind ( '/' ) + 1
if ident_start == 0 : # special values such as spdx : noassertion
    special = self . to_special_value ( lics )
    if special == lics :
        if self . LICS_REF_REGEX . match ( lics ) : # Is a license ref i . e LicenseRef - 1
            return document . License . from_identifier ( lics )
        else : # Not a known license form
            raise SPDXValueError ( 'License' )
    else : # is a special value
        return special
else : # license url
    return document . License . from_identifier ( lics [ ident_start : ] )


signature:
def get_extr_license_ident ( self , extr_lic ) :
    """Return an a license identifier from an ExtractedLicense or None ."""

implementation:
identifier_tripples = list ( self . graph . triples ( ( extr_lic , self . spdx_namespace [ 'licenseId' ] , None ) ) )
if not identifier_tripples :
    self . error = True
    msg = 'Extracted license must have licenseId property.'
    self . logger . log ( msg )
    return
if len ( identifier_tripples ) > 1 :
    self . more_than_one_error ( 'extracted license identifier_tripples' )
    return
identifier_tripple = identifier_tripples [ 0 ]
_s , _p , identifier = identifier_tripple
return identifier


signature:
def get_extr_license_text ( self , extr_lic ) :
    """Return extracted text from an ExtractedLicense or None ."""

implementation:
text_tripples = list ( self . graph . triples ( ( extr_lic , self . spdx_namespace [ 'extractedText' ] , None ) ) )
if not text_tripples :
    self . error = True
    msg = 'Extracted license must have extractedText property'
    self . logger . log ( msg )
    return
if len ( text_tripples ) > 1 :
    self . more_than_one_error ( 'extracted license text' )
    return
text_tripple = text_tripples [ 0 ]
_s , _p , text = text_tripple
return text


signature:
def get_extr_lic_name ( self , extr_lic ) :
    """Return the license name from an ExtractedLicense or None"""

implementation:
extr_name_list = list ( self . graph . triples ( ( extr_lic , self . spdx_namespace [ 'licenseName' ] , None ) ) )
if len ( extr_name_list ) > 1 :
    self . more_than_one_error ( 'extracted license name' )
    return
elif len ( extr_name_list ) == 0 :
    return
return self . to_special_value ( extr_name_list [ 0 ] [ 2 ] )


signature:
def get_extr_lics_xref ( self , extr_lic ) :
    """Return a list of cross references ."""

implementation:
xrefs = list ( self . graph . triples ( ( extr_lic , RDFS . seeAlso , None ) ) )
return map ( lambda xref_triple : xref_triple [ 2 ] , xrefs )


signature:
def get_extr_lics_comment ( self , extr_lics ) :
    """Return license comment or None ."""

implementation:
comment_list = list ( self . graph . triples ( ( extr_lics , RDFS . comment , None ) ) )
if len ( comment_list ) > 1 :
    self . more_than_one_error ( 'extracted license comment' )
    return
elif len ( comment_list ) == 1 :
    return comment_list [ 0 ] [ 2 ]
else :
    return


signature:
def parse_only_extr_license ( self , extr_lic ) :
    """Return an ExtractedLicense object to represent a license object .
 But does not add it to the SPDXDocument model .
 Return None if failed ."""

implementation:
# Grab all possible values
ident = self . get_extr_license_ident ( extr_lic )
text = self . get_extr_license_text ( extr_lic )
comment = self . get_extr_lics_comment ( extr_lic )
xrefs = self . get_extr_lics_xref ( extr_lic )
name = self . get_extr_lic_name ( extr_lic )
if not ident : # Must have identifier
    return
# Set fields
# FIXME : the constructor of the license should alwas accept a name
lic = document . ExtractedLicense ( ident )
if text is not None :
    lic . text = text
if name is not None :
    lic . full_name = name
if comment is not None :
    lic . comment = comment
lic . cross_ref = map ( lambda x : six . text_type ( x ) , xrefs )
return lic


signature:
def handle_extracted_license ( self , extr_lic ) :
    """Build and return an ExtractedLicense or None .
 Note that this function adds the license to the document ."""

implementation:
lic = self . parse_only_extr_license ( extr_lic )
if lic is not None :
    self . doc . add_extr_lic ( lic )
return lic


signature:
def _handle_license_list ( self , lics_set , cls = None ) :
    """Return a license representing a ` cls ` object ( LicenseConjunction
 or LicenseDisjunction ) from a list of license resources or None ."""

implementation:
licenses = [ ]
for _ , _ , lics_member in self . graph . triples ( ( lics_set , self . spdx_namespace [ 'member' ] , None ) ) :
    try :
        if ( lics_member , RDF . type , self . spdx_namespace [ 'ExtractedLicensingInfo' ] ) in self . graph :
            lics = self . handle_extracted_license ( lics_member )
            if lics is not None :
                licenses . append ( lics )
        else :
            licenses . append ( self . handle_lics ( lics_member ) )
    except CardinalityError :
        self . value_error ( 'LICS_LIST_MEMBER' , lics_member )
        break
if len ( licenses ) > 1 :
    return reduce ( lambda a , b : cls ( a , b ) , licenses )
else :
    self . value_error ( 'PKG_CONC_LIST' , '' )
    return


signature:
def parse_package ( self , p_term ) :
    """Parses package fields ."""

implementation:
# Check there is a pacakge name
if not ( p_term , self . spdx_namespace [ 'name' ] , None ) in self . graph :
    self . error = True
    self . logger . log ( 'Package must have a name.' )
    # Create dummy package so that we may continue parsing the rest of
    # the package fields .
    self . builder . create_package ( self . doc , 'dummy_package' )
else :
    for _s , _p , o in self . graph . triples ( ( p_term , self . spdx_namespace [ 'name' ] , None ) ) :
        try :
            self . builder . create_package ( self . doc , six . text_type ( o ) )
        except CardinalityError :
            self . more_than_one_error ( 'Package name' )
            break
self . p_pkg_vinfo ( p_term , self . spdx_namespace [ 'versionInfo' ] )
self . p_pkg_fname ( p_term , self . spdx_namespace [ 'packageFileName' ] )
self . p_pkg_suppl ( p_term , self . spdx_namespace [ 'supplier' ] )
self . p_pkg_originator ( p_term , self . spdx_namespace [ 'originator' ] )
self . p_pkg_down_loc ( p_term , self . spdx_namespace [ 'downloadLocation' ] )
self . p_pkg_homepg ( p_term , self . doap_namespace [ 'homepage' ] )
self . p_pkg_chk_sum ( p_term , self . spdx_namespace [ 'checksum' ] )
self . p_pkg_src_info ( p_term , self . spdx_namespace [ 'sourceInfo' ] )
self . p_pkg_verif_code ( p_term , self . spdx_namespace [ 'packageVerificationCode' ] )
self . p_pkg_lic_conc ( p_term , self . spdx_namespace [ 'licenseConcluded' ] )
self . p_pkg_lic_decl ( p_term , self . spdx_namespace [ 'licenseDeclared' ] )
self . p_pkg_lics_info_from_files ( p_term , self . spdx_namespace [ 'licenseInfoFromFiles' ] )
self . p_pkg_comments_on_lics ( p_term , self . spdx_namespace [ 'licenseComments' ] )
self . p_pkg_cr_text ( p_term , self . spdx_namespace [ 'copyrightText' ] )
self . p_pkg_summary ( p_term , self . spdx_namespace [ 'summary' ] )
self . p_pkg_descr ( p_term , self . spdx_namespace [ 'description' ] )


signature:
def handle_pkg_lic ( self , p_term , predicate , builder_func ) :
    """Handles package lics concluded or declared ."""

implementation:
try :
    for _ , _ , licenses in self . graph . triples ( ( p_term , predicate , None ) ) :
        if ( licenses , RDF . type , self . spdx_namespace [ 'ConjunctiveLicenseSet' ] ) in self . graph :
            lics = self . handle_conjunctive_list ( licenses )
            builder_func ( self . doc , lics )
        elif ( licenses , RDF . type , self . spdx_namespace [ 'DisjunctiveLicenseSet' ] ) in self . graph :
            lics = self . handle_disjunctive_list ( licenses )
            builder_func ( self . doc , lics )
        else :
            try :
                lics = self . handle_lics ( licenses )
                builder_func ( self . doc , lics )
            except SPDXValueError :
                self . value_error ( 'PKG_SINGLE_LICS' , licenses )
except CardinalityError :
    self . more_than_one_error ( 'package {0}' . format ( predicate ) )


signature:
def get_file_name ( self , f_term ) :
    """Returns first found fileName property or None if not found ."""

implementation:
for _ , _ , name in self . graph . triples ( ( f_term , self . spdx_namespace [ 'fileName' ] , None ) ) :
    return name
return


signature:
def p_file_depends ( self , f_term , predicate ) :
    """Sets file dependencies ."""

implementation:
for _ , _ , other_file in self . graph . triples ( ( f_term , predicate , None ) ) :
    name = self . get_file_name ( other_file )
    if name is not None :
        self . builder . add_file_dep ( six . text_type ( name ) )
    else :
        self . error = True
        msg = 'File depends on file with no name'
        self . logger . log ( msg )


signature:
def p_file_contributor ( self , f_term , predicate ) :
    """Parse all file contributors and adds them to the model ."""

implementation:
for _ , _ , contributor in self . graph . triples ( ( f_term , predicate , None ) ) :
    self . builder . add_file_contribution ( self . doc , six . text_type ( contributor ) )


signature:
def p_file_notice ( self , f_term , predicate ) :
    """Sets file notice text ."""

implementation:
try :
    for _ , _ , notice in self . graph . triples ( ( f_term , predicate , None ) ) :
        self . builder . set_file_notice ( self . doc , six . text_type ( notice ) )
except CardinalityError :
    self . more_than_one_error ( 'file notice' )


signature:
def p_file_comment ( self , f_term , predicate ) :
    """Sets file comment text ."""

implementation:
try :
    for _ , _ , comment in self . graph . triples ( ( f_term , predicate , None ) ) :
        self . builder . set_file_comment ( self . doc , six . text_type ( comment ) )
except CardinalityError :
    self . more_than_one_error ( 'file comment' )


signature:
def p_file_artifact ( self , f_term , predicate ) :
    """Handles file artifactOf .
 Note : does not handle artifact of project URI ."""

implementation:
for _ , _ , project in self . graph . triples ( ( f_term , predicate , None ) ) :
    if ( project , RDF . type , self . doap_namespace [ 'Project' ] ) :
        self . p_file_project ( project )
    else :
        self . error = True
        msg = 'File must be artifact of doap:Project'
        self . logger . log ( msg )


signature:
def p_file_project ( self , project ) :
    """Helper function for parsing doap : project name and homepage .
 and setting them using the file builder ."""

implementation:
for _ , _ , name in self . graph . triples ( ( project , self . doap_namespace [ 'name' ] , None ) ) :
    self . builder . set_file_atrificat_of_project ( self . doc , 'name' , six . text_type ( name ) )
for _ , _ , homepage in self . graph . triples ( ( project , self . doap_namespace [ 'homepage' ] , None ) ) :
    self . builder . set_file_atrificat_of_project ( self . doc , 'home' , six . text_type ( homepage ) )


signature:
def p_file_cr_text ( self , f_term , predicate ) :
    """Sets file copyright text ."""

implementation:
try :
    for _ , _ , cr_text in self . graph . triples ( ( f_term , predicate , None ) ) :
        self . builder . set_file_copyright ( self . doc , six . text_type ( cr_text ) )
except CardinalityError :
    self . more_than_one_error ( 'file copyright text' )


signature:
def p_file_comments_on_lics ( self , f_term , predicate ) :
    """Sets file license comment ."""

implementation:
try :
    for _ , _ , comment in self . graph . triples ( ( f_term , predicate , None ) ) :
        self . builder . set_file_license_comment ( self . doc , six . text_type ( comment ) )
except CardinalityError :
    self . more_than_one_error ( 'file comments on license' )


signature:
def p_file_lic_info ( self , f_term , predicate ) :
    """Sets file license information ."""

implementation:
for _ , _ , info in self . graph . triples ( ( f_term , predicate , None ) ) :
    lic = self . handle_lics ( info )
    if lic is not None :
        self . builder . set_file_license_in_file ( self . doc , lic )


signature:
def p_file_type ( self , f_term , predicate ) :
    """Sets file type ."""

implementation:
try :
    for _ , _ , ftype in self . graph . triples ( ( f_term , predicate , None ) ) :
        try :
            if ftype . endswith ( 'binary' ) :
                ftype = 'BINARY'
            elif ftype . endswith ( 'source' ) :
                ftype = 'SOURCE'
            elif ftype . endswith ( 'other' ) :
                ftype = 'OTHER'
            elif ftype . endswith ( 'archive' ) :
                ftype = 'ARCHIVE'
            self . builder . set_file_type ( self . doc , ftype )
        except SPDXValueError :
            self . value_error ( 'FILE_TYPE' , ftype )
except CardinalityError :
    self . more_than_one_error ( 'file type' )


signature:
def p_file_chk_sum ( self , f_term , predicate ) :
    """Sets file checksum . Assumes SHA1 algorithm without checking ."""

implementation:
try :
    for _s , _p , checksum in self . graph . triples ( ( f_term , predicate , None ) ) :
        for _ , _ , value in self . graph . triples ( ( checksum , self . spdx_namespace [ 'checksumValue' ] , None ) ) :
            self . builder . set_file_chksum ( self . doc , six . text_type ( value ) )
except CardinalityError :
    self . more_than_one_error ( 'File checksum' )


signature:
def p_file_lic_conc ( self , f_term , predicate ) :
    """Sets file licenses concluded ."""

implementation:
try :
    for _ , _ , licenses in self . graph . triples ( ( f_term , predicate , None ) ) :
        if ( licenses , RDF . type , self . spdx_namespace [ 'ConjunctiveLicenseSet' ] ) in self . graph :
            lics = self . handle_conjunctive_list ( licenses )
            self . builder . set_concluded_license ( self . doc , lics )
        elif ( licenses , RDF . type , self . spdx_namespace [ 'DisjunctiveLicenseSet' ] ) in self . graph :
            lics = self . handle_disjunctive_list ( licenses )
            self . builder . set_concluded_license ( self . doc , lics )
        else :
            try :
                lics = self . handle_lics ( licenses )
                self . builder . set_concluded_license ( self . doc , lics )
            except SPDXValueError :
                self . value_error ( 'FILE_SINGLE_LICS' , licenses )
except CardinalityError :
    self . more_than_one_error ( 'file {0}' . format ( predicate ) )


signature:
def get_review_date ( self , r_term ) :
    """Returns review date or None if not found .
 Reports error on failure .
 Note does not check value format ."""

implementation:
reviewed_list = list ( self . graph . triples ( ( r_term , self . spdx_namespace [ 'reviewDate' ] , None ) ) )
if len ( reviewed_list ) != 1 :
    self . error = True
    msg = 'Review must have exactlyone review date'
    self . logger . log ( msg )
    return
return six . text_type ( reviewed_list [ 0 ] [ 2 ] )


signature:
def get_reviewer ( self , r_term ) :
    """Returns reviewer as creator object or None if failed .
 Reports errors on failure ."""

implementation:
reviewer_list = list ( self . graph . triples ( ( r_term , self . spdx_namespace [ 'reviewer' ] , None ) ) )
if len ( reviewer_list ) != 1 :
    self . error = True
    msg = 'Review must have exactly one reviewer'
    self . logger . log ( msg )
    return
try :
    return self . builder . create_entity ( self . doc , six . text_type ( reviewer_list [ 0 ] [ 2 ] ) )
except SPDXValueError :
    self . value_error ( 'REVIEWER_VALUE' , reviewer_list [ 0 ] [ 2 ] )


signature:
def get_annotation_type ( self , r_term ) :
    """Returns annotation type or None if found none or more than one .
 Reports errors on failure ."""

implementation:
for _ , _ , typ in self . graph . triples ( ( r_term , self . spdx_namespace [ 'annotationType' ] , None ) ) :
    if typ is not None :
        return typ
    else :
        self . error = True
        msg = 'Annotation must have exactly one annotation type.'
        self . logger . log ( msg )
        return


signature:
def get_annotation_comment ( self , r_term ) :
    """Returns annotation comment or None if found none or more than one .
 Reports errors ."""

implementation:
comment_list = list ( self . graph . triples ( ( r_term , RDFS . comment , None ) ) )
if len ( comment_list ) > 1 :
    self . error = True
    msg = 'Annotation can have at most one comment.'
    self . logger . log ( msg )
    return
else :
    return six . text_type ( comment_list [ 0 ] [ 2 ] )


signature:
def get_annotation_date ( self , r_term ) :
    """Returns annotation date or None if not found .
 Reports error on failure .
 Note does not check value format ."""

implementation:
annotation_date_list = list ( self . graph . triples ( ( r_term , self . spdx_namespace [ 'annotationDate' ] , None ) ) )
if len ( annotation_date_list ) != 1 :
    self . error = True
    msg = 'Annotation must have exactly one annotation date.'
    self . logger . log ( msg )
    return
return six . text_type ( annotation_date_list [ 0 ] [ 2 ] )


signature:
def parse ( self , fil ) :
    """Parses a file and returns a document object .
 File , a file like object ."""

implementation:
self . error = False
self . graph = Graph ( )
self . graph . parse ( file = fil , format = 'xml' )
self . doc = document . Document ( )
for s , _p , o in self . graph . triples ( ( None , RDF . type , self . spdx_namespace [ 'SpdxDocument' ] ) ) :
    self . parse_doc_fields ( s )
for s , _p , o in self . graph . triples ( ( None , RDF . type , self . spdx_namespace [ 'ExternalDocumentRef' ] ) ) :
    self . parse_ext_doc_ref ( s )
for s , _p , o in self . graph . triples ( ( None , RDF . type , self . spdx_namespace [ 'CreationInfo' ] ) ) :
    self . parse_creation_info ( s )
for s , _p , o in self . graph . triples ( ( None , RDF . type , self . spdx_namespace [ 'Package' ] ) ) :
    self . parse_package ( s )
for s , _p , o in self . graph . triples ( ( None , self . spdx_namespace [ 'referencesFile' ] , None ) ) :
    self . parse_file ( o )
for s , _p , o in self . graph . triples ( ( None , self . spdx_namespace [ 'reviewed' ] , None ) ) :
    self . parse_review ( o )
for s , _p , o in self . graph . triples ( ( None , self . spdx_namespace [ 'annotation' ] , None ) ) :
    self . parse_annotation ( o )
validation_messages = [ ]
# Report extra errors if self . error is False otherwise there will be
# redundent messages
validation_messages = self . doc . validate ( validation_messages )
if not self . error :
    if validation_messages :
        for msg in validation_messages :
            self . logger . log ( msg )
        self . error = True
return self . doc , self . error


signature:
def parse_creation_info ( self , ci_term ) :
    """Parse creators , created and comment ."""

implementation:
for _s , _p , o in self . graph . triples ( ( ci_term , self . spdx_namespace [ 'creator' ] , None ) ) :
    try :
        ent = self . builder . create_entity ( self . doc , six . text_type ( o ) )
        self . builder . add_creator ( self . doc , ent )
    except SPDXValueError :
        self . value_error ( 'CREATOR_VALUE' , o )
for _s , _p , o in self . graph . triples ( ( ci_term , self . spdx_namespace [ 'created' ] , None ) ) :
    try :
        self . builder . set_created_date ( self . doc , six . text_type ( o ) )
    except SPDXValueError :
        self . value_error ( 'CREATED_VALUE' , o )
    except CardinalityError :
        self . more_than_one_error ( 'created' )
        break
for _s , _p , o in self . graph . triples ( ( ci_term , RDFS . comment , None ) ) :
    try :
        self . builder . set_creation_comment ( self . doc , six . text_type ( o ) )
    except CardinalityError :
        self . more_than_one_error ( 'CreationInfo comment' )
        break
for _s , _p , o in self . graph . triples ( ( ci_term , self . spdx_namespace [ 'licenseListVersion' ] , None ) ) :
    try :
        self . builder . set_lics_list_ver ( self . doc , six . text_type ( o ) )
    except CardinalityError :
        self . more_than_one_error ( 'licenseListVersion' )
        break
    except SPDXValueError :
        self . value_error ( 'LL_VALUE' , o )


signature:
def parse_doc_fields ( self , doc_term ) :
    """Parses the version , data license , name , SPDX Identifier , namespace ,
 and comment ."""

implementation:
try :
    self . builder . set_doc_spdx_id ( self . doc , doc_term )
except SPDXValueError :
    self . value_error ( 'DOC_SPDX_ID_VALUE' , doc_term )
try :
    if doc_term . count ( '#' , 0 , len ( doc_term ) ) <= 1 :
        doc_namespace = doc_term . split ( '#' ) [ 0 ]
        self . builder . set_doc_namespace ( self . doc , doc_namespace )
    else :
        self . value_error ( 'DOC_NAMESPACE_VALUE' , doc_term )
except SPDXValueError :
    self . value_error ( 'DOC_NAMESPACE_VALUE' , doc_term )
for _s , _p , o in self . graph . triples ( ( doc_term , self . spdx_namespace [ 'specVersion' ] , None ) ) :
    try :
        self . builder . set_doc_version ( self . doc , six . text_type ( o ) )
    except SPDXValueError :
        self . value_error ( 'DOC_VERS_VALUE' , o )
    except CardinalityError :
        self . more_than_one_error ( 'specVersion' )
        break
for _s , _p , o in self . graph . triples ( ( doc_term , self . spdx_namespace [ 'dataLicense' ] , None ) ) :
    try :
        self . builder . set_doc_data_lic ( self . doc , six . text_type ( o ) )
    except SPDXValueError :
        self . value_error ( 'DOC_D_LICS' , o )
    except CardinalityError :
        self . more_than_one_error ( 'dataLicense' )
        break
for _s , _p , o in self . graph . triples ( ( doc_term , self . spdx_namespace [ 'name' ] , None ) ) :
    try :
        self . builder . set_doc_name ( self . doc , six . text_type ( o ) )
    except CardinalityError :
        self . more_than_one_error ( 'name' )
        break
for _s , _p , o in self . graph . triples ( ( doc_term , RDFS . comment , None ) ) :
    try :
        self . builder . set_doc_comment ( self . doc , six . text_type ( o ) )
    except CardinalityError :
        self . more_than_one_error ( 'Document comment' )
        break


signature:
def parse_ext_doc_ref ( self , ext_doc_ref_term ) :
    """Parses the External Document ID , SPDX Document URI and Checksum ."""

implementation:
for _s , _p , o in self . graph . triples ( ( ext_doc_ref_term , self . spdx_namespace [ 'externalDocumentId' ] , None ) ) :
    try :
        self . builder . set_ext_doc_id ( self . doc , six . text_type ( o ) )
    except SPDXValueError :
        self . value_error ( 'EXT_DOC_REF_VALUE' , 'External Document ID' )
        break
for _s , _p , o in self . graph . triples ( ( ext_doc_ref_term , self . spdx_namespace [ 'spdxDocument' ] , None ) ) :
    try :
        self . builder . set_spdx_doc_uri ( self . doc , six . text_type ( o ) )
    except SPDXValueError :
        self . value_error ( 'EXT_DOC_REF_VALUE' , 'SPDX Document URI' )
        break
for _s , _p , checksum in self . graph . triples ( ( ext_doc_ref_term , self . spdx_namespace [ 'checksum' ] , None ) ) :
    for _ , _ , value in self . graph . triples ( ( checksum , self . spdx_namespace [ 'checksumValue' ] , None ) ) :
        try :
            self . builder . set_chksum ( self . doc , six . text_type ( value ) )
        except SPDXValueError :
            self . value_error ( 'EXT_DOC_REF_VALUE' , 'Checksum' )
            break


signature:
def validate ( self , messages ) :
    """Validate the package fields .
 Append user friendly error messages to the ` messages ` list ."""

implementation:
messages = self . validate_checksum ( messages )
messages = self . validate_optional_str_fields ( messages )
messages = self . validate_mandatory_str_fields ( messages )
messages = self . validate_files ( messages )
messages = self . validate_mandatory_fields ( messages )
messages = self . validate_optional_fields ( messages )
return messages


signature:
def validate_optional_str_fields ( self , messages ) :
    """Fields marked as optional and of type string in class
 docstring must be of a type that provides _ _ str _ _ method ."""

implementation:
FIELDS = [ 'file_name' , 'version' , 'homepage' , 'source_info' , 'summary' , 'description' ]
messages = self . validate_str_fields ( FIELDS , True , messages )
return messages


signature:
def validate_mandatory_str_fields ( self , messages ) :
    """Fields marked as Mandatory and of type string in class
 docstring must be of a type that provides _ _ str _ _ method ."""

implementation:
FIELDS = [ 'name' , 'download_location' , 'verif_code' , 'cr_text' ]
messages = self . validate_str_fields ( FIELDS , False , messages )
return messages


signature:
def validate_str_fields ( self , fields , optional , messages ) :
    """Helper for validate _ mandatory _ str _ field and
 validate _ optional _ str _ fields"""

implementation:
for field_str in fields :
    field = getattr ( self , field_str )
    if field is not None : # FIXME : this does not make sense ? ? ?
        attr = getattr ( field , '__str__' , None )
        if not callable ( attr ) :
            messages = messages + [ '{0} must provide __str__ method.' . format ( field ) ]
            # Continue checking .
    elif not optional :
        messages = messages + [ 'Package {0} can not be None.' . format ( field_str ) ]
return messages


signature:
def set_doc_data_lic ( self , doc , res ) :
    """Set the document data license .
 Raise exceptions :
 - SPDXValueError if malformed value ,
 - CardinalityError if already defined ."""

implementation:
if not self . doc_data_lics_set :
    self . doc_data_lics_set = True
    # TODO : what is this split ?
    res_parts = res . split ( '/' )
    if len ( res_parts ) != 0 :
        identifier = res_parts [ - 1 ]
        doc . data_license = document . License . from_identifier ( identifier )
    else :
        raise SPDXValueError ( 'Document::License' )
else :
    raise CardinalityError ( 'Document::License' )


signature:
def set_doc_comment ( self , doc , comment ) :
    """Sets document comment , Raises CardinalityError if
 comment already set ."""

implementation:
if not self . doc_comment_set :
    self . doc_comment_set = True
    doc . comment = comment
else :
    raise CardinalityError ( 'Document::Comment' )


signature:
def set_chksum ( self , doc , chk_sum ) :
    """Sets the external document reference ' s check sum , if not already set .
 chk _ sum - The checksum value in the form of a string ."""

implementation:
if chk_sum :
    doc . ext_document_references [ - 1 ] . check_sum = checksum . Algorithm ( 'SHA1' , chk_sum )
else :
    raise SPDXValueError ( 'ExternalDocumentRef::Checksum' )


signature:
def set_creation_comment ( self , doc , comment ) :
    """Sets creation comment , Raises CardinalityError if
 comment already set .
 Raises SPDXValueError if not free form text ."""

implementation:
if not self . creation_comment_set :
    self . creation_comment_set = True
    doc . creation_info . comment = comment
    return True
else :
    raise CardinalityError ( 'CreationInfo::Comment' )


signature:
def set_pkg_chk_sum ( self , doc , chk_sum ) :
    """Sets the package check sum , if not already set .
 chk _ sum - A string
 Raises CardinalityError if already defined .
 Raises OrderError if no package previously defined ."""

implementation:
self . assert_package_exists ( )
if not self . package_chk_sum_set :
    self . package_chk_sum_set = True
    doc . package . check_sum = checksum . Algorithm ( 'SHA1' , chk_sum )
else :
    raise CardinalityError ( 'Package::CheckSum' )


signature:
def set_pkg_source_info ( self , doc , text ) :
    """Sets the package ' s source information , if not already set .
 text - Free form text .
 Raises CardinalityError if already defined .
 Raises OrderError if no package previously defined ."""

implementation:
self . assert_package_exists ( )
if not self . package_source_info_set :
    self . package_source_info_set = True
    doc . package . source_info = text
    return True
else :
    raise CardinalityError ( 'Package::SourceInfo' )


signature:
def set_pkg_verif_code ( self , doc , code ) :
    """Sets the package verification code , if not already set .
 code - A string .
 Raises CardinalityError if already defined .
 Raises OrderError if no package previously defined ."""

implementation:
self . assert_package_exists ( )
if not self . package_verif_set :
    self . package_verif_set = True
    doc . package . verif_code = code
else :
    raise CardinalityError ( 'Package::VerificationCode' )


signature:
def set_pkg_excl_file ( self , doc , filename ) :
    """Sets the package ' s verification code excluded file .
 Raises OrderError if no package previously defined ."""

implementation:
self . assert_package_exists ( )
doc . package . add_exc_file ( filename )


signature:
def set_pkg_license_comment ( self , doc , text ) :
    """Sets the package ' s license comment .
 Raises OrderError if no package previously defined .
 Raises CardinalityError if already set ."""

implementation:
self . assert_package_exists ( )
if not self . package_license_comment_set :
    self . package_license_comment_set = True
    doc . package . license_comment = text
    return True
else :
    raise CardinalityError ( 'Package::LicenseComment' )


signature:
def set_pkg_cr_text ( self , doc , text ) :
    """Sets the package ' s license comment .
 Raises OrderError if no package previously defined .
 Raises CardinalityError if already set ."""

implementation:
self . assert_package_exists ( )
if not self . package_cr_text_set :
    self . package_cr_text_set = True
    doc . package . cr_text = text
else :
    raise CardinalityError ( 'Package::CopyrightText' )


signature:
def set_pkg_summary ( self , doc , text ) :
    """Set ' s the package summary .
 Raises CardinalityError if summary already set .
 Raises OrderError if no package previously defined ."""

implementation:
self . assert_package_exists ( )
if not self . package_summary_set :
    self . package_summary_set = True
    doc . package . summary = text
else :
    raise CardinalityError ( 'Package::Summary' )


signature:
def set_pkg_desc ( self , doc , text ) :
    """Set ' s the package ' s description .
 Raises CardinalityError if description already set .
 Raises OrderError if no package previously defined ."""

implementation:
self . assert_package_exists ( )
if not self . package_desc_set :
    self . package_desc_set = True
    doc . package . description = text
else :
    raise CardinalityError ( 'Package::Description' )


signature:
def set_file_chksum ( self , doc , chk_sum ) :
    """Sets the file check sum , if not already set .
 chk _ sum - A string
 Raises CardinalityError if already defined .
 Raises OrderError if no package previously defined ."""

implementation:
if self . has_package ( doc ) and self . has_file ( doc ) :
    if not self . file_chksum_set :
        self . file_chksum_set = True
        self . file ( doc ) . chk_sum = checksum . Algorithm ( 'SHA1' , chk_sum )
        return True
    else :
        raise CardinalityError ( 'File::CheckSum' )
else :
    raise OrderError ( 'File::CheckSum' )


signature:
def set_file_license_comment ( self , doc , text ) :
    """Raises OrderError if no package or file defined .
 Raises CardinalityError if more than one per file ."""

implementation:
if self . has_package ( doc ) and self . has_file ( doc ) :
    if not self . file_license_comment_set :
        self . file_license_comment_set = True
        self . file ( doc ) . license_comment = text
        return True
    else :
        raise CardinalityError ( 'File::LicenseComment' )
else :
    raise OrderError ( 'File::LicenseComment' )


signature:
def set_file_copyright ( self , doc , text ) :
    """Raises OrderError if no package or file defined .
 Raises CardinalityError if more than one ."""

implementation:
if self . has_package ( doc ) and self . has_file ( doc ) :
    if not self . file_copytext_set :
        self . file_copytext_set = True
        self . file ( doc ) . copyright = text
        return True
    else :
        raise CardinalityError ( 'File::CopyRight' )
else :
    raise OrderError ( 'File::CopyRight' )


signature:
def set_file_comment ( self , doc , text ) :
    """Raises OrderError if no package or no file defined .
 Raises CardinalityError if more than one comment set ."""

implementation:
if self . has_package ( doc ) and self . has_file ( doc ) :
    if not self . file_comment_set :
        self . file_comment_set = True
        self . file ( doc ) . comment = text
        return True
    else :
        raise CardinalityError ( 'File::Comment' )
else :
    raise OrderError ( 'File::Comment' )


signature:
def set_file_notice ( self , doc , text ) :
    """Raises OrderError if no package or file defined .
 Raises CardinalityError if more than one ."""

implementation:
if self . has_package ( doc ) and self . has_file ( doc ) :
    if not self . file_notice_set :
        self . file_notice_set = True
        self . file ( doc ) . notice = tagvaluebuilders . str_from_text ( text )
        return True
    else :
        raise CardinalityError ( 'File::Notice' )
else :
    raise OrderError ( 'File::Notice' )


signature:
def add_review_comment ( self , doc , comment ) :
    """Sets the review comment . Raises CardinalityError if
 already set . OrderError if no reviewer defined before ."""

implementation:
if len ( doc . reviews ) != 0 :
    if not self . review_comment_set :
        self . review_comment_set = True
        doc . reviews [ - 1 ] . comment = comment
        return True
    else :
        raise CardinalityError ( 'ReviewComment' )
else :
    raise OrderError ( 'ReviewComment' )


signature:
def add_annotation_comment ( self , doc , comment ) :
    """Sets the annotation comment . Raises CardinalityError if
 already set . OrderError if no annotator defined before ."""

implementation:
if len ( doc . annotations ) != 0 :
    if not self . annotation_comment_set :
        self . annotation_comment_set = True
        doc . annotations [ - 1 ] . comment = comment
        return True
    else :
        raise CardinalityError ( 'AnnotationComment' )
else :
    raise OrderError ( 'AnnotationComment' )


signature:
def add_annotation_type ( self , doc , annotation_type ) :
    """Sets the annotation type . Raises CardinalityError if
 already set . OrderError if no annotator defined before ."""

implementation:
if len ( doc . annotations ) != 0 :
    if not self . annotation_type_set :
        if annotation_type . endswith ( 'annotationType_other' ) :
            self . annotation_type_set = True
            doc . annotations [ - 1 ] . annotation_type = 'OTHER'
            return True
        elif annotation_type . endswith ( 'annotationType_review' ) :
            self . annotation_type_set = True
            doc . annotations [ - 1 ] . annotation_type = 'REVIEW'
            return True
        else :
            raise SPDXValueError ( 'Annotation::AnnotationType' )
    else :
        raise CardinalityError ( 'Annotation::AnnotationType' )
else :
    raise OrderError ( 'Annotation::AnnotationType' )


signature:
def validate ( self , messages ) :
    """Validate all fields of the document and update the
 messages list with user friendly error messages for display ."""

implementation:
messages = self . validate_version ( messages )
messages = self . validate_data_lics ( messages )
messages = self . validate_name ( messages )
messages = self . validate_spdx_id ( messages )
messages = self . validate_namespace ( messages )
messages = self . validate_ext_document_references ( messages )
messages = self . validate_creation_info ( messages )
messages = self . validate_package ( messages )
messages = self . validate_extracted_licenses ( messages )
messages = self . validate_reviews ( messages )
return messages


signature:
def include ( f ) :
    '''includes the contents of a file on disk .
 takes a filename'''

implementation:
fl = open ( f , 'r' )
data = fl . read ( )
fl . close ( )
return raw ( data )


signature:
def system ( cmd , data = None ) :
    '''pipes the output of a program'''

implementation:
import subprocess
s = subprocess . Popen ( cmd , shell = True , stdout = subprocess . PIPE , stdin = subprocess . PIPE )
out , err = s . communicate ( data )
return out . decode ( 'utf8' )


signature:
def escape ( data , quote = True ) : # stoled from std lib cgi
    '''Escapes special characters into their html entities
 Replace special characters " & " , " < " and " > " to HTML - safe sequences .
 If the optional flag quote is true , the quotation mark character ( " )
 is also translated .
 This is used to escape content that appears in the body of an HTML cocument'''

implementation:
data = data . replace ( "&" , "&" )
# Must be done first !
data = data . replace ( "<" , "<" )
data = data . replace ( ">" , ">" )
if quote :
    data = data . replace ( '"' , """ )
return data


signature:
def unescape ( data ) :
    '''unescapes html entities . the opposite of escape .'''

implementation:
cc = re . compile ( r'&(?:(?:#(\d+))|([^;]+));' )
result = [ ]
m = cc . search ( data )
while m :
    result . append ( data [ 0 : m . start ( ) ] )
    d = m . group ( 1 )
    if d :
        d = int ( d )
        result . append ( unichr ( d ) )
    else :
        d = _unescape . get ( m . group ( 2 ) , ord ( '?' ) )
        result . append ( unichr ( d ) )
    data = data [ m . end ( ) : ]
    m = cc . search ( data )
result . append ( data )
return '' . join ( result )


signature:
def attr ( * args , ** kwargs ) :
    '''Set attributes on the current active tag context'''

implementation:
ctx = dom_tag . _with_contexts [ _get_thread_context ( ) ]
if ctx and ctx [ - 1 ] :
    dicts = args + ( kwargs , )
    for d in dicts :
        for attr , value in d . items ( ) :
            ctx [ - 1 ] . tag . set_attribute ( * dom_tag . clean_pair ( attr , value ) )
else :
    raise ValueError ( 'not in a tag context' )


signature:
def set_attribute ( self , key , value ) :
    '''Add or update the value of an attribute .'''

implementation:
if isinstance ( key , int ) :
    self . children [ key ] = value
elif isinstance ( key , basestring ) :
    self . attributes [ key ] = value
else :
    raise TypeError ( 'Only integer and string types are valid for assigning ' 'child tags and attributes, respectively.' )


signature:
def setdocument ( self , doc ) :
    '''Creates a reference to the parent document to allow for partial - tree
 validation .'''

implementation:
# assume that a document is correct in the subtree
if self . document != doc :
    self . document = doc
    for i in self . children :
        if not isinstance ( i , dom_tag ) :
            return
        i . setdocument ( doc )


signature:
def add ( self , * args ) :
    '''Add new child tags .'''

implementation:
for obj in args :
    if isinstance ( obj , numbers . Number ) : # Convert to string so we fall into next if block
        obj = str ( obj )
    if isinstance ( obj , basestring ) :
        obj = escape ( obj )
        self . children . append ( obj )
    elif isinstance ( obj , dom_tag ) :
        ctx = dom_tag . _with_contexts [ _get_thread_context ( ) ]
        if ctx and ctx [ - 1 ] :
            ctx [ - 1 ] . used . add ( obj )
        self . children . append ( obj )
        obj . parent = self
        obj . setdocument ( self . document )
    elif isinstance ( obj , dict ) :
        for attr , value in obj . items ( ) :
            self . set_attribute ( * dom_tag . clean_pair ( attr , value ) )
    elif hasattr ( obj , '__iter__' ) :
        for subobj in obj :
            self . add ( subobj )
    else : # wtf is it ?
        raise ValueError ( '%r not a tag or string.' % obj )
if len ( args ) == 1 :
    return args [ 0 ]
return args


signature:
def get ( self , tag = None , ** kwargs ) :
    '''Recursively searches children for tags of a certain
 type with matching attributes .'''

implementation:
# Stupid workaround since we can not use dom _ tag in the method declaration
if tag is None :
    tag = dom_tag
attrs = [ ( dom_tag . clean_attribute ( attr ) , value ) for attr , value in kwargs . items ( ) ]
results = [ ]
for child in self . children :
    if ( isinstance ( tag , basestring ) and type ( child ) . __name__ == tag ) or ( not isinstance ( tag , basestring ) and isinstance ( child , tag ) ) :
        if all ( child . attributes . get ( attribute ) == value for attribute , value in attrs ) : # If the child is of correct type and has all attributes and values
        # in kwargs add as a result
            results . append ( child )
    if isinstance ( child , dom_tag ) : # If the child is a dom _ tag extend the search down through its children
        results . extend ( child . get ( tag , ** kwargs ) )
return results


signature:
def clean_attribute ( attribute ) :
    '''Normalize attribute names for shorthand and work arounds for limitations
 in Python ' s syntax'''

implementation:
# Shorthand
attribute = { 'cls' : 'class' , 'className' : 'class' , 'class_name' : 'class' , 'fr' : 'for' , 'html_for' : 'for' , 'htmlFor' : 'for' , } . get ( attribute , attribute )
# Workaround for Python ' s reserved words
if attribute [ 0 ] == '_' :
    attribute = attribute [ 1 : ]
# Workaround for dash
if attribute in set ( [ 'http_equiv' ] ) or attribute . startswith ( 'data_' ) :
    attribute = attribute . replace ( '_' , '-' ) . lower ( )
# Workaround for colon
if attribute . split ( '_' ) [ 0 ] in ( 'xlink' , 'xml' , 'xmlns' ) :
    attribute = attribute . replace ( '_' , ':' , 1 ) . lower ( )
return attribute


signature:
def clean_pair ( cls , attribute , value ) :
    '''This will call ` clean _ attribute ` on the attribute and also allows for the
 creation of boolean attributes .
 Ex . input ( selected = True ) is equivalent to input ( selected = " selected " )'''

implementation:
attribute = cls . clean_attribute ( attribute )
# Check for boolean attributes
# ( i . e . selected = True becomes selected = " selected " )
if value is True :
    value = attribute
if value is False :
    value = "false"
return ( attribute , value )


signature:
def render ( self , * args , ** kwargs ) :
    '''Creates a < title > tag if not present and renders the DOCTYPE and tag tree .'''

implementation:
r = [ ]
# Validates the tag tree and adds the doctype if one was set
if self . doctype :
    r . append ( self . doctype )
    r . append ( '\n' )
r . append ( super ( document , self ) . render ( * args , ** kwargs ) )
return u'' . join ( r )


signature:
def getElementById ( self , id ) :
    '''DOM API : Returns single element with matching id value .'''

implementation:
results = self . get ( id = id )
if len ( results ) > 1 :
    raise ValueError ( 'Multiple tags with id "%s".' % id )
elif results :
    return results [ 0 ]
else :
    return None


signature:
def getElementsByTagName ( self , name ) :
    '''DOM API : Returns all tags that match name .'''

implementation:
if isinstance ( name , basestring ) :
    return self . get ( name . lower ( ) )
else :
    return None


signature:
def start ( self ) :
    """Create the Interchange process and connect to it ."""

implementation:
self . outgoing_q = zmq_pipes . TasksOutgoing ( "127.0.0.1" , self . interchange_port_range )
self . incoming_q = zmq_pipes . ResultsIncoming ( "127.0.0.1" , self . interchange_port_range )
self . is_alive = True
self . _queue_management_thread = None
self . _start_queue_management_thread ( )
self . _start_local_queue_process ( )
logger . debug ( "Created management thread: {}" . format ( self . _queue_management_thread ) )
if self . provider : # debug _ opts = " - - debug " if self . worker _ debug else " "
    l_cmd = self . launch_cmd . format ( # debug = debug _ opts ,
    task_url = self . worker_task_url , workers_per_node = self . workers_per_node , logdir = "{}/{}" . format ( self . run_dir , self . label ) )
    self . launch_cmd = l_cmd
    logger . debug ( "Launch command: {}" . format ( self . launch_cmd ) )
    self . _scaling_enabled = self . provider . scaling_enabled
    logger . debug ( "Starting LowLatencyExecutor with provider:\n%s" , self . provider )
    if hasattr ( self . provider , 'init_blocks' ) :
        try :
            for i in range ( self . provider . init_blocks ) :
                block = self . provider . submit ( self . launch_cmd , 1 , self . workers_per_node )
                logger . debug ( "Launched block {}:{}" . format ( i , block ) )
                if not block :
                    raise ( ScalingFailed ( self . provider . label , "Attempts to provision nodes via provider has failed" ) )
                self . blocks . extend ( [ block ] )
        except Exception as e :
            logger . error ( "Scaling out failed: {}" . format ( e ) )
            raise e
else :
    self . _scaling_enabled = False
    logger . debug ( "Starting LowLatencyExecutor with no provider" )


signature:
def _start_local_queue_process ( self ) :
    """TODO : docstring"""

implementation:
comm_q = Queue ( maxsize = 10 )
self . queue_proc = Process ( target = interchange . starter , args = ( comm_q , ) , kwargs = { "client_ports" : ( self . outgoing_q . port , self . incoming_q . port ) , "worker_port" : self . worker_port , "worker_port_range" : self . worker_port_range # TODO : logdir and logging level
} )
self . queue_proc . start ( )
try :
    worker_port = comm_q . get ( block = True , timeout = 120 )
    logger . debug ( "Got worker port {} from interchange" . format ( worker_port ) )
except queue . Empty :
    logger . error ( "Interchange has not completed initialization in 120s. Aborting" )
    raise Exception ( "Interchange failed to start" )
self . worker_task_url = "tcp://{}:{}" . format ( self . address , worker_port )


signature:
def _start_queue_management_thread ( self ) :
    """TODO : docstring"""

implementation:
if self . _queue_management_thread is None :
    logger . debug ( "Starting queue management thread" )
    self . _queue_management_thread = threading . Thread ( target = self . _queue_management_worker )
    self . _queue_management_thread . daemon = True
    self . _queue_management_thread . start ( )
    logger . debug ( "Started queue management thread" )
else :
    logger . debug ( "Management thread already exists, returning" )


signature:
def _queue_management_worker ( self ) :
    """TODO : docstring"""

implementation:
logger . debug ( "[MTHREAD] queue management worker starting" )
while True :
    task_id , buf = self . incoming_q . get ( )
    # TODO : why does this hang ?
    msg = deserialize_object ( buf ) [ 0 ]
    # TODO : handle exceptions
    task_fut = self . tasks [ task_id ]
    logger . debug ( "Got response for task id {}" . format ( task_id ) )
    if "result" in msg :
        task_fut . set_result ( msg [ "result" ] )
    elif "exception" in msg : # TODO : handle exception
        pass
    elif 'exception' in msg :
        logger . warning ( "Task: {} has returned with an exception" )
        try :
            s , _ = deserialize_object ( msg [ 'exception' ] )
            exception = ValueError ( "Remote exception description: {}" . format ( s ) )
            task_fut . set_exception ( exception )
        except Exception as e : # TODO could be a proper wrapped exception ?
            task_fut . set_exception ( DeserializationError ( "Received exception, but handling also threw an exception: {}" . format ( e ) ) )
    else :
        raise BadMessage ( "Message received is neither result nor exception" )
    if not self . is_alive :
        break
logger . info ( "[MTHREAD] queue management worker finished" )


signature:
def scale_in ( self , blocks ) :
    """Scale in the number of active blocks by specified amount .
 The scale in method here is very rude . It doesn ' t give the workers
 the opportunity to finish current tasks or cleanup . This is tracked
 in issue # 530
 Raises :
 NotImplementedError"""

implementation:
to_kill = self . blocks [ : blocks ]
if self . provider :
    r = self . provider . cancel ( to_kill )
return r


signature:
def create_reg_message ( self ) :
    """Creates a registration message to identify the worker to the interchange"""

implementation:
msg = { 'parsl_v' : PARSL_VERSION , 'python_v' : "{}.{}.{}" . format ( sys . version_info . major , sys . version_info . minor , sys . version_info . micro ) , 'os' : platform . system ( ) , 'hname' : platform . node ( ) , 'dir' : os . getcwd ( ) , }
b_msg = json . dumps ( msg ) . encode ( 'utf-8' )
return b_msg


signature:
def heartbeat ( self ) :
    """Send heartbeat to the incoming task queue"""

implementation:
heartbeat = ( HEARTBEAT_CODE ) . to_bytes ( 4 , "little" )
r = self . task_incoming . send ( heartbeat )
logger . debug ( "Return from heartbeat : {}" . format ( r ) )


signature:
def recv_result_from_workers ( self ) :
    """Receives a results from the MPI worker pool and send it out via 0mq
 Returns :
 result : task result from the workers"""

implementation:
info = MPI . Status ( )
result = self . comm . recv ( source = MPI . ANY_SOURCE , tag = RESULT_TAG , status = info )
logger . debug ( "Received result from workers: {}" . format ( result ) )
return result


signature:
def recv_task_request_from_workers ( self ) :
    """Receives 1 task request from MPI comm
 Returns :
 worker _ rank : worker _ rank id"""

implementation:
info = MPI . Status ( )
comm . recv ( source = MPI . ANY_SOURCE , tag = TASK_REQUEST_TAG , status = info )
worker_rank = info . Get_source ( )
logger . info ( "Received task request from worker:{}" . format ( worker_rank ) )
return worker_rank


signature:
def pull_tasks ( self , kill_event ) :
    """Pulls tasks from the incoming tasks 0mq pipe onto the internal
 pending task queue
 Parameters :
 kill _ event : threading . Event
 Event to let the thread know when it is time to die ."""

implementation:
logger . info ( "[TASK PULL THREAD] starting" )
poller = zmq . Poller ( )
poller . register ( self . task_incoming , zmq . POLLIN )
# Send a registration message
msg = self . create_reg_message ( )
logger . debug ( "Sending registration message: {}" . format ( msg ) )
self . task_incoming . send ( msg )
last_beat = time . time ( )
last_interchange_contact = time . time ( )
task_recv_counter = 0
poll_timer = 1
while not kill_event . is_set ( ) :
    time . sleep ( LOOP_SLOWDOWN )
    ready_worker_count = self . ready_worker_queue . qsize ( )
    pending_task_count = self . pending_task_queue . qsize ( )
    logger . debug ( "[TASK_PULL_THREAD] ready workers:{}, pending tasks:{}" . format ( ready_worker_count , pending_task_count ) )
    if time . time ( ) > last_beat + self . heartbeat_period :
        self . heartbeat ( )
        last_beat = time . time ( )
    if pending_task_count < self . max_queue_size and ready_worker_count > 0 :
        logger . debug ( "[TASK_PULL_THREAD] Requesting tasks: {}" . format ( ready_worker_count ) )
        msg = ( ( ready_worker_count ) . to_bytes ( 4 , "little" ) )
        self . task_incoming . send ( msg )
    socks = dict ( poller . poll ( timeout = poll_timer ) )
    if self . task_incoming in socks and socks [ self . task_incoming ] == zmq . POLLIN :
        _ , pkl_msg = self . task_incoming . recv_multipart ( )
        tasks = pickle . loads ( pkl_msg )
        last_interchange_contact = time . time ( )
        if tasks == 'STOP' :
            logger . critical ( "[TASK_PULL_THREAD] Received stop request" )
            kill_event . set ( )
            break
        elif tasks == HEARTBEAT_CODE :
            logger . debug ( "Got heartbeat from interchange" )
        else : # Reset timer on receiving message
            poll_timer = 1
            task_recv_counter += len ( tasks )
            logger . debug ( "[TASK_PULL_THREAD] Got tasks: {} of {}" . format ( [ t [ 'task_id' ] for t in tasks ] , task_recv_counter ) )
            for task in tasks :
                self . pending_task_queue . put ( task )
    else :
        logger . debug ( "[TASK_PULL_THREAD] No incoming tasks" )
        # Limit poll duration to heartbeat _ period
        # heartbeat _ period is in s vs poll _ timer in ms
        poll_timer = min ( self . heartbeat_period * 1000 , poll_timer * 2 )
        # Only check if no messages were received .
        if time . time ( ) > last_interchange_contact + self . heartbeat_threshold :
            logger . critical ( "[TASK_PULL_THREAD] Missing contact with interchange beyond heartbeat_threshold" )
            kill_event . set ( )
            logger . critical ( "[TASK_PULL_THREAD] Exiting" )
            break


signature:
def push_results ( self , kill_event ) :
    """Listens on the pending _ result _ queue and sends out results via 0mq
 Parameters :
 kill _ event : threading . Event
 Event to let the thread know when it is time to die ."""

implementation:
# We set this timeout so that the thread checks the kill _ event and does not
# block forever on the internal result queue
timeout = 0.1
# timer = time . time ( )
logger . debug ( "[RESULT_PUSH_THREAD] Starting thread" )
while not kill_event . is_set ( ) :
    time . sleep ( LOOP_SLOWDOWN )
    try :
        items = [ ]
        while not self . pending_result_queue . empty ( ) :
            r = self . pending_result_queue . get ( block = True )
            items . append ( r )
        if items :
            self . result_outgoing . send_multipart ( items )
    except queue . Empty :
        logger . debug ( "[RESULT_PUSH_THREAD] No results to send in past {}seconds" . format ( timeout ) )
    except Exception as e :
        logger . exception ( "[RESULT_PUSH_THREAD] Got an exception : {}" . format ( e ) )
logger . critical ( "[RESULT_PUSH_THREAD] Exiting" )


signature:
def start ( self ) :
    """Start the Manager process .
 The worker loops on this :
 1 . If the last message sent was older than heartbeat period we send a heartbeat
 TODO : Move task receiving to a thread"""

implementation:
self . comm . Barrier ( )
logger . debug ( "Manager synced with workers" )
self . _kill_event = threading . Event ( )
self . _task_puller_thread = threading . Thread ( target = self . pull_tasks , args = ( self . _kill_event , ) )
self . _result_pusher_thread = threading . Thread ( target = self . push_results , args = ( self . _kill_event , ) )
self . _task_puller_thread . start ( )
self . _result_pusher_thread . start ( )
start = None
result_counter = 0
task_recv_counter = 0
task_sent_counter = 0
logger . info ( "Loop start" )
while not self . _kill_event . is_set ( ) :
    time . sleep ( LOOP_SLOWDOWN )
    # In this block we attempt to probe MPI for a set amount of time ,
    # and if we have exhausted all available MPI events , we move on
    # to the next block . The timer and counter trigger balance
    # fairness and responsiveness .
    timer = time . time ( ) + 0.05
    counter = min ( 10 , comm . size )
    while time . time ( ) < timer :
        info = MPI . Status ( )
        if counter > 10 :
            logger . debug ( "Hit max mpi events per round" )
            break
        if not self . comm . Iprobe ( status = info ) :
            logger . debug ( "Timer expired, processed {} mpi events" . format ( counter ) )
            break
        else :
            tag = info . Get_tag ( )
            logger . info ( "Message with tag {} received" . format ( tag ) )
            counter += 1
            if tag == RESULT_TAG :
                result = self . recv_result_from_workers ( )
                self . pending_result_queue . put ( result )
                result_counter += 1
            elif tag == TASK_REQUEST_TAG :
                worker_rank = self . recv_task_request_from_workers ( )
                self . ready_worker_queue . put ( worker_rank )
            else :
                logger . error ( "Unknown tag {} - ignoring this message and continuing" . format ( tag ) )
    available_worker_cnt = self . ready_worker_queue . qsize ( )
    available_task_cnt = self . pending_task_queue . qsize ( )
    logger . debug ( "[MAIN] Ready workers: {} Ready tasks: {}" . format ( available_worker_cnt , available_task_cnt ) )
    this_round = min ( available_worker_cnt , available_task_cnt )
    for i in range ( this_round ) :
        worker_rank = self . ready_worker_queue . get ( )
        task = self . pending_task_queue . get ( )
        comm . send ( task , dest = worker_rank , tag = worker_rank )
        task_sent_counter += 1
        logger . debug ( "Assigning worker:{} task:{}" . format ( worker_rank , task [ 'task_id' ] ) )
    if not start :
        start = time . time ( )
    logger . debug ( "Tasks recvd:{} Tasks dispatched:{} Results recvd:{}" . format ( task_recv_counter , task_sent_counter , result_counter ) )
    # print ( " [ { } ] Received : { } " . format ( self . identity , msg ) )
    # time . sleep ( random . randint ( 4,10 ) / 10)
self . _task_puller_thread . join ( )
self . _result_pusher_thread . join ( )
self . task_incoming . close ( )
self . result_outgoing . close ( )
self . context . term ( )
delta = time . time ( ) - start
logger . info ( "mpi_worker_pool ran for {} seconds" . format ( delta ) )


signature:
def async_process ( fn ) :
    """Decorator function to launch a function as a separate process"""

implementation:
def run ( * args , ** kwargs ) :
    proc = mp . Process ( target = fn , args = args , kwargs = kwargs )
    proc . start ( )
    return proc
return run


signature:
def udp_messenger ( domain_name , UDP_IP , UDP_PORT , sock_timeout , message ) :
    """Send UDP messages to usage tracker asynchronously
 This multiprocessing based messenger was written to overcome the limitations
 of signalling / terminating a thread that is blocked on a system call . This
 messenger is created as a separate process , and initialized with 2 queues ,
 to _ send to receive messages to be sent to the internet .
 Args :
 - domain _ name ( str ) : Domain name string
 - UDP _ IP ( str ) : IP address YYY . YYY . YYY . YYY
 - UDP _ PORT ( int ) : UDP port to send out on
 - sock _ timeout ( int ) : Socket timeout
 - to _ send ( multiprocessing . Queue ) : Queue of outgoing messages to internet"""

implementation:
try :
    if message is None :
        raise ValueError ( "message was none" )
    encoded_message = bytes ( message , "utf-8" )
    if encoded_message is None :
        raise ValueError ( "utf-8 encoding of message failed" )
    if domain_name :
        try :
            UDP_IP = socket . gethostbyname ( domain_name )
        except Exception : # ( False , " Domain lookup failed , defaulting to { 0 } " . format ( UDP _ IP ) )
            pass
    if UDP_IP is None :
        raise Exception ( "UDP_IP is None" )
    if UDP_PORT is None :
        raise Exception ( "UDP_PORT is None" )
    sock = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM )
    # UDP
    sock . settimeout ( sock_timeout )
    sock . sendto ( bytes ( message , "utf-8" ) , ( UDP_IP , UDP_PORT ) )
    sock . close ( )
except socket . timeout :
    logger . debug ( "Failed to send usage tracking data: socket timeout" )
except OSError as e :
    logger . debug ( "Failed to send usage tracking data: OSError: {}" . format ( e ) )
except Exception as e :
    logger . debug ( "Failed to send usage tracking data: Exception: {}" . format ( e ) )


signature:
def check_tracking_enabled ( self ) :
    """By default tracking is enabled .
 If Test mode is set via env variable PARSL _ TESTING , a test flag is set
 Tracking is disabled if :
 1 . config [ " globals " ] [ " usageTracking " ] is set to False ( Bool )
 2 . Environment variable PARSL _ TRACKING is set to false ( case insensitive )"""

implementation:
track = True
# By default we track usage
test = False
# By default we are not in testing mode
testvar = str ( os . environ . get ( "PARSL_TESTING" , 'None' ) ) . lower ( )
if testvar == 'true' :
    test = True
if not self . config . usage_tracking :
    track = False
envvar = str ( os . environ . get ( "PARSL_TRACKING" , True ) ) . lower ( )
if envvar == "false" :
    track = False
return test , track


signature:
def construct_start_message ( self ) :
    """Collect preliminary run info at the start of the DFK .
 Returns :
 - Message dict dumped as json string , ready for UDP"""

implementation:
uname = getpass . getuser ( ) . encode ( 'latin1' )
hashed_username = hashlib . sha256 ( uname ) . hexdigest ( ) [ 0 : 10 ]
hname = socket . gethostname ( ) . encode ( 'latin1' )
hashed_hostname = hashlib . sha256 ( hname ) . hexdigest ( ) [ 0 : 10 ]
message = { 'uuid' : self . uuid , 'uname' : hashed_username , 'hname' : hashed_hostname , 'test' : self . test_mode , 'parsl_v' : self . parsl_version , 'python_v' : self . python_version , 'os' : platform . system ( ) , 'os_v' : platform . release ( ) , 'start' : time . time ( ) }
return json . dumps ( message )


signature:
def construct_end_message ( self ) :
    """Collect the final run information at the time of DFK cleanup .
 Returns :
 - Message dict dumped as json string , ready for UDP"""

implementation:
app_count = self . dfk . task_count
site_count = len ( [ x for x in self . dfk . config . executors if x . managed ] )
app_fails = len ( [ t for t in self . dfk . tasks if self . dfk . tasks [ t ] [ 'status' ] in FINAL_FAILURE_STATES ] )
message = { 'uuid' : self . uuid , 'end' : time . time ( ) , 't_apps' : app_count , 'sites' : site_count , 'c_time' : None , 'failed' : app_fails , 'test' : self . test_mode , }
return json . dumps ( message )


signature:
def send_UDP_message ( self , message ) :
    """Send UDP message ."""

implementation:
x = 0
if self . tracking_enabled :
    try :
        proc = udp_messenger ( self . domain_name , self . UDP_IP , self . UDP_PORT , self . sock_timeout , message )
        self . procs . append ( proc )
    except Exception as e :
        logger . debug ( "Usage tracking failed: {}" . format ( e ) )
else :
    x = - 1
return x


signature:
def send_message ( self ) :
    """Send message over UDP .
 If tracking is disables , the bytes _ sent will always be set to - 1
 Returns :
 ( bytes _ sent , time _ taken )"""

implementation:
start = time . time ( )
message = None
if not self . initialized :
    message = self . construct_start_message ( )
    self . initialized = True
else :
    message = self . construct_end_message ( )
self . send_UDP_message ( message )
end = time . time ( )
return end - start


signature:
def set_file_logger ( filename : str , name : str = 'parsl' , level : int = logging . DEBUG , format_string : Optional [ str ] = None ) :
    """Add a stream log handler .
 Args :
 - filename ( string ) : Name of the file to write logs to
 - name ( string ) : Logger name
 - level ( logging . LEVEL ) : Set the logging level .
 - format _ string ( string ) : Set the format string
 Returns :
 - None"""

implementation:
if format_string is None :
    format_string = "%(asctime)s.%(msecs)03d %(name)s:%(lineno)d [%(levelname)s]  %(message)s"
logger = logging . getLogger ( name )
logger . setLevel ( logging . DEBUG )
handler = logging . FileHandler ( filename )
handler . setLevel ( level )
formatter = logging . Formatter ( format_string , datefmt = '%Y-%m-%d %H:%M:%S' )
handler . setFormatter ( formatter )
logger . addHandler ( handler )
# see note in set _ stream _ logger for notes about logging
# concurrent . futures
futures_logger = logging . getLogger ( "concurrent.futures" )
futures_logger . addHandler ( handler )


signature:
def start_file_logger ( filename , name = 'database_manager' , level = logging . DEBUG , format_string = None ) :
    """Add a stream log handler .
 Parameters
 filename : string
 Name of the file to write logs to . Required .
 name : string
 Logger name . Default = " parsl . executors . interchange "
 level : logging . LEVEL
 Set the logging level . Default = logging . DEBUG
 - format _ string ( string ) : Set the format string
 format _ string : string
 Format string to use .
 Returns
 None ."""

implementation:
if format_string is None :
    format_string = "%(asctime)s %(name)s:%(lineno)d [%(levelname)s]  %(message)s"
global logger
logger = logging . getLogger ( name )
logger . setLevel ( level )
handler = logging . FileHandler ( filename )
handler . setLevel ( level )
formatter = logging . Formatter ( format_string , datefmt = '%Y-%m-%d %H:%M:%S' )
handler . setFormatter ( formatter )
logger . addHandler ( handler )
return logger


signature:
def dbm_starter ( priority_msgs , resource_msgs , * args , ** kwargs ) :
    """Start the database manager process
 The DFK should start this function . The args , kwargs match that of the monitoring config"""

implementation:
dbm = DatabaseManager ( * args , ** kwargs )
dbm . start ( priority_msgs , resource_msgs )


signature:
def _create_task_log_info ( self , task_id , fail_mode = None ) :
    """Create the dictionary that will be included in the log ."""

implementation:
info_to_monitor = [ 'func_name' , 'fn_hash' , 'memoize' , 'checkpoint' , 'fail_count' , 'fail_history' , 'status' , 'id' , 'time_submitted' , 'time_returned' , 'executor' ]
task_log_info = { "task_" + k : self . tasks [ task_id ] [ k ] for k in info_to_monitor }
task_log_info [ 'run_id' ] = self . run_id
task_log_info [ 'timestamp' ] = datetime . datetime . now ( )
task_log_info [ 'task_status_name' ] = self . tasks [ task_id ] [ 'status' ] . name
task_log_info [ 'tasks_failed_count' ] = self . tasks_failed_count
task_log_info [ 'tasks_completed_count' ] = self . tasks_completed_count
task_log_info [ 'task_inputs' ] = str ( self . tasks [ task_id ] [ 'kwargs' ] . get ( 'inputs' , None ) )
task_log_info [ 'task_outputs' ] = str ( self . tasks [ task_id ] [ 'kwargs' ] . get ( 'outputs' , None ) )
task_log_info [ 'task_stdin' ] = self . tasks [ task_id ] [ 'kwargs' ] . get ( 'stdin' , None )
task_log_info [ 'task_stdout' ] = self . tasks [ task_id ] [ 'kwargs' ] . get ( 'stdout' , None )
task_log_info [ 'task_depends' ] = None
if self . tasks [ task_id ] [ 'depends' ] is not None :
    task_log_info [ 'task_depends' ] = "," . join ( [ str ( t . _tid ) for t in self . tasks [ task_id ] [ 'depends' ] ] )
task_log_info [ 'task_elapsed_time' ] = None
if self . tasks [ task_id ] [ 'time_returned' ] is not None :
    task_log_info [ 'task_elapsed_time' ] = ( self . tasks [ task_id ] [ 'time_returned' ] - self . tasks [ task_id ] [ 'time_submitted' ] ) . total_seconds ( )
if fail_mode is not None :
    task_log_info [ 'task_fail_mode' ] = fail_mode
return task_log_info


signature:
def _count_deps ( self , depends ) :
    """Internal .
 Count the number of unresolved futures in the list depends ."""

implementation:
count = 0
for dep in depends :
    if isinstance ( dep , Future ) :
        if not dep . done ( ) :
            count += 1
return count


signature:
def handle_exec_update ( self , task_id , future ) :
    """This function is called only as a callback from an execution
 attempt reaching a final state ( either successfully or failing ) .
 It will launch retries if necessary , and update the task
 structure .
 Args :
 task _ id ( string ) : Task id which is a uuid string
 future ( Future ) : The future object corresponding to the task which
 makes this callback
 KWargs :
 memo _ cbk ( Bool ) : Indicates that the call is coming from a memo update ,
 that does not require additional memo updates ."""

implementation:
try :
    res = future . result ( )
    if isinstance ( res , RemoteExceptionWrapper ) :
        res . reraise ( )
except Exception :
    logger . exception ( "Task {} failed" . format ( task_id ) )
    # We keep the history separately , since the future itself could be
    # tossed .
    self . tasks [ task_id ] [ 'fail_history' ] . append ( future . _exception )
    self . tasks [ task_id ] [ 'fail_count' ] += 1
    if not self . _config . lazy_errors :
        logger . debug ( "Eager fail, skipping retry logic" )
        self . tasks [ task_id ] [ 'status' ] = States . failed
        if self . monitoring :
            task_log_info = self . _create_task_log_info ( task_id , 'eager' )
            self . monitoring . send ( MessageType . TASK_INFO , task_log_info )
        return
    if self . tasks [ task_id ] [ 'fail_count' ] <= self . _config . retries :
        self . tasks [ task_id ] [ 'status' ] = States . pending
        logger . debug ( "Task {} marked for retry" . format ( task_id ) )
    else :
        logger . info ( "Task {} failed after {} retry attempts" . format ( task_id , self . _config . retries ) )
        self . tasks [ task_id ] [ 'status' ] = States . failed
        self . tasks_failed_count += 1
        self . tasks [ task_id ] [ 'time_returned' ] = datetime . datetime . now ( )
else :
    self . tasks [ task_id ] [ 'status' ] = States . done
    self . tasks_completed_count += 1
    logger . info ( "Task {} completed" . format ( task_id ) )
    self . tasks [ task_id ] [ 'time_returned' ] = datetime . datetime . now ( )
if self . monitoring :
    task_log_info = self . _create_task_log_info ( task_id , 'lazy' )
    self . monitoring . send ( MessageType . TASK_INFO , task_log_info )
# it might be that in the course of the update , we ' ve gone back to being
# pending - in which case , we should consider ourself for relaunch
if self . tasks [ task_id ] [ 'status' ] == States . pending :
    self . launch_if_ready ( task_id )
return


signature:
def handle_app_update ( self , task_id , future , memo_cbk = False ) :
    """This function is called as a callback when an AppFuture
 is in its final state .
 It will trigger post - app processing such as checkpointing
 and stageout .
 Args :
 task _ id ( string ) : Task id
 future ( Future ) : The relevant app future ( which should be
 consistent with the task structure ' app _ fu ' entry
 KWargs :
 memo _ cbk ( Bool ) : Indicates that the call is coming from a memo update ,
 that does not require additional memo updates ."""

implementation:
if not self . tasks [ task_id ] [ 'app_fu' ] . done ( ) :
    logger . error ( "Internal consistency error: app_fu is not done for task {}" . format ( task_id ) )
if not self . tasks [ task_id ] [ 'app_fu' ] == future :
    logger . error ( "Internal consistency error: callback future is not the app_fu in task structure, for task {}" . format ( task_id ) )
if not memo_cbk : # Update the memoizer with the new result if this is not a
# result from a memo lookup and the task has reached a terminal state .
    self . memoizer . update_memo ( task_id , self . tasks [ task_id ] , future )
    if self . checkpoint_mode == 'task_exit' :
        self . checkpoint ( tasks = [ task_id ] )
# Submit _ * _ stage _ out tasks for output data futures that correspond with remote files
if ( self . tasks [ task_id ] [ 'app_fu' ] and self . tasks [ task_id ] [ 'app_fu' ] . done ( ) and self . tasks [ task_id ] [ 'app_fu' ] . exception ( ) is None and self . tasks [ task_id ] [ 'executor' ] != 'data_manager' and self . tasks [ task_id ] [ 'func_name' ] != '_ftp_stage_in' and self . tasks [ task_id ] [ 'func_name' ] != '_http_stage_in' ) :
    for dfu in self . tasks [ task_id ] [ 'app_fu' ] . outputs :
        f = dfu . file_obj
        if isinstance ( f , File ) and f . is_remote ( ) :
            self . data_manager . stage_out ( f , self . tasks [ task_id ] [ 'executor' ] )
return