|
|
@@ -0,0 +1,509 @@
|
|
|
+import re
|
|
|
+import os
|
|
|
+import pandas as pd
|
|
|
+import numpy as np
|
|
|
+
|
|
|
+def _getDictionnaryKeys(dictionnary):
|
|
|
+ """
|
|
|
+ Function that get keys from a dict object and flatten sub dict.
|
|
|
+ """
|
|
|
+
|
|
|
+ keys_array = []
|
|
|
+ for key in dictionnary.keys():
|
|
|
+ keys_array.append(key)
|
|
|
+ if (type(dictionnary[key]) == type({})):
|
|
|
+ keys_array = keys_array+_getDictionnaryKeys(dictionnary[key])
|
|
|
+ return(keys_array)
|
|
|
+
|
|
|
+class pandasToBrat:
|
|
|
+
|
|
|
+ """
|
|
|
+ Class for Pandas brat folder management.
|
|
|
+ For each brat folder, there is an instance of pandasToBrat.
|
|
|
+ It supports importation and exportation of configurations for relations and entities.
|
|
|
+ Documents importation and exportation.
|
|
|
+ Annotations and entities importation and exportation.
|
|
|
+
|
|
|
+ Inputs :
|
|
|
+ folder, str : path of brat folder
|
|
|
+ """
|
|
|
+
|
|
|
+ def __init__(self, folder):
|
|
|
+ self.folder = folder
|
|
|
+ self.conf_file = 'annotation.conf'
|
|
|
+
|
|
|
+ self.emptyDFCols = {
|
|
|
+ "annotations":["id","type_id", "word", "label", "start", "end"],
|
|
|
+ "relations":["id","type_id","relation","Arg1","Arg2"]
|
|
|
+ }
|
|
|
+
|
|
|
+ # Adding '/' to folder path if missing
|
|
|
+ if(self.folder[-1] != '/'):
|
|
|
+ self.folder += '/'
|
|
|
+
|
|
|
+ # Creating folder if do not exist
|
|
|
+ if (os.path.isdir(self.folder)) == False:
|
|
|
+ os.mkdir(self.folder)
|
|
|
+
|
|
|
+ # Loading conf file if exists | creating empty conf file if not
|
|
|
+ self.read_conf()
|
|
|
+
|
|
|
+ def _emptyData(self):
|
|
|
+ fileList = self._getFileList()
|
|
|
+ nb_files = fileList.shape[0]
|
|
|
+
|
|
|
+ confirmation = input("Deleting all data ({} files), press y to confirm :".format(nb_files))
|
|
|
+ if confirmation == 'y':
|
|
|
+ fileList["filename"].apply(lambda x: os.remove(self.folder+x))
|
|
|
+ print("{} files deleted.".format(nb_files))
|
|
|
+
|
|
|
+ def _generateEntitiesStr (self, conf, data = '', level = 0):
|
|
|
+
|
|
|
+ if (type(conf) != type({})):
|
|
|
+ return data
|
|
|
+
|
|
|
+ # Parsing keys
|
|
|
+ for key in conf.keys():
|
|
|
+ value = conf[key]
|
|
|
+
|
|
|
+ if value == True:
|
|
|
+ data += '\n'+level*'\t'+key
|
|
|
+ elif value == False:
|
|
|
+ data += '\n'+level*'\t'+'!'+key
|
|
|
+ elif type(value) == type({}):
|
|
|
+ data += '\n'+level*'\t'+key
|
|
|
+ data = self._generateEntitiesStr(value, data, level+1)
|
|
|
+
|
|
|
+ return data
|
|
|
+
|
|
|
+ def _writeEntitiesLevel (self, conf, data, last_n = -1):
|
|
|
+
|
|
|
+ for n in range(last_n,len(conf)):
|
|
|
+ # If empty : pass, if not the last line : pass
|
|
|
+ if (conf[n] != '' and n > last_n):
|
|
|
+ level = len(conf[n].split("\t"))-1
|
|
|
+ if (n+1 <= len(conf)): # Level of next item
|
|
|
+ next_level = len(conf[n+1].split("\t"))-1
|
|
|
+ else:
|
|
|
+ next_level = level
|
|
|
+
|
|
|
+ splitted_str = conf[n].split("\t")
|
|
|
+ str_clean = splitted_str[len(splitted_str)-1]
|
|
|
+
|
|
|
+ if (level >= next_level): # On écrit les lignes de même niveau
|
|
|
+ if (str_clean[0] == '!'):
|
|
|
+ data[str_clean[1:]] = False
|
|
|
+ else:
|
|
|
+ data[str_clean] = True
|
|
|
+
|
|
|
+ if (level > next_level):
|
|
|
+ # On casse la boucle
|
|
|
+ break
|
|
|
+ elif (level < next_level): # On écrit les lignes inférieurs par récurence
|
|
|
+ splitted_str = conf[n].split("\t")
|
|
|
+ last_n, data[str_clean] = self._writeEntitiesLevel(conf, {}, n)
|
|
|
+
|
|
|
+ return(n, data)
|
|
|
+
|
|
|
+ def _readRelations(self, relations, entities = []):
|
|
|
+ data = {}
|
|
|
+
|
|
|
+ for relation in relations.split("\n"):
|
|
|
+ if relation != '':
|
|
|
+ relation_data = relation.split("\t")[0]
|
|
|
+ args = list(map(lambda x: x.split(":")[1], relation.split("\t")[1].split(", ")))
|
|
|
+ args_valid = list(filter(lambda x: x in entities, args))
|
|
|
+
|
|
|
+ if (len(args_valid) > 0):
|
|
|
+ data[relation_data] = {"args":args_valid}
|
|
|
+
|
|
|
+ return data
|
|
|
+
|
|
|
+ def _writeRelations(self, relations, entities = []):
|
|
|
+ data = ''
|
|
|
+ for relation in relations:
|
|
|
+ args_array = list(filter(lambda x: x in entities, relations[relation]["args"]))
|
|
|
+
|
|
|
+ if (len(args_array) > 0):
|
|
|
+ data += '\n'+relation+'\t'
|
|
|
+
|
|
|
+ for n in range(0, len(args_array)):
|
|
|
+ data += int(bool(n))*', '+'Arg'+str(n+1)+':'+args_array[n]
|
|
|
+
|
|
|
+ return data
|
|
|
+
|
|
|
+ def read_conf (self):
|
|
|
+ """
|
|
|
+ Get the current Brat configuration.
|
|
|
+ Output :
|
|
|
+ Dict containing "entities" and "relations" configurations.
|
|
|
+ """
|
|
|
+
|
|
|
+ if (os.path.isfile(self.folder+self.conf_file)):
|
|
|
+
|
|
|
+ # Reading file
|
|
|
+ file = open(self.folder+self.conf_file)
|
|
|
+ conf_str = file.read()
|
|
|
+ file.close()
|
|
|
+
|
|
|
+ # Splitting conf_str
|
|
|
+ conf_data = re.split(re.compile(r"\[[a-zA-Z]+\]", re.DOTALL), conf_str)[1:]
|
|
|
+
|
|
|
+ data = {}
|
|
|
+
|
|
|
+ # Reading enteties
|
|
|
+ data["entities"] = self._writeEntitiesLevel(conf_data[0].split("\n"), {})[1]
|
|
|
+
|
|
|
+ # Reading relations
|
|
|
+ entitiesKeys = _getDictionnaryKeys(data["entities"])
|
|
|
+ data["relations"] = self._readRelations(conf_data[1], entitiesKeys)
|
|
|
+
|
|
|
+ return(data)
|
|
|
+
|
|
|
+ else:
|
|
|
+ self.write_conf()
|
|
|
+ self.read_conf()
|
|
|
+
|
|
|
+ def write_conf(self, entities = {}, relations = {}, events = {}, attributes = {}):
|
|
|
+ """
|
|
|
+ Write or overwrite configuration file.
|
|
|
+ It actually doesn't suppport events and attributes configuration data.
|
|
|
+
|
|
|
+ inputs :
|
|
|
+ entities, dict : dict containing the entities. If an entities do have children, his value is an other dict, otherwise, it is set as True.
|
|
|
+ relations, dict : dict containing the relations between entities, each key is a relation name, the value is a dict with a "args" key containing the list of related entities.
|
|
|
+ """
|
|
|
+
|
|
|
+ # TODO : Add events and attributes support.
|
|
|
+
|
|
|
+ conf_str = ''
|
|
|
+
|
|
|
+ # Entities
|
|
|
+ conf_str += '\n\n[entities]'
|
|
|
+ conf_str += self._generateEntitiesStr(entities)
|
|
|
+
|
|
|
+ # relations
|
|
|
+ conf_str += '\n\n[relations]'
|
|
|
+ entitiesKeys = _getDictionnaryKeys(entities)
|
|
|
+ conf_str += self._writeRelations(relations, entitiesKeys)
|
|
|
+
|
|
|
+ # attributes
|
|
|
+ conf_str += '\n\n[attributes]'
|
|
|
+
|
|
|
+ # events
|
|
|
+ conf_str += '\n\n[events]'
|
|
|
+
|
|
|
+ # Write conf file
|
|
|
+ file = open(self.folder+self.conf_file,'w')
|
|
|
+ file.write(conf_str)
|
|
|
+ file.close()
|
|
|
+
|
|
|
+ def _getFileList(self):
|
|
|
+ # Listing files
|
|
|
+ filesDF = pd.DataFrame({'filename':pd.Series(os.listdir(self.folder))})
|
|
|
+ filesDFSplitted = filesDF["filename"].str.split(".", expand = True)
|
|
|
+ filesDF["id"] = filesDFSplitted[0]
|
|
|
+ filesDF["filetype"] = filesDFSplitted[1]
|
|
|
+
|
|
|
+ filesDF = filesDF[filesDF["filetype"].isin(["txt","ann"])]
|
|
|
+
|
|
|
+ return(filesDF)
|
|
|
+
|
|
|
+ def _parseData(self):
|
|
|
+
|
|
|
+ # Listing files
|
|
|
+ filesDF = self._getFileList()
|
|
|
+
|
|
|
+ # Getting data from txt and ann
|
|
|
+ filesDF_txt = filesDF.rename(columns = {"filename":"text_data"}).loc[filesDF["filetype"] == "txt", ["id","text_data"]]
|
|
|
+ filesDF_ann = filesDF.rename(columns = {"filename":"annotation"}).loc[filesDF["filetype"] == "ann", ["id","annotation"]]
|
|
|
+ dataDF = filesDF_txt.join(filesDF_ann.set_index("id"), on = "id")
|
|
|
+ dataDF["text_data"] = dataDF["text_data"].apply(lambda x: open(self.folder+x).read())
|
|
|
+ dataDF["annotation"] = dataDF["annotation"].apply(lambda x: open(self.folder+x).read())
|
|
|
+
|
|
|
+ return(dataDF)
|
|
|
+
|
|
|
+ def read_text(self):
|
|
|
+
|
|
|
+ """
|
|
|
+ read_text
|
|
|
+ Get a pandas DataFrame containing the brat documents.
|
|
|
+
|
|
|
+ Input : None
|
|
|
+ Output : Pandas dataframe
|
|
|
+ """
|
|
|
+
|
|
|
+ dataDF = self._parseData()
|
|
|
+
|
|
|
+ return(dataDF[["id","text_data"]])
|
|
|
+
|
|
|
+ def read_annotation(self, ids = []):
|
|
|
+
|
|
|
+ """
|
|
|
+ read_annotation
|
|
|
+ Get annotations from the brat folder.
|
|
|
+ You can get specific annotation by filtering by id.
|
|
|
+
|
|
|
+ input :
|
|
|
+ ids, list (optionnal) : list of id for which you want the annotation data, if empty all annotations are returned.
|
|
|
+
|
|
|
+ output :
|
|
|
+ dict containing an annotations and relations data.
|
|
|
+ """
|
|
|
+
|
|
|
+ data = {}
|
|
|
+ data["annotations"] = pd.DataFrame(columns=self.emptyDFCols["annotations"])
|
|
|
+ data["relations"] = pd.DataFrame(columns=self.emptyDFCols["relations"])
|
|
|
+
|
|
|
+ dataDF = self._parseData()[["id","annotation"]]
|
|
|
+ dataDF = dataDF[(dataDF["annotation"].isna() == False) & (dataDF["annotation"] != '')] # Removing empty annotation
|
|
|
+
|
|
|
+ # Filtering by ids
|
|
|
+ if (len(ids) > 0):
|
|
|
+ dataDF = dataDF[dataDF["id"].isin(pd.Series(ids).astype(str))]
|
|
|
+
|
|
|
+ if (dataDF.shape[0] > 0):
|
|
|
+
|
|
|
+ # Ann data to pandas
|
|
|
+ dataDF = dataDF.join(dataDF["annotation"].str.split("\n").apply(pd.Series).stack().reset_index(level = 0).set_index("level_0")).reset_index(drop = True).drop("annotation", axis = 1).rename(columns = {0: "annotation"})
|
|
|
+ dataDF = dataDF[dataDF["annotation"].str.len() > 0].reset_index(drop = True)
|
|
|
+ dataDF = dataDF.join(dataDF["annotation"].str.split("\t", expand = True).rename(columns = {0: 'type_id', 1: 'data', 2: 'word'})).drop("annotation", axis = 1)
|
|
|
+ dataDF["type"] = dataDF["type_id"].str.slice(0,1)
|
|
|
+
|
|
|
+ ## Annotations
|
|
|
+ data["annotations"] = dataDF[dataDF["type"] == 'T']
|
|
|
+ if (data["annotations"].shape[0] > 0):
|
|
|
+ data["annotations"] = data["annotations"].join(data["annotations"]["data"].str.split(" ", expand = True).rename(columns = {0: "label", 1: "start", 2: "end"})).drop(columns = ["data","type"])
|
|
|
+
|
|
|
+ ## Relations
|
|
|
+ data["relations"] = dataDF[dataDF["type"] == 'R']
|
|
|
+
|
|
|
+ if (data["relations"].shape[0] > 0):
|
|
|
+ tmp_splitted = data["relations"]["data"].str.split(" ", expand = True).rename(columns = {0: "relation"})
|
|
|
+
|
|
|
+ ### Col names
|
|
|
+ rename_dict = dict(zip(list(tmp_splitted.columns.values[1:]), list("Arg"+tmp_splitted.columns.values[1:].astype(str).astype(object))))
|
|
|
+ tmp_splitted = tmp_splitted.rename(columns = rename_dict)
|
|
|
+
|
|
|
+ ### Merging data
|
|
|
+ tmp_splitted = tmp_splitted[["relation"]].join(tmp_splitted.loc[:,tmp_splitted.columns[tmp_splitted.columns != 'relation']].applymap(lambda x: x.split(":")[1]))
|
|
|
+ data["relations"] = data["relations"].join(tmp_splitted).drop(columns = ["data","type","word"])
|
|
|
+
|
|
|
+ return(data)
|
|
|
+
|
|
|
+ def _write_function(self, x, filetype = "txt", overwrite = False):
|
|
|
+
|
|
|
+ filenames = []
|
|
|
+
|
|
|
+ if (filetype == 'txt' or filetype == 'both'):
|
|
|
+ filenames.append(self.folder+str(x["filename"])+'.txt')
|
|
|
+
|
|
|
+ if (filetype == 'ann' or filetype == 'both'):
|
|
|
+ filenames.append(self.folder+str(x["filename"])+'.ann')
|
|
|
+
|
|
|
+ for filename in filenames:
|
|
|
+ try:
|
|
|
+ open(str(filename), "r")
|
|
|
+ is_file = True
|
|
|
+ except FileNotFoundError:
|
|
|
+ is_file = False
|
|
|
+
|
|
|
+ if ((is_file == False) or (overwrite == True)):
|
|
|
+ file = open(str(filename), "w")
|
|
|
+ file.write(x["content"])
|
|
|
+ file.close()
|
|
|
+
|
|
|
+ def write_text(self, text_id, text, empty = False, overWriteAnnotations = False):
|
|
|
+
|
|
|
+ """
|
|
|
+ write_text
|
|
|
+ Send text data from the brat folder.
|
|
|
+
|
|
|
+ input :
|
|
|
+ text_id, pd.Series : pandas series containing documents ids
|
|
|
+ text, pd.Series : pandas series containing documents text in the same order as text_id
|
|
|
+ empty, boolean : if True the brat folder is emptyied of all but configuration data (text and ann files) before writting
|
|
|
+ overwriteAnnotations, boolean : if True, the current annotation files are replaced by blank one
|
|
|
+ """
|
|
|
+
|
|
|
+ if overWriteAnnotations == True: # On controle la façon dont la variable est écrite
|
|
|
+ overwriteAnn = True
|
|
|
+ else:
|
|
|
+ overwriteAnn = False
|
|
|
+
|
|
|
+ if (type(text) == type(pd.Series()) and type(text_id) == type(pd.Series()) and text.shape[0] == text_id.shape[0]):
|
|
|
+
|
|
|
+ # ID check : check should be smaller than text : check if not inverted
|
|
|
+ if (text_id.astype(str).str.len().max() < text.astype(str).str.len().max()):
|
|
|
+
|
|
|
+ # empty : option to erase existing data
|
|
|
+ if (empty):
|
|
|
+ self._emptyData()
|
|
|
+
|
|
|
+ # Writting data
|
|
|
+ print("Writting data")
|
|
|
+ df_text = pd.DataFrame({"filename":text_id, "content":text})
|
|
|
+ df_ann = pd.DataFrame({"filename":text_id, "content":""})
|
|
|
+
|
|
|
+ df_text.apply(lambda x: self._write_function(x, filetype = "txt", overwrite = True), axis = 1)
|
|
|
+ df_ann.apply(lambda x: self._write_function(x, filetype = "ann", overwrite = overwriteAnn), axis = 1)
|
|
|
+ print("data written.")
|
|
|
+
|
|
|
+ else:
|
|
|
+ raise ValueError('ID is larger than text, maybe you inverted them.')
|
|
|
+
|
|
|
+ else:
|
|
|
+ raise ValueError('Incorrect variable type, expected two Pandas Series of same shape.')
|
|
|
+
|
|
|
+ def write_annotations(self, df, text_id, word, label, start, end, overwrite = False):
|
|
|
+
|
|
|
+ """
|
|
|
+ write_annotations
|
|
|
+ Send annotation data from the brat folder. Useful to pre-anotate some data.
|
|
|
+
|
|
|
+ input :
|
|
|
+ df, pd.Dataframe : dataframe containing annotations data, should contains the text id, the annotated word, the annotated label, the start and end offset.
|
|
|
+ text_id, str : name of the column in df which contains the document id
|
|
|
+ word, str : name of the column in df which contains the annotated word
|
|
|
+ label, str : name of the column in df which contains the label of the annotated word
|
|
|
+ start, str : name of the column in df which contains the start offset
|
|
|
+ end, str : name of the column in df which contains the end offset
|
|
|
+ overwrite, boolean : if True, the current annotation files are replaced by new data, otherwise, the new annotations are merged with existing one
|
|
|
+ """
|
|
|
+
|
|
|
+ # Checking data types
|
|
|
+ if (type(df) == type(pd.DataFrame())):
|
|
|
+
|
|
|
+ # Loading df
|
|
|
+ df = df.rename(columns = {text_id:"id",word:"word",label:"label",start:"start",end:"end"})
|
|
|
+ df["type_id"] = df.groupby("id").cumcount()+1
|
|
|
+
|
|
|
+ # List of ids
|
|
|
+ ids = df["id"].unique()
|
|
|
+
|
|
|
+ # Loading current data
|
|
|
+ current_annotation = self.read_annotation(ids)
|
|
|
+ current_annotations = current_annotation["annotations"]
|
|
|
+ tmaxDFAnnotations = current_annotations.set_index(["id"])["type_id"].str.slice(1,).astype(int).reset_index().groupby("id").max().rename(columns = {"type_id":"Tmax"})
|
|
|
+
|
|
|
+ if (overwrite == True):
|
|
|
+ df["type_id"] = "T"+df["type_id"].astype(str)
|
|
|
+ new_annotations = df
|
|
|
+ else:
|
|
|
+ df = df.join(tmaxDFAnnotations, on = "id").fillna(0)
|
|
|
+ df["type_id"] = "T"+(df["type_id"]+df["Tmax"]).astype(int).astype(str)
|
|
|
+ df = df.drop(columns = ["Tmax"])
|
|
|
+
|
|
|
+ new_annotations = pd.concat((current_annotations, df[self.emptyDFCols["annotations"]])).reset_index(drop = True)
|
|
|
+
|
|
|
+ new_annotations.drop_duplicates() ## Removing duplicates
|
|
|
+
|
|
|
+ # Injecting new annotations
|
|
|
+ current_annotation["annotations"] = new_annotations
|
|
|
+
|
|
|
+ # Calling write function
|
|
|
+ self._write_annotation(current_annotation["annotations"], current_annotation["relations"])
|
|
|
+
|
|
|
+ else:
|
|
|
+ raise ValueError('Incorrect variable type, expected a Pandas DF.')
|
|
|
+
|
|
|
+
|
|
|
+ def write_relations(self, df, text_id, relation, overwrite = False):
|
|
|
+
|
|
|
+ # Checking data types
|
|
|
+ if (type(df) == type(pd.DataFrame())):
|
|
|
+
|
|
|
+ # Loading df
|
|
|
+ df = df.rename(columns = {text_id:"id",relation:"relation"})
|
|
|
+ df["type_id"] = df.groupby("id").cumcount()+1 # type_id
|
|
|
+
|
|
|
+ # Columns names
|
|
|
+ old_columns = df.columns[np.isin(df.columns, ["id", "relation","type_id"]) == False]
|
|
|
+ new_columns = "Arg"+np.array(list(range(1,len(old_columns)+1))).astype(str).astype(object)
|
|
|
+ df = df.rename(columns = dict(zip(old_columns, new_columns)))
|
|
|
+
|
|
|
+ # List of ids
|
|
|
+ ids = df["id"].unique()
|
|
|
+
|
|
|
+ # Loading current data
|
|
|
+ current_annotation = self.read_annotation(ids)
|
|
|
+ current_relations = current_annotation["relations"]
|
|
|
+ rmaxDFrelations = current_relations.set_index(["id"])["type_id"].str.slice(1,).astype(int).reset_index().groupby("id").max().rename(columns = {"type_id":"Rmax"})
|
|
|
+
|
|
|
+ if (overwrite == True):
|
|
|
+ df["type_id"] = "R"+df["type_id"].astype(str)
|
|
|
+ new_relations = df
|
|
|
+ else:
|
|
|
+ df = df.join(rmaxDFrelations, on = "id").fillna(0)
|
|
|
+ df["type_id"] = "R"+(df["type_id"]+df["Rmax"]).astype(int).astype(str)
|
|
|
+ df = df.drop(columns = ["Rmax"])
|
|
|
+
|
|
|
+ # Adding missing columns
|
|
|
+ if (len(df.columns) > len(current_relations.columns)):
|
|
|
+ for column in df.columns[np.isin(df.columns, current_relations.columns) == False]:
|
|
|
+ current_relations[column] = np.nan
|
|
|
+ else:
|
|
|
+ for column in current_relations.columns[np.isin(current_relations.columns, df.columns) == False]:
|
|
|
+ df[column] = np.nan
|
|
|
+
|
|
|
+ new_relations = pd.concat((current_relations, df[current_relations.columns])).reset_index(drop = True)
|
|
|
+
|
|
|
+ new_relations.drop_duplicates() ## Removing duplicates
|
|
|
+
|
|
|
+ # Injecting new annotations
|
|
|
+ current_annotation["relations"] = new_relations
|
|
|
+
|
|
|
+ # Calling write function
|
|
|
+ self._write_annotation(current_annotation["annotations"], current_annotation["relations"])
|
|
|
+
|
|
|
+ else:
|
|
|
+ raise ValueError('Incorrect variable type, expected a Pandas DF.')
|
|
|
+
|
|
|
+ def _generate_annotations_str (self, annotations):
|
|
|
+
|
|
|
+ annotations = annotations.reset_index(drop = True)
|
|
|
+ annotations["label_span"] = annotations[["label","start","end"]].apply(lambda x: ' '.join(x.astype(str).values), axis = 1)
|
|
|
+ annotations_str = '\n'.join(annotations[["type_id","label_span","word"]].apply(lambda x: '\t'.join(x.astype(str).values), axis = 1).values)
|
|
|
+
|
|
|
+ return(annotations_str)
|
|
|
+
|
|
|
+ def _generate_relations_str (self, relations):
|
|
|
+
|
|
|
+
|
|
|
+ relations = relations.fillna('').applymap(lambda x: '' if x == 'nan' else x) #cleaning data
|
|
|
+ columns = relations.columns[np.isin(relations.columns, ["id","type_id","relation"]) == False].values.tolist()
|
|
|
+ boolmap = relations[columns].transpose().applymap(lambda x: int(x != ''))
|
|
|
+ rct = relations[columns].transpose()
|
|
|
+
|
|
|
+ temp_relations = (boolmap*(np.array(np.repeat(rct.index,rct.shape[1])).reshape(rct.shape)+':')+rct.astype(str)).transpose()
|
|
|
+
|
|
|
+ relations_str = '\n'.join(relations[["type_id","relation"]].join(temp_relations[columns]).apply(lambda x: '\t'.join(x.values), axis = 1).values)
|
|
|
+
|
|
|
+ return(relations_str)
|
|
|
+
|
|
|
+ def _write_file(self, data):
|
|
|
+ file = open(self.folder+str(data["id"])+".ann", "w")
|
|
|
+ file.write(data["str_to_write"])
|
|
|
+ file.close()
|
|
|
+
|
|
|
+ def _write_annotation(self,annotations,relations):
|
|
|
+
|
|
|
+ # Checking data types
|
|
|
+ if (type(annotations) == type(pd.DataFrame()) and type(relations) == type(pd.DataFrame())):
|
|
|
+
|
|
|
+ # Gerenating str
|
|
|
+ data_annotations = annotations.groupby("id").agg(lambda x: self._generate_annotations_str(x)).iloc[:,0]
|
|
|
+ data_relations = relations.groupby("id").agg(lambda x: self._generate_relations_str(x)).iloc[:,0]
|
|
|
+
|
|
|
+ # Merging data
|
|
|
+ data = pd.DataFrame({"annotations":data_annotations, "relations":data_relations}).fillna('')
|
|
|
+ data["str_to_write"] = data.apply(lambda x : '\n'.join(x.values), axis = 1)
|
|
|
+ data = data.reset_index().rename(columns = {"index":"id"})
|
|
|
+
|
|
|
+ # Writting files
|
|
|
+ data.apply(self._write_file, axis = 1)
|
|
|
+
|
|
|
+ return(data)
|
|
|
+
|
|
|
+ else:
|
|
|
+ raise ValueError('Incorrect variable type, expected a Pandas DF.')
|
