medcat.components.addons.relation_extraction.rel_dataset

Classes:

• RelData –

Attributes:

• logger –

logger module-attribute

logger = getLogger(__name__)

RelData

RelData(tokenizer: BaseTokenizerWrapper, config: ConfigRelCAT, cdb: CDB = CDB(Config()))

Bases: Dataset

Use this class to create a dataset for relation annotations from CSV exports, MedCAT exports or Spacy Documents (assuming the documents got generated by MedCAT, if they did not then please set the required parameters manually to match MedCAT output, see /medcat/cat.py#_get_entity)

If you are using this to create relations from CSV it is assumed that your entities/concepts of interest are surrounded by the special tokens, see create_base_relations_from_csv doc.

Parameters:

• tokenizer

  (BaseTokenizerWrapper) –

  tokenizer used to generate token ids from input text

• config

  (ConfigRelCAT) –

  same config used in RelCAT

• cdb

  (CDB, default: CDB(Config()) ) –

  Optional, used to add concept ids and types to detected ents, useful when creating datasets from MedCAT output. Defaults to CDB().

Methods:

• create_base_relations_from_csv –

  Assumes the columns are as follows
  

• create_base_relations_from_doc –

  Creates a list of tuples based on pairs of entities detected

• create_relations_from_export –

  Args:

• generate_base_relations –

  Util function, should be used if you want to train from spacy docs

• get_labels –

  This is used to update labels in config with unencountered

Attributes:

• cdb (CDB) –
• config (ConfigRelCAT) –
• dataset (dict[Any, Any]) –
• name –
• tokenizer (BaseTokenizerWrapper) –

Source code in medcat-v2/medcat/components/addons/relation_extraction/rel_dataset.py

def __init__(self, tokenizer: BaseTokenizerWrapper,
             config: ConfigRelCAT, cdb: CDB = CDB(CoreConfig())):
    """Use this class to create a dataset for relation annotations from
    CSV exports, MedCAT exports or Spacy Documents (assuming the documents
    got generated by MedCAT, if they did not then please set the required
    parameters manually to match MedCAT output, see
    /medcat/cat.py#_get_entity)

    If you are using this to create relations from CSV it is assumed that
    your entities/concepts of interest are surrounded by the special
    tokens, see create_base_relations_from_csv doc.

    Args:
        tokenizer (BaseTokenizerWrapper):
            tokenizer used to generate token ids from input text
        config (ConfigRelCAT): same config used in RelCAT
        cdb (CDB): Optional, used to add concept ids and types to
            detected ents,  useful when creating datasets from MedCAT
            output. Defaults to CDB().
    """

    self.cdb: CDB = cdb
    self.config: ConfigRelCAT = config
    self.tokenizer: BaseTokenizerWrapper = tokenizer
    self.dataset: dict[Any, Any] = {}

    logger.setLevel(self.config.general.log_level)

cdb instance-attribute

cdb: CDB = cdb

config instance-attribute

config: ConfigRelCAT = config

dataset instance-attribute

dataset: dict[Any, Any] = {}

name class-attribute instance-attribute

name = 'rel_dataset'

tokenizer instance-attribute

tokenizer: BaseTokenizerWrapper = tokenizer

create_base_relations_from_csv

create_base_relations_from_csv(csv_path: str, keep_source_text: bool = False)

Assumes the columns are as follows
["relation_token_span_ids",
 "ent1_ent2_start", "ent1", "ent2", "label",
 "label_id", "ent1_type", "ent2_type",
 "ent1_id", "ent2_id", "ent1_cui", "ent2_cui", "doc_id", "sents"],
last column is the actual source text.

The entities inside the text MUST be annotated with special
tokens i.e:
    ...text..[s1] first ent [e1].....[s2] second ent [e2]........
You have to store the start position, aka index position of token
[e1] and also of token [e2] in the (ent1_ent2_start) column.

Parameters:

• csv_path

  (str) –

  Path to csv file, must have specific columns, tab separated.

• keep_source_text

  (bool, default: False ) –

  If the text clumn should be retained in the 'sents' df column, used for debugging or creating custom datasets.

Returns:

• dict –

  { "output_relations": relation_instances,

  NOTE: see create_base_relations_from_doc/csv

      for data columns
  

  "nclasses": self.config.model.padding_idx, # dummy class "labels2idx": {}, "idx2label": {}}

• –

  }

Source code in medcat-v2/medcat/components/addons/relation_extraction/rel_dataset.py

def create_base_relations_from_csv(self, csv_path: str,
                                   keep_source_text: bool = False):
    """
        Assumes the columns are as follows
        ["relation_token_span_ids",
         "ent1_ent2_start", "ent1", "ent2", "label",
         "label_id", "ent1_type", "ent2_type",
         "ent1_id", "ent2_id", "ent1_cui", "ent2_cui", "doc_id", "sents"],
        last column is the actual source text.

        The entities inside the text MUST be annotated with special
        tokens i.e:
            ...text..[s1] first ent [e1].....[s2] second ent [e2]........
        You have to store the start position, aka index position of token
        [e1] and also of token [e2] in the (ent1_ent2_start) column.

    Args:
        csv_path (str): Path to csv file, must have specific columns,
            tab separated.
        keep_source_text (bool): If the text clumn should be retained in
            the 'sents' df column, used for debugging or creating
            custom datasets.

    Returns:
        dict : {
            "output_relations": relation_instances,
            # NOTE: see create_base_relations_from_doc/csv
                    for data columns
            "nclasses": self.config.model.padding_idx, # dummy class
            "labels2idx": {},
            "idx2label": {}}
        }
    """

    df = pandas.read_csv(csv_path, index_col=False,
                         encoding='utf-8', sep="\t")

    tmp_col_rel_token_col = df.pop("relation_token_span_ids")

    df.insert(0, "relation_token_span_ids", tmp_col_rel_token_col)

    text_cols = ["sents", "text"]

    df["ent1_ent2_start"] = df["ent1_ent2_start"].apply(
        lambda x: literal_eval(str(x)))

    for col in text_cols:
        if col in df.columns:
            out_rels = []
            for row_idx in range(len(df[col])):
                _text = df.iloc[row_idx][col]
                _ent1_ent2_start = df.iloc[row_idx]["ent1_ent2_start"]
                _rels = self.create_base_relations_from_doc(
                    _text, doc_id=str(row_idx),
                    ent1_ent2_tokens_start_pos=_ent1_ent2_start,)
                out_rels.append(_rels)

            rows_to_remove = []
            for row_idx in range(len(out_rels)):
                if len(out_rels[row_idx]["output_relations"]) < 1:
                    rows_to_remove.append(row_idx)

            relation_token_span_ids = []
            out_ent1_ent2_starts = []

            for rel in out_rels:
                if len(rel["output_relations"]) > 0:
                    relation_token_span_ids.append(
                        rel["output_relations"][0][0])
                    out_ent1_ent2_starts.append(
                        rel["output_relations"][0][1])
                else:
                    relation_token_span_ids.append([])
                    out_ent1_ent2_starts.append([])

            df["label"] = [i.strip() for i in df["label"]]

            df["relation_token_span_ids"] = relation_token_span_ids
            df["ent1_ent2_start"] = out_ent1_ent2_starts

            df = df.drop(index=rows_to_remove)
            text_col = df.pop(col)
            df = df.assign(col=text_col)
            if keep_source_text:
                df = df.assign(col=text_col)
            break

    nclasses, labels2idx, idx2label = RelData.get_labels(
        df["label"], self.config)

    output_relations = df.values.tolist()

    logger.info(
        "CSV dataset | No. of relations detected: %d "
        "| from : %s | nclasses: %d | idx2label: %s",
        len(output_relations), csv_path, nclasses,
        str(idx2label))

    logger.info("Samples per class: ")
    for label_num in list(idx2label.keys()):
        sample_count = 0
        for output_relation in output_relations:
            if idx2label[label_num] == output_relation[4]:
                sample_count += 1
        logger.info(" label: %s | samples: %d",
                    idx2label[label_num], sample_count)

    # replace/update label_id with actual detected label number
    for idx in range(len(output_relations)):
        output_relations[idx][5] = int(
            labels2idx[output_relations[idx][4]])

    return {"output_relations": output_relations, "nclasses": nclasses,
            "labels2idx": labels2idx, "idx2label": idx2label}

create_base_relations_from_doc

create_base_relations_from_doc(doc: Union[MutableDocument, str], doc_id: str, ent1_ent2_tokens_start_pos: Union[list, tuple] = (-1, -1)) -> dict

Creates a list of tuples based on pairs of entities detected (relation, ent1, ent2) for one spacy document or text string.

Parameters:

• doc

  (Union[MutableDocument, str]) –

  SpacyDoc or string of text, each will get handled slightly differently

• doc_id

  (str) –

  Document id

• ent1_ent2_tokens_start_pos

  (Union[list, tuple], default: (-1, -1) ) –

  Start of [s1][s2] tokens, if left default we assume we are dealing with a SpacyDoc. Defaults to (-1, -1).

Returns:

• dict ( dict ) –

  {

  NOTE: see create_base_relations_from_doc/csv

  for data columns

  "output_relations": relation_instances, "nclasses": self.config.model.padding_idx, # dummy class "labels2idx": {}, "idx2label": {}}

• dict –

  }

Source code in medcat-v2/medcat/components/addons/relation_extraction/rel_dataset.py

def create_base_relations_from_doc(
        self, doc: Union[MutableDocument, str], doc_id: str,
        ent1_ent2_tokens_start_pos: Union[list, tuple] = (-1, -1)) -> dict:
    """Creates a list of tuples based on pairs of entities detected
    (relation, ent1, ent2) for one spacy document or text string.

    Args:
        doc (Union[MutableDocument, str]): SpacyDoc or string of text,
            each will get handled slightly differently
        doc_id (str): Document id
        ent1_ent2_tokens_start_pos (Union[list, tuple], optional):
            Start of [s1][s2] tokens, if left default
            we assume we are dealing with a SpacyDoc.
            Defaults to (-1, -1).

    Returns:
            dict : {
                # NOTE: see create_base_relations_from_doc/csv
                #       for data columns
                "output_relations": relation_instances,
                "nclasses": self.config.model.padding_idx,  # dummy class
                "labels2idx": {},
                "idx2label": {}}
            }
    """

    (_ent1_start_tkn_id, _ent1_end_tkn_id,
     _ent2_start_tkn_id, _ent2_end_tkn_id) = 0, 0, 0, 0

    chars_to_exclude = ":!@#$%^&*()-+?_=.,;<>/[]{}"

    if self.config.general.annotation_schema_tag_ids:
        # we assume that ent1 start token is pos 0 and ent2 start token is
        # pos 2
        # e.g: [s1], [e1], [s2], [e2]
        _ent1_start_tkn_id = (
            self.config.general.annotation_schema_tag_ids[0])
        _ent1_end_tkn_id = self.config.general.annotation_schema_tag_ids[1]
        _ent2_start_tkn_id = (
            self.config.general.annotation_schema_tag_ids[2])
        _ent2_end_tkn_id = self.config.general.annotation_schema_tag_ids[3]

    relation_instances = []

    tokenized_text_data = None

    if isinstance(doc, str):
        doc_text = doc
    else:
        doc_text = doc.base.text

    tokenized_text_data = cast(dict[str, Any],
                               self.tokenizer(doc_text, truncation=False))

    doc_length_tokens = len(tokenized_text_data["tokens"])

    if ent1_ent2_tokens_start_pos != (-1, -1) and isinstance(doc, str):
        ent1_token_start_pos = tokenized_text_data[
            "input_ids"].index(_ent1_start_tkn_id)
        ent2_token_start_pos = tokenized_text_data[
            "input_ids"].index(_ent2_start_tkn_id)
        ent1_token_end_pos = tokenized_text_data[
            "input_ids"].index(_ent1_end_tkn_id)
        ent2_token_end_pos = tokenized_text_data[
            "input_ids"].index(_ent2_end_tkn_id)

        ent1_start_char_pos, ent1_end_char_pos = tokenized_text_data[
            "offset_mapping"][ent1_token_start_pos]
        ent2_start_char_pos, ent2_end_char_pos = tokenized_text_data[
            "offset_mapping"][ent2_token_start_pos]

        relation_instances.append(self._create_relation_validation(
            text=doc_text,
            doc_id=doc_id,
            tokenized_text_data=tokenized_text_data,
            ent1_start_char_pos=ent1_start_char_pos,
            ent2_start_char_pos=ent2_start_char_pos,
            ent1_end_char_pos=ent1_end_char_pos,
            ent2_end_char_pos=ent2_end_char_pos,
            ent1_token_start_pos=ent1_token_start_pos,
            ent2_token_start_pos=ent2_token_start_pos,
            ent1_token_end_pos=ent1_token_end_pos,
            ent2_token_end_pos=ent2_token_end_pos
            ))
    elif not isinstance(doc, str):
        relation_instances.extend(
            self._create_base_relations_from_mutable_doc(
                doc, doc_text, doc_id, tokenized_text_data,
                doc_length_tokens, chars_to_exclude))

    # remove duplicates by using ent1_ent2_start_pos
    dupe_ent1_ent2_start = []

    _new_rel_instances = []
    for rel in relation_instances:
        if rel != []:
            if rel[1] not in dupe_ent1_ent2_start:
                dupe_ent1_ent2_start.append(rel[1])
                _new_rel_instances.append(rel)
            else:
                logger.debug("removing duplicate relation" + str(rel[1]))

    # cleanup
    relation_instances = _new_rel_instances

    return {
        "output_relations": relation_instances,
        "nclasses": self.config.model.padding_idx,
        "labels2idx": {}, "idx2label": {}
    }

create_relations_from_export

create_relations_from_export(data: dict)

Parameters:

• data

  (dict) –

  MedCAT Export data.

Returns:

• dict –

  {

  NOTE: see create_base_relations_from_doc/csv

      for data columns
  

  "output_relations": relation_instances, "nclasses": self.config.model.padding_idx, # dummy class "labels2idx": {}, "idx2label": {}}

• –

  }

Source code in medcat-v2/medcat/components/addons/relation_extraction/rel_dataset.py

def create_relations_from_export(self, data: dict):
    """
        Args:
            data (dict):
                MedCAT Export data.

        Returns:
            dict : {
                # NOTE: see create_base_relations_from_doc/csv
                        for data columns
                "output_relations": relation_instances,
                "nclasses": self.config.model.padding_idx,  # dummy class
                "labels2idx": {},
                "idx2label": {}}
            }
    """

    output_relations = []

    for project in data["projects"]:
        for _doc_id, document in enumerate(project["documents"]):
            relation_instances = self._create_relations_for_doc(
                document, data)
            output_relations.extend(relation_instances)

    all_relation_labels = [relation[4] for relation in output_relations]

    nclasses, labels2idx, idx2label = self.get_labels(
        all_relation_labels, self.config)

    # replace label_id with actual detected label number
    for idx in range(len(output_relations)):
        output_relations[idx][5] = int(
            labels2idx[output_relations[idx][4]])

    logger.info("MCT export dataset | nclasses: %d | idx2label: %s",
                nclasses, str(idx2label))
    logger.info("Samples per class: ")

    logger.error(str(idx2label))

    for label_num in list(idx2label.keys()):
        sample_count = 0
        for output_relation in output_relations:
            if idx2label[label_num] == output_relation[4]:
                sample_count += 1
        logger.info(
            " label: %s | samples: %s",
            idx2label[label_num], str(sample_count))

    return {"output_relations": output_relations, "nclasses": nclasses,
            "labels2idx": labels2idx, "idx2label": idx2label}

generate_base_relations

generate_base_relations(docs: Iterable[MutableDocument]) -> list[dict]

Util function, should be used if you want to train from spacy docs

Parameters:

• docs

  (Iterable[MutableDocument]) –

  Generate relations from Spacy CAT docs.

Returns:

• output_relations ( list[dict] ) –

  list[dict] : [] "output_relations": relation_instances,

  NOTE: see create_base_relations_from_doc/csv

      for data columns
  

  "nclasses": self.config.model.padding_idx # dummy class "labels2idx": {}, "idx2label": {}} ]

Source code in medcat-v2/medcat/components/addons/relation_extraction/rel_dataset.py

def generate_base_relations(self, docs: Iterable[MutableDocument]
                            ) -> list[dict]:
    """ Util function, should be used if you want to train from spacy docs

    Args:
        docs (Iterable[MutableDocument]):
            Generate relations from Spacy CAT docs.

    Returns:
        output_relations: list[dict] : []
            "output_relations": relation_instances,
            # NOTE: see create_base_relations_from_doc/csv
                    for data columns
            "nclasses": self.config.model.padding_idx # dummy class
            "labels2idx": {},
            "idx2label": {}}
            ]
    """

    output_relations = []
    for doc_id, doc in enumerate(docs):
        output_relations.append(
            self.create_base_relations_from_doc(doc, doc_id=str(doc_id),))

    return output_relations

get_labels classmethod

get_labels(relation_labels: list[str], config: ConfigRelCAT) -> tuple[int, dict[str, int], dict[int, str]]

This is used to update labels in config with unencountered classes/labels ( if any are encountered during training).

Parameters:

• relation_labels

  (list[str]) –

  new labels to add

• config

  (ConfigRelCAT) –

  config

Returns:

• tuple[int, dict[str, int], dict[int, str]] –

  tuple[int, dict[str, int], dict[int, str]]: label count, labesl2idx mapping, idx2labels mapping

Source code in medcat-v2/medcat/components/addons/relation_extraction/rel_dataset.py

@classmethod
def get_labels(cls, relation_labels: list[str], config: ConfigRelCAT
               ) -> tuple[int, dict[str, int], dict[int, str]]:
    """This is used to update labels in config with unencountered
    classes/labels ( if any are encountered during training).

    Args:
        relation_labels (list[str]): new labels to add
        config (ConfigRelCAT): config

    Returns:
        tuple[int, dict[str, int], dict[int, str]]: label count,
            labesl2idx mapping, idx2labels mapping
    """
    curr_class_id = 0

    config_labels2idx: dict[str, int] = config.general.labels2idx
    config_idx2labels: dict[int, str] = config.general.idx2labels

    relation_labels = [relation_label.strip()
                       for relation_label in relation_labels]

    for relation_label in set(relation_labels):
        if relation_label not in config_labels2idx.keys():
            while curr_class_id in [
                    int(label_idx) for
                    label_idx in config_idx2labels.keys()]:
                curr_class_id += 1
            config_labels2idx[relation_label] = int(curr_class_id)
            config_idx2labels[int(curr_class_id)] = relation_label

    return (len(config_labels2idx.keys()), config_labels2idx,
            config_idx2labels,)