knowledgeEngineering/knowledge_base_loader.py at master · diana3372/knowledgeEngineering · 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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import pandas as pd

class Disorder:
    def __init__(self, name, id):
        self.name = name
        self.id = id

    def build_symptoms_dict(self, symptoms, symptoms_ids):
        self.symptom_id_to_severity = {}
        for s_id, item in enumerate(symptoms.iteritems()):
            s, v = item
            assert s == symptoms_ids[s_id], 'Wrong mapping symptom to id'
            self.symptom_id_to_severity[s_id] = v

class KB_loader:
    def __init__(self):
        kb_folder = 'knowledge_base'
        symptoms_ids_file = '{}/symptoms_to_id.csv'.format(kb_folder)
        disorders_symptoms_file = '{}/disorders_symptoms.csv'.format(kb_folder)
        symptoms_questions_file = '{}/symptoms_to_questions.csv'.format(kb_folder)

        self.id_to_symptom_name = self.load_symptoms(symptoms_ids_file)
        self.disorders = self.load_disorders(disorders_symptoms_file)
        self.symptom_id_to_questions = self.load_questions(symptoms_questions_file)

    def load_symptoms(self, symptoms_ids_file):
        symptoms_ids_df = pd.read_csv(symptoms_ids_file, header=None)
        symptoms_ids = {} # dictionary id to symptom name
        for row in symptoms_ids_df.iterrows():
            row_data = row[1]
            name = row_data[0]
            id = row_data[1]
            symptoms_ids[id] = name

        return symptoms_ids

    def load_disorders(self, disorders_symptoms_file):
        # Read data
        disorders_symptoms_df = pd.read_csv(disorders_symptoms_file)
        # Filter out columns with unnamed headers
        disorders_symptoms_df = disorders_symptoms_df.loc[:, ~disorders_symptoms_df.columns.str.contains('^Unnamed')]

        disorders = []
        for disorder_count, row in enumerate(disorders_symptoms_df.iterrows()):
            row_data = row[1]
            disorder_name = row_data[0]
            symptoms = row_data[1:]
            disorder = Disorder(disorder_name, disorder_count)
            disorder.build_symptoms_dict(symptoms, self.id_to_symptom_name)
            disorders.append(disorder)

        return disorders

    def load_questions(self, symptoms_questions_file):
        symptoms_questions_df = pd.read_csv(symptoms_questions_file)

        symptoms_to_questions = {} #dictionary symptom id to questions
        for symptom_id, row in enumerate(symptoms_questions_df.iterrows()):
            row_data = row[1]

            # Create empty list of questions
            symptoms_to_questions[symptom_id] = []

            for i, column in enumerate(row_data):
                if i > 0 and not pd.isnull(column):
                    symptoms_to_questions[symptom_id].append(column)

        return symptoms_to_questions