my_utils/file_utils.py

import requests, logging, json
from lib.CoNLL_Annotation import read_conll, read_conll_generator

logger = logging.getLogger(__name__)


def dict_to_file(my_dict, out_path):
    with open(out_path, "w") as out:
        out.write(json.dump(my_dict))

def file_to_dict(file_path):
    d = {}
    with open(file_path) as f:
        d = f.load(f)
    return d  


def file_generator(file_path):
    with open(file_path, "r") as data_file:
        logger.info("Reading instances from lines in file at: %s", file_path)
        for line in data_file:
            if not line: continue
            yield line


def get_file_text_chunk(line_generator, chunk_size, token_class, comment_str="###C:"):
    file_has_next = True
    chunk, n_sents = read_conll(line_generator, chunk_size, token_class, comment_str=comment_str)
    if n_sents == 0: file_has_next = False
    sents, gld, meta = [], [], []
    for anno in chunk:
        if len(anno.metadata) > 0: meta.append("\n".join(anno.metadata))
        sents.append(anno.get_sentence())
        gld.append(anno.get_pos_tags())
    return sents, gld, file_has_next


def get_file_chunk(line_generator, chunk_size, token_class, comment_str="###C:"):
    file_has_next = True
    chunk, n_sents = read_conll(line_generator, chunk_size, token_class, comment_str=comment_str)
    if n_sents < chunk_size: file_has_next = False
    raw_text = ""
    for anno in chunk:
        if len(anno.metadata) > 0: 
            raw_text += "\n".join(anno.metadata) + "\n"
        else:
            raw_text += "\n"
        for tok in anno.tokens:
            raw_text += tok.get_conllU_line() + "\n"
        raw_text += "\n"
    return raw_text, file_has_next, n_sents


def turku_parse_file(raw_text, filename, chunk_ix):
    out_file_str = f"{filename}.parsed.{chunk_ix}.conllu"
    # For each file make a request to obtain the parse back
    logger.info(f"Sending Request {chunk_ix} to Parser Server...")
    response = requests.post("http://localhost:7689/", data=raw_text.encode('utf-8'))
    response_to_file(response.text, out_file_str)



def response_to_file(response_str, fname):
    fout = open(fname, "w")
    fout.write(response_str)
    fout.close()