utility/gen_dict_diff.py at main · AMP-SCZ/utility · 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
#!/usr/bin/env python

import sys
from os.path import abspath, dirname, join as pjoin
from os import getcwd, chdir
import pandas as pd
from datetime import datetime
SCRIPTDIR=dirname(abspath(__file__))

# ground truth dictionary
df1=pd.read_csv(sys.argv[1], encoding='ISO-8859-1')
# network's dictionary
df2=pd.read_csv(sys.argv[2], encoding='ISO-8859-1')
# network name
network=sys.argv[3]

datestamp=datetime.now().strftime('%Y%m%d')
suffix=f'{network}_{datestamp}'

# when downloaded through GUI
var_header='Variable / Field Name'
form_header='Form Name'
branch_header='Branching Logic (Show field only if...)'
calc_header='Choices, Calculations, OR Slider Labels'

# when downloaded through API
var_header='field_name'
form_header='form_name'
branch_header='branching_logic'
calc_header='select_choices_or_calculations'

ground_groups=df1.groupby(form_header)
target_groups=df2.groupby(form_header)


nonexistent_vars=[]
df_branch=pd.DataFrame(columns=[var_header,'AMP-SCZ logic','Network logic'])
df_branch.set_index(var_header, inplace=True)
df_calc=pd.DataFrame(columns=[var_header,'AMP-SCZ calculation','Network calculation'])
df_calc.set_index(var_header, inplace=True)

for form in ground_groups.groups.keys():
    if form not in target_groups.groups.keys():
        continue

    if '_consent_' in form and 'informed_consent_run_sheet' not in form:
        continue

    print(form)

    dfg= ground_groups.get_group(form)
    dft= target_groups.get_group(form)

    dfg.set_index(var_header, inplace=True)
    dft.set_index(var_header, inplace=True)

    for v,row in dfg.iterrows():
        try:
            dft.loc[v]
        except KeyError:
            nonexistent_vars.append(v)
            continue


        # compare branching logic
        if not pd.isna(row[branch_header]) and row[branch_header]!=dft.loc[v,branch_header]:
            df_branch.loc[v]= [row[branch_header], dft.loc[v,branch_header]]

        # compare calculation
        if row['field_type']=='calc':
            if not pd.isna(row[calc_header]) and row[calc_header]!=dft.loc[v,calc_header]:
                df_calc.loc[v]= [row[calc_header], dft.loc[v,calc_header]]


df_var=pd.DataFrame(data={var_header:nonexistent_vars})

dir_bak=getcwd()
chdir(pjoin(SCRIPTDIR,'dict_diff'))
df_var.to_csv(f'ampscz_vars_absent_in_{suffix}.csv', index=False)
df_branch.to_csv(f'branch_logic_diff_ampscz_{suffix}.csv')
df_calc.to_csv(f'calc_diff_ampscz_{suffix}.csv')
chdir(dir_bak)