immersioncoolingdatabase/database.py at main · ezhil384/immersioncoolingdatabase · GitHub

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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
# -*- coding: utf-8 -*-
"""
database.py
~~~~~~~~~~~~~~~~~~~~~~~~~~~

ImmersionDataBase data structures.
author: Shu Huang (sh2009@cam.ac.uk)
"""

from chemdataextractor_immersion.chemdataextractor15 import Document
import json
import copy
from chemdataextractor_immersion.chemdataextractor15.model.model import ThermalConductivity, Viscosity, DielectricConstant, FlashPoint


class ImmersionDataBase:

    def __init__(self, paper_root, save_root, filename):
        self.dic = None
        self.filename = filename
        self.paper_root = paper_root
        self.count = 0
        self.save_root = save_root

    def write_into_file(self):
        with open('{}/{}.json'.format(self.save_root, self.filename), 'a', encoding='utf-8') as json_file:
            json.dump(self.dic, json_file, ensure_ascii=False)
            json_file.write('\n')
        return

    def extract(self, file):
        """

        :param file: The parsing files (HTML/XML...)
        :return: Write the record into the documents
        """
        # try:
        f = open(file, 'rb')
        d = Document.from_file(f)
        print('parsing ' + file)
        rough = d.records.serialize()
        data = []
        for dic in rough:
            if 'Compound' in dic:
                continue
            try:
                dic['metadata'] = d.metadata[0].serialize()
                if dic['metadata']['doi'] == "None":
                    pass
            except BaseException:
                pass
            self.count += 1
            if self.is_valid(dic):
                dic_list = self.distribute(dic)
                data += dic_list
        if len(data) <= 3:
            for i in data:
                i['warning'] = 1
        for new_dic in data:
            self.dic = new_dic
            self.write_into_file()
        print(str(self.count) + ' relations in total')
        print(file + ' is done')
        f.close()
        # except BaseException:
        #     pass

    def is_valid(self, dic):
        """
        Check if the data record is valid or not
        :param dic:
        :return:
        """
        try:
                if 'names' in next(iter(dic.values()))['compound']['Compound']:
                    return True
        except BaseException:
                return False

    def distribute(self, dic):
        """
        Extract chemical names if a length of a list > 1

        :param dic: A dictionary returned by CDE
        :return: A list of dictionaries with valid records
        """
        # Create a key 'names' (list)
        name_length = next(iter(dic.values()))['compound']['Compound']['names']
        next(iter(dic.values()))['names'] = [name_length[0]]
        if len(name_length) > 1:
            for j in name_length[1:]:
                if j.lower() not in [x.lower()
                                     for x in next(iter(dic.values()))['names']]:
                    next(iter(dic.values()))['names'].append(j)

        # Update the key 'value' as a list of float
        # next(iter(dic.values()))['value'] = json.loads(
        #     next(iter(dic.values()))['value'])
        val_obj = next(iter(dic.values()))
        if isinstance(val_obj['value'], str):
            val_obj['value'] = json.loads(val_obj['value'])
        # Distribute
        dic_lists = self.distribute_value_and_names(dic)

        return dic_lists

    def distribute_value_and_names(self, dic):
        dic_list = []
        # Extract the actual lists from the dictionary
        val_data = next(iter(dic.values()))
        names = val_data['names']
        values = val_data['value']

        # Ensure values is always a list for consistent iteration
        if not isinstance(values, list):
            values = [values]
        if not isinstance(names, list):
            names = [names]

        len_names = len(names)
        len_values = len(values)

        for name in names:
            for value in values:
                # CRITICAL FIX: If value is still a list, take the last element
                current_val = None if isinstance(value, list) and len(value) == 0 else (value[-1] if isinstance(value, list) else value)

                try:
                    copydic = copy.deepcopy(dic)
                    target = next(iter(copydic.values()))
                    target['value'] = float(current_val)
                    target['names'] = name
                    dic_list.append(copydic)
                except (ValueError, TypeError):
                    # Skip values that cannot be converted to float (e.g., None or empty strings)
                    continue
        return dic_list