def filter_and_process_statements(
    sts,
    grounding_score_cutoff: float = 0.8,
    belief_score_cutoff: float = 0.85,
    concepts_of_interest: List[str] = [],
):

    """ Filter preassembled statements according to certain rules. """
    filtered_sts = []
    counters = {}

    def update_counter(counter_name):
        if counter_name in counters:
            counters[counter_name] += 1
        else:
            counters[counter_name] = 1

    for s in tqdm(sts):

        update_counter("Original number of statements")

        # Apply belief score threshold cutoff

        if not s.belief > belief_score_cutoff:
            continue
        update_counter(f"Statements with belief score > {belief_score_cutoff}")

        # Select statements with UN groundings

        if s.subj.db_refs.get("UN") is None or s.obj.db_refs.get("UN") is None:
            continue
        update_counter("Statements with UN groundings")

        # Apply grounding score cutoffs

        if not all(
            x[1] > grounding_score_cutoff
            for x in (y.db_refs["UN"][0] for y in (s.subj, s.obj))
        ):
            continue
        update_counter(
            f"Statements with subj and obj grounding scores > {grounding_score_cutoff}"
        )

        # Assign default polarities

        if s.subj_delta["polarity"] is None:
            s.subj_delta["polarity"] = 1
        if s.obj_delta["polarity"] is None:
            s.obj_delta["polarity"] = 1

        filtered_sts.append(s)

    for k, v in counters.items():
        print(f"{k}: {v}")

    return filtered_sts
def filter_and_process_statements ( sts , grounding_score_cutoff : float = 0.8 , belief_score_cutoff : float = 0.85 , concepts_of_interest : List [ str ] = [ ] , ) : filtered_sts = [ ] counters = { } def update_counter ( counter_name ) : if counter_name in counters : counters [ counter_name ] += 1 else : counters [ counter_name ] = 1 for s in tqdm ( sts ) : update_counter ( "Original number of statements" ) if not s . belief > belief_score_cutoff : continue update_counter ( f"Statements with belief score > {belief_score_cutoff}" ) if s . subj . db_refs . get ( "UN" ) is None or s . obj . db_refs . get ( "UN" ) is None : continue update_counter ( "Statements with UN groundings" ) if not all ( x [ 1 ] > grounding_score_cutoff for x in ( y . db_refs [ "UN" ] [ 0 ] for y in ( s . subj , s . obj ) ) ) : continue update_counter ( f"Statements with subj and obj grounding scores > {grounding_score_cutoff}" ) if s . subj_delta [ "polarity" ] is None : s . subj_delta [ "polarity" ] = 1 if s . obj_delta [ "polarity" ] is None : s . obj_delta [ "polarity" ] = 1 filtered_sts . append ( s ) for k , v in counters . items ( ) : print ( f"{k}: {v}" ) return filtered_sts
Filter preassembled statements according to certain rules.