CEDARS Administrator Manual

def query_to_patterns(query: str) -> list:
    """
    Expected query will be a set of keywords separated
    by OR keyword.

    Each expression separated by OR can have expressions
    combined by AND or NOT and the keywords can also contain
    wildcards.

    ##### Spacy Requirements:
    - ! - negation
    - Each dictionary in a list matches one token only
    - A list matches all the dictionaries inside it (and condition)
    - A list of list contains OR conditions
      - [{"TEXT": {"REGEX": "abc*"}}] represents one token with regex match
      - [{"LOWER": "dvt"}] matches case-insenstitive  DVT
      - [{"LEMMA": "embolus"}] matches the lemmatized version of embolus as well in text

    ##### Implementation:
      1. Split the query by OR
      2. Split each expression by AND
      3. Split each expression by NOT
      4. Split each expression by wildcard
      5. Convert each expression to a spacy pattern
      6. Combine the patterns
      7. Return the combined pattern
    """

    or_expressions = query.split(" OR ")
    res = [[] for _ in range(len(or_expressions))]
    for i, expression in enumerate(or_expressions):
        spacy_pattern = []
        expression = expression.strip().replace("(", "").replace(")", "")
        and_expressions = expression.split(" AND ")
        for tok in and_expressions:
            tok = tok.strip()
            if not tok:
                continue
            if "*" in tok or "?" in tok:
                spacy_pattern.append(get_regex_dict(tok))
            elif "!" in tok:
                spacy_pattern.append(get_negated_dict(tok.replace("!", "")))
            else:
                spacy_pattern.append(get_lemma_dict(tok))
        # logger.debug(f"{expression} -> {spacy_pattern}")
        res[i] = spacy_pattern
    return res

def is_negated(span):
    """
    ##### Negation Detection

    This function takes a spacy token and determines if it has been negated in the sentence.
    ```
    Ex.
    This is not an apple.
    In the above sentence, the token apple is negated.
    ```

    Args:
        spacy token : This is a token of a single word after spacy
        runs a model on some text.

    Returns:
        (bool) : True if the token is negated in the sentence.
    """
    neg_words = ['no', 'not', "n't", "wouldn't", 'never', 'nobody', 'nothing',
                 'neither', 'nowhere', 'noone', 'no-one', 'hardly', 'scarcely', 'barely']

    for token in span.subtree:
        parents = list(token.ancestors)
        children = list(token.children)

        for parent in token.ancestors:
            children.extend(list(parent.children))

        if ("neg" in [child.dep_ for child in children]) or ("neg" in [par.dep_ for par in parents]):
            return True

        parents_text = [par.text for par in parents]
        children_text = [child.text for child in children]

        for word in neg_words:
            if word in parents_text or word in children_text:
                return True

    return False

def process_notes(self, patient_id, processes=1, batch_size=20):
    """
    ##### Process Query Matching

    This function takes a medical note and a regex query as input and annotates
    the relevant sections of the text.
    """
    # nlp_model = spacy.load(model_name)
    matcher = Matcher(self.nlp_model.vocab)
    assert len(matcher) == 0
    query = db.get_search_query()

    # load previosly processed documents
    # document_processed = load_progress()
    spacy_patterns = query_to_patterns(query)
    for i, item in enumerate(spacy_patterns):
        matcher.add(f"DVT_{i}", [item])

    # check all documents already processed
    documents_to_process = []
    if patient_id is not None:
        # get all note for patient which are not reviewed
        documents_to_process = db.get_documents_to_annotate(patient_id)
    else:
        # get all notes which are not in annotation collection.
        documents_to_process = db.get_documents_to_annotate()

    document_list = [document for document in documents_to_process]
    if len(document_list) == 0:
        # no notes found to annotate
        logger.info(f"No documents to process for patient {patient_id}")
        if db.get_search_query("tag_query")["nlp_apply"] is True:
            self.process_patient_pines(patient_id)
        return

    document_text = [document["text"] for document in document_list]
    if patient_id is not None:
        logger.info(f"Found {len(document_list)}/{db.get_total_counts('NOTES', patient_id=patient_id)} to process")
    else:
        logger.info(f"Found {len(document_list)}/{db.get_total_counts('NOTES')} documents to process")

    # logger.info(f"sample document: {document_text[0][:100]}")
    annotations = self.nlp_model.pipe([document["text"].lower() for document in document_list],
                                      n_process=processes,
                                      batch_size=batch_size)
    logger.info(f"Starting to process document annotations: {len(document_text)}")
    count = 0
    docs_with_annotations = 0
    for document, doc in zip(document_list, annotations):
        docs_with_annotations += 1
        match_count = 0
        for sent_no, sentence_annotation in enumerate(doc.sents):
            sentence_text = sentence_annotation.text
            matches = matcher(sentence_annotation)
            for match in matches:
                match_count += 1
                _, start, end = match
                token = sentence_annotation[start:end]
                has_negation = is_negated(token)
                start_index = sentence_text.find(token.text, start)
                end_index = start_index + len(token.text)
                token_start = token.start_char
                token_end = token_start + len(token.text)
                annotation = {
                                "sentence": sentence_text,
                                "token": token.text,
                                "isNegated": has_negation,
                                "start_index": start_index,
                                "end_index": end_index,
                                "note_start_index": token_start,
                                "note_end_index": token_end,
                                "sentence_number": sent_no
                                }
                annotation['note_id'] = document["text_id"]
                annotation["text_date"] = document["text_date"]
                annotation["patient_id"] = document["patient_id"]
                annotation["event_date"] = None
                annotation["reviewed"] = False
                db.insert_one_annotation(annotation)
        if match_count == 0:
            db.mark_note_reviewed(document["text_id"], reviewed_by="CEDARS")
        count += 1
        if (count) % 10 == 0:
            logger.info(f"Processed {count} / {len(document_list)} documents")

    # check if nlp processing is enabled
    if docs_with_annotations > 0 and db.get_search_query("tag_query")["nlp_apply"] is True:
        logger.info(f"Processing {docs_with_annotations} documents with PINES")
        self.process_patient_pines(patient_id)

