-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbasic_functions.py
More file actions
108 lines (86 loc) · 3.88 KB
/
Copy pathbasic_functions.py
File metadata and controls
108 lines (86 loc) · 3.88 KB
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
from object_classes import ClinicalTrial, ArmGroup, Intervention
from typing import List, Tuple, Optional
import pandas as pd
from thefuzz import fuzz
def parse_clinical_trial(data, trial_source: Optional[str] = None, annotation_status: Optional[str] = None) -> ClinicalTrial:
"""Parse clinical trial data"""
protocol = data['protocolSection']
derived = data.get('derivedSection', {})
# Get mesh terms from the meshes list
conditions_mesh_terms = []
for mesh in derived.get('conditionBrowseModule', {}).get('meshes', []):
if 'term' in mesh:
conditions_mesh_terms.append(mesh['term'])
interventions_mesh_terms = []
for mesh in derived.get('interventionBrowseModule', {}).get('meshes', []):
if 'term' in mesh:
interventions_mesh_terms.append(mesh['term'])
# Extract interventions
interventions = []
for inv in protocol.get('armsInterventionsModule', {}).get('interventions', []):
intervention: Intervention = {
"type": inv.get('type', ''),
"name": inv.get('name', ''),
"description": inv.get('description', ''),
"arm_group_labels": inv.get('armGroupLabels', []),
"other_names": inv.get('otherNames', [])
}
interventions.append(intervention)
# Extract arm groups
arm_groups = []
for arm in protocol.get('armsInterventionsModule', {}).get('armGroups', []):
arm_group: ArmGroup = {
"label": arm.get('label', ''),
"type": arm.get('type', ''),
"description": arm.get('description', ''),
"intervention_names": arm.get('interventionNames', [])
}
arm_groups.append(arm_group)
# Create main trial object
trial: ClinicalTrial = {
"nct_id": protocol['identificationModule']['nctId'],
"brief_title": protocol['identificationModule']['briefTitle'],
"official_title": protocol['identificationModule']['officialTitle'],
"brief_summary": protocol.get('descriptionModule', {}).get('briefSummary', ''),
"detailed_description": protocol.get('descriptionModule', {}).get('detailedDescription', ''),
"conditions": protocol.get('conditionsModule', {}).get('conditions', []),
"keywords": protocol.get('conditionsModule', {}).get('keywords', []),
"study_type": protocol.get('designModule', {}).get('studyType', ''),
"arm_groups": arm_groups,
"interventions": interventions,
"condition_mesh_terms": conditions_mesh_terms,
"intervention_mesh_terms": interventions_mesh_terms,
"eligibility_criteria": protocol.get('eligibilityModule', {}).get('eligibilityCriteria', ''),
"trial_source": trial_source,
"annotation_status": annotation_status,
}
return trial
def match_drugs_to_nci_thesaurus(drug, nci_df) -> dict:
"""
Match drug names against NCI thesaurus and return matches with codes and definitions
"""
best_match = None
best_score = 0
# Search through display names and synonyms
for _, row in nci_df.iterrows():
# Check display name
score = fuzz.ratio(drug.lower(), str(row['display name']).lower())
if score > best_score:
best_score = score
best_match = row
# Check synonyms if present
if pd.notna(row['synonyms']):
for synonym in str(row['synonyms']).split('|'):
score = fuzz.ratio(drug.lower(), synonym.lower())
if score > best_score:
best_score = score
best_match = row
match = {}
if best_score > 97: # Only include true matches
match = {
'drug_name': drug,
'nci_code': best_match['code'],
'nci_term': best_match['display name'],
'definition': best_match['definition']
}
return match