aboutsummaryrefslogtreecommitdiff
path: root/workflows/pull-data/genbank/utils.py
blob: 96920a5c613828e4852dbd72c878b392a3553afa (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import os


def is_integer(string_to_check):
    try:
        int(string_to_check)
        return True
    except ValueError:
        return False


def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i + n]


def check_and_get_ontology_dictionaries(dir_ontology_dictionaries):
    """
    Check duplicated entry by looking in all dictionaries
    """

    field_to_term_to_uri_dict = {}

    path_dict_xxx_csv_list = [
        os.path.join(dir_ontology_dictionaries, name_xxx_csv) for name_xxx_csv in
        os.listdir(dir_ontology_dictionaries) if name_xxx_csv.endswith('.csv')
    ]

    for path_dict_xxx_csv in path_dict_xxx_csv_list:
        print(f'Read {path_dict_xxx_csv}')

        with open(path_dict_xxx_csv) as f:
            for line in f:
                if len(line.split(',')) > 2:
                    term, uri = line.strip('\n').split('",')
                else:
                    term, uri = line.strip('\n').split(',')

                term = term.strip('"')

                if term in field_to_term_to_uri_dict:
                    print(f'Warning: in the dictionaries there are more entries for the same term ({term}).')
                    continue

                field_to_term_to_uri_dict[term] = uri

    # Prepare separated dictionaries (to avoid, for example, that a valid IRI for species is accepted as specimen)
    field_to_term_to_uri_dict = {}

    for path_dict_xxx_csv in path_dict_xxx_csv_list:
        field = os.path.basename(path_dict_xxx_csv).split('.')[0]

        field_to_term_to_uri_dict[field] = {}

        with open(path_dict_xxx_csv) as f:
            for line in f:
                if len(line.split(',')) > 2:
                    term, uri = line.strip('\n').split('",')
                else:
                    term, uri = line.strip('\n').split(',')

                term = term.strip('"')

                if term in field_to_term_to_uri_dict[field]:
                    print(f'Warning: in the {field} dictionary there are more entries for the same term ({term}).')
                    continue

                field_to_term_to_uri_dict[field][term] = uri

    return field_to_term_to_uri_dict