def get_prediction(note: str) -> float:
    """
    ##### PINES predictions

    Get prediction from endpoint. Text goes in the POST request.
    """
    url = f'{os.getenv("PINES_API_URL")}/predict'
    data = {'text': note}
    log_notes = None
    try:
        response = requests.post(url, json=data, timeout=3600)
        response.raise_for_status()
        res = response.json()["prediction"]
        score = res.get("score")
        label = res.get("label")
        if isinstance(label, str):
            score = 1 - score if "0" in label else score
        else:
            score = 1 - score if label == 0 else score
        log_notes = re.sub(r'\d', '*', note[:20])
        logger.debug(f"Got prediction for note: {log_notes} with score: {score} and label: {label}")
        return score
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to get prediction for note: {log_notes}")
        raise e

def predict_and_save(text_ids: Optional[list[str]] = None,
                     note_collection_name: str = "NOTES",
                     pines_collection_name: str = "PINES",
                     force_update: bool = False) -> None:
    """
    ##### Save PINES predictions

    Predict and save the predictions for the given text_ids.
    """
    notes_collection = mongo.db[note_collection_name]
    pines_collection = mongo.db[pines_collection_name]
    query = {}
    if text_ids is not None:
        query = {"text_id": {"$in": text_ids}}

    cedars_notes = notes_collection.find(query)
    count = 0
    for note in cedars_notes:
        note_id = note.get("text_id")
        if force_update or get_note_prediction_from_db(note_id, pines_collection_name) is None:
            logger.info(f"Predicting for note: {note_id}")
            prediction = get_prediction(note.get("text"))
            pines_collection.insert_one({
                "text_id": note_id,
                "text": note.get("text"),
                "patient_id": note.get("patient_id"),
                "predicted_score": prediction,
                "report_type": note.get("text_tag_3"),
                "document_type": note.get("text_tag_1")
                })
        count += 1

@bp.route('/download_annotations')
@auth.admin_required
def download_file(filename='annotations.csv'):
    """
    ##### Download Completed Annotations

    This generates a CSV file with the following specifications:
    1. Find all patients in the PATIENTS database, these patients become a single row in the CSV file.
    2. For each patient -
        a. list the number of total notes in the database
        b. list the number of reviewed notes
        c. list the number of total sentences from annotations
        d. list the number of reviewed sentences
        e. list all sentences as a list of strings
        f. add event date from the annotations for each patient
        g. add the first and last note date for each patient
    3. Convert all columns to proper datatypes
    """
    logger.info("Downloading annotations")
    if db.download_annotations(filename):
        file = minio.get_object(g.bucket_name, f"annotated_files/{filename}")
        logger.info(f"Downloaded annotations from s3: {filename}")
        return flask.Response(
            file.stream(32*1024),
            mimetype='text/csv',
            headers={"Content-Disposition": "attachment;filename=cedars_annotations.csv"}
        )
    else:
        flask.jsonify({"error": f"Annotations with filename '{filename}' not found."}), 404

def terminate_project():
    """
    ##### Terminate the Project

    Reset the database to the initial state.
    """
    logger.info("Terminating project.")
    # Delete all mongo DB collections
    mongo.db.drop_collection("ANNOTATIONS")
    mongo.db.drop_collection("NOTES")
    mongo.db.drop_collection("PATIENTS")
    mongo.db.drop_collection("USERS")
    mongo.db.drop_collection("QUERY")
    mongo.db.drop_collection("PINES")
    mongo.db.drop_collection("TASK")

    # reset all rq queues
    flask.current_app.task_queue.empty()

    create_project(project_name=fake.slug(),
                   investigator_name=fake.name())