| """ |
| Check specific SNOMED CT concept codes against: |
| 1. Local OMOP Athena vocabularies (vocabularies/CONCEPT.csv) |
| 2. Live SNOMED International Snowstorm API & US Edition API |
| 3. NIH NLM Clinical Tables API |
| """ |
|
|
| import csv |
| import json |
| import os |
| import sys |
| import urllib.error |
| import urllib.parse |
| import urllib.request |
|
|
| if hasattr(sys.stdout, 'reconfigure'): |
| sys.stdout.reconfigure(encoding='utf-8') |
|
|
| TARGET_CODES = { |
| "410604004": "Subject of record (Person)", |
| "419149003": "Recommendation (Qualifying value)", |
| "1178652002": "Clinical recommendation statements (Record artifact)", |
| "86111000119109": "Clinical guideline (Record artifact)", |
| } |
|
|
| print("=" * 80) |
| print("🔍 VERIFYING TARGET SNOMED CT CODES") |
| print("=" * 80) |
|
|
| |
| print("\n[Phase 1] Scanning vocabularies/CONCEPT.csv for target codes...") |
| local_results = {code: [] for code in TARGET_CODES} |
|
|
| if os.path.exists("vocabularies/CONCEPT.csv"): |
| with open("vocabularies/CONCEPT.csv", "r", encoding="utf-8") as f: |
| reader = csv.DictReader(f, delimiter="\t") |
| for row in reader: |
| ccode = row.get("concept_code") |
| if ccode in TARGET_CODES: |
| local_results[ccode].append({ |
| "concept_id": row.get("concept_id"), |
| "concept_code": ccode, |
| "concept_name": row.get("concept_name"), |
| "vocabulary_id": row.get("vocabulary_id"), |
| "domain_id": row.get("domain_id"), |
| "concept_class_id": row.get("concept_class_id"), |
| "standard_concept": row.get("standard_concept"), |
| "invalid_reason": row.get("invalid_reason"), |
| "valid_start_date": row.get("valid_start_date"), |
| "valid_end_date": row.get("valid_end_date"), |
| }) |
| else: |
| print("vocabularies/CONCEPT.csv not found.") |
|
|
| for code, expected_desc in TARGET_CODES.items(): |
| matches = local_results.get(code, []) |
| print(f"\nTarget: {code} [{expected_desc}]") |
| if matches: |
| for m in matches: |
| print(f" [LOCAL] OMOP ID: {m['concept_id']} | Vocab: {m['vocabulary_id']} | Std: {m['standard_concept']} | Class: {m['concept_class_id']} | Domain: {m['domain_id']}") |
| print(f" Name: '{m['concept_name']}' | Invalid: {m['invalid_reason']} | Dates: {m['valid_start_date']} to {m['valid_end_date']}") |
| else: |
| print(" [LOCAL] NOT FOUND in local CONCEPT.csv (may be newer release or national extension)") |
|
|
| |
| print("\n" + "=" * 80) |
| print("[Phase 2] Checking Live Online SNOMED CT Ontologies / APIs...") |
| print("=" * 80) |
|
|
| headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'} |
|
|
| snowstorm_branches = [ |
| ("International (MAIN)", "https://browser.ihtsdotools.org/snowstorm/snomed-ct/browser/MAIN/concepts/"), |
| ("US Edition (MAIN/SNOMEDCT-US)", "https://browser.ihtsdotools.org/snowstorm/snomed-ct/browser/MAIN/SNOMEDCT-US/concepts/"), |
| ] |
|
|
| for code, expected_desc in TARGET_CODES.items(): |
| print(f"\n>>> Checking Code: {code} ({expected_desc})") |
| |
| |
| found_online = False |
| for branch_name, base_url in snowstorm_branches: |
| url = f"{base_url}{code}" |
| try: |
| req = urllib.request.Request(url, headers=headers) |
| with urllib.request.urlopen(req, timeout=8) as resp: |
| data = json.loads(resp.read().decode('utf-8')) |
| fsn = data.get('fsn', {}).get('term', 'N/A') |
| pt = data.get('pt', {}).get('term', 'N/A') |
| active = data.get('active', False) |
| sem_tag = data.get('fsn', {}).get('semanticTag', 'N/A') |
| print(f" [{branch_name}]: Active={active} | PT='{pt}' | FSN='{fsn}' | SemanticTag='{sem_tag}'") |
| found_online = True |
| break |
| except urllib.error.HTTPError as he: |
| if he.code == 404: |
| continue |
| else: |
| print(f" [{branch_name}]: HTTP {he.code}") |
| except Exception as e: |
| print(f" [{branch_name}]: Error {e}") |
| |
| if not found_online: |
| print(f" [Snowstorm]: Not found on MAIN or US branches.") |
| |
| |
| nlm_url = f"https://clinicaltables.nlm.nih.gov/api/snmd/v3/search?terms={code}&maxList=5" |
| try: |
| req = urllib.request.Request(nlm_url, headers=headers) |
| with urllib.request.urlopen(req, timeout=8) as resp: |
| data = json.loads(resp.read().decode('utf-8')) |
| count = data[0] |
| if count > 0: |
| print(f" [NIH NLM Search by Code {code}]: {count} matches") |
| for item in data[3]: |
| print(f" -> Code: {item[0]} | Name: {item[1]}") |
| else: |
| print(f" [NIH NLM Search by Code {code}]: 0 matches") |
| except Exception as e: |
| print(f" [NIH NLM Search Error]: {e}") |
|
|
