-
Notifications
You must be signed in to change notification settings - Fork 1
/
example_data_source.py
44 lines (29 loc) · 1.27 KB
/
example_data_source.py
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
from data_sources.patient_data_source import PatientDataSource, Patient, Observation
import pandas as pd
class ExamplePatient(Patient):
def __init__(self, patient_info):
self.patient_info = patient_info
def get_indentifier_value(self):
return str(self.patient_info['patient_id'])
def get_name(self):
split_name = self.patient_info['patient_name'].split(' ')
return split_name[1], split_name[0]
class ExampleObservation(Observation):
def __init__(self, observation_info):
self.observation_info = observation_info
def get_observation_type(self):
return 'bodyweight'
def get_value(self):
return self.observation_info['body_weight_kg']
def get_time(self):
return self.observation_info['date']
class ExampleDataSource(PatientDataSource):
def __init__(self, csv_file):
self.data = pd.read_csv(csv_file)
def get_all_patients(self):
unique_patients = self.data.drop_duplicates('patient_id')
return (ExamplePatient(row) for _, row in unique_patients.iterrows())
def get_patient_observations(self, patient):
patient_id = int(patient.get_indentifier_value())
observations_for_patient = self.data[self.data['patient_id']==patient_id]
return (ExampleObservation(row) for _, row in observations_for_patient.iterrows())