__init__.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573
  1. """
  2. (c) 2019 Healthcare/IO 1.0
  3. Vanderbilt University Medical Center, Health Information Privacy Laboratory
  4. https://hiplab.mc.vanderbilt.edu/healthcareio
  5. Authors:
  6. Khanhly Nguyen,
  7. Steve L. Nyemba<steve.l.nyemba@vanderbilt.edu>
  8. License:
  9. MIT, terms are available at https://opensource.org/licenses/MIT
  10. This parser was originally written by Khanhly Nguyen for her internship and is intended to parse x12 835,837 and others provided the appropriate configuration
  11. USAGE :
  12. - COMMAND LINE
  13. - EMBEDDED
  14. """
  15. import hashlib
  16. import json
  17. import os
  18. import sys
  19. from itertools import islice
  20. from multiprocessing import Process
  21. import transport
  22. import jsonmerge
  23. class void :
  24. pass
  25. class Formatters :
  26. def __init__(self):
  27. # self.config = config
  28. self.get = void()
  29. self.get.config = self.get_config
  30. self.parse = void()
  31. self.parse.sv3 = self.sv3
  32. self.parse.sv2 = self.sv2
  33. self.sv2_parser = self.sv2
  34. self.sv3_parser = self.sv3
  35. self.sv3_parse = self.sv3
  36. self.format_proc = self.procedure
  37. self.format_diag = self.diagnosis
  38. self.parse.procedure = self.procedure
  39. self.parse.diagnosis = self.diagnosis
  40. self.parse.date = self.date
  41. self.format_date = self.date
  42. self.format_pos = self.pos
  43. self.format_time = self.time
  44. def split(self,row,sep='*',prefix='HI') :
  45. """
  46. This function is designed to split an x12 row and
  47. """
  48. value = []
  49. if row.startswith(prefix) is False:
  50. for row_value in row.replace('~','').split(sep) :
  51. if '>' in row_value and not row_value.startswith('HC'):
  52. # if row_value.startswith('HC') or row_value.startswith('AD'):
  53. if row_value.startswith('AD'):
  54. value += row_value.split('>')[:2]
  55. pass
  56. else:
  57. value += [row_value]
  58. # value += row_value.split('>') if row.startswith('CLM') is False else [row_value]
  59. else :
  60. value.append(row_value.replace('\n',''))
  61. value = [xchar.replace('\r','') for xchar in value] #row.replace('~','').split(sep)
  62. else:
  63. value = [ [prefix]+ self.split(item,'>') for item in row.replace('~','').split(sep)[1:] ]
  64. return value if type(value) == list and type(value[0]) != list else value[0]
  65. def get_config(self,config,row):
  66. """
  67. This function will return the meaningfull parts of the configuration for a given item
  68. """
  69. _row = list(row) if type(row[0]) == str else list(row[0])
  70. _info = config[_row[0]] if _row[0] in config else {}
  71. key = None
  72. if '@ref' in _info:
  73. key = list(set(_row) & set(_info['@ref'].keys()))
  74. if key :
  75. key = key[0]
  76. return _info['@ref'][key]
  77. else:
  78. return {}
  79. if not _info and 'SIMILAR' in config:
  80. #
  81. # Let's look for the nearest key using the edit distance
  82. if _row[0] in config['SIMILAR'] :
  83. key = config['SIMILAR'][_row[0]]
  84. _info = config[key]
  85. return _info
  86. def hash(self,value):
  87. salt = os.environ['HEALTHCAREIO_SALT'] if 'HEALTHCAREIO_SALT' in os.environ else ''
  88. _value = str(value)+ salt
  89. if sys.version_info[0] > 2 :
  90. return hashlib.md5(_value.encode('utf-8')).hexdigest()
  91. else:
  92. return hashlib.md5(_value).hexdigest()
  93. def suppress (self,value):
  94. return 'N/A'
  95. def date(self,value):
  96. if len(value) > 8 or '-' in value:
  97. value = value.split('-')[0]
  98. if len(value) == 8 :
  99. year = value[:4]
  100. month = value[4:6]
  101. day = value[6:]
  102. return "-".join([year,month,day])[:10] #{"year":year,"month":month,"day":day}
  103. elif len(value) == 6 :
  104. year = '20' + value[:2]
  105. month = value[2:4]
  106. day = value[4:]
  107. #
  108. # We have a date formatting issue
  109. return "-".join([year,month,day])
  110. def time(self,value):
  111. pass
  112. def sv3(self,value):
  113. if '>' in value [1]:
  114. terms = value[1].split('>')
  115. return {'type':terms[0],'code':terms[1],"amount":float(value[2])}
  116. else:
  117. return {"code":value[2],"type":value[1],"amount":float(value[3])}
  118. def sv2(self,value):
  119. #
  120. # @TODO: Sometimes there's a suffix (need to inventory all the variations)
  121. #
  122. if '>' in value or ':' in value:
  123. xchar = '>' if '>' in value else ':'
  124. _values = value.split(xchar)
  125. modifier = {}
  126. if len(_values) > 2 :
  127. modifier= {"code":_values[2]}
  128. if len(_values) > 3 :
  129. modifier['type'] = _values[3]
  130. _value = {"code":_values[1],"type":_values[0]}
  131. if modifier :
  132. _value['modifier'] = modifier
  133. return _value
  134. else:
  135. return value
  136. def procedure(self,value):
  137. for xchar in [':','<','|','>'] :
  138. if xchar in value and len(value.split(xchar)) > 1 :
  139. #_value = {"type":value.split(':')[0].strip(),"code":value.split(':')[1].strip()}
  140. _value = {"type":value.split(xchar)[0].strip(),"code":value.split(xchar)[1].strip()}
  141. if len(value.split(xchar)) >2 :
  142. index = 1;
  143. for modifier in value.split(xchar)[2:] :
  144. _value['modifier_'+str(index)] = modifier
  145. index += 1
  146. break
  147. else:
  148. _value = str(value)
  149. return _value
  150. def diagnosis(self,value):
  151. return [ {"code":item[2], "type":item[1]} for item in value if len(item) > 1]
  152. def pos(self,value):
  153. """
  154. formatting place of service information within a segment (REF)
  155. @TODO: In order to accomodate the other elements they need to be specified in the configuration
  156. Otherwise it causes problems on export
  157. """
  158. xchar = '>' if '>' in value else ':'
  159. x = value.split(xchar)
  160. x = {"code":x[0],"indicator":x[1],"frequency":x[2]} if len(x) == 3 else {"code":x[0],"indicator":None,"frequency":None}
  161. return x['code']
  162. class Parser (Process):
  163. def __init__(self,path):
  164. """
  165. :path path of the configuration file (it can be absolute)
  166. """
  167. Process.__init__(self)
  168. self.utils = Formatters()
  169. self.get = void()
  170. self.get.value = self.get_map
  171. self.get.default_value = self.get_default_value
  172. _config = json.loads(open(path).read())
  173. self._custom_config = self.get_custom(path)
  174. self.config = _config['parser']
  175. self.store = _config['store']
  176. self.files = []
  177. self.set = void()
  178. self.set.files = self.set_files
  179. self.emit = void()
  180. self.emit.pre = None
  181. self.emit.post = None
  182. def get_custom(self,path) :
  183. """
  184. :path path of the configuration file (it can be absolute)
  185. """
  186. #
  187. #
  188. _path = path.replace('config.json','')
  189. if _path.endswith(os.sep) :
  190. _path = _path[:-1]
  191. _config = {}
  192. _path = os.sep.join([_path,'custom'])
  193. if os.path.exists(_path) :
  194. files = os.listdir(_path)
  195. if files :
  196. fullname = os.sep.join([_path,files[0]])
  197. _config = json.loads ( (open(fullname)).read() )
  198. return _config
  199. def set_files(self,files):
  200. self.files = files
  201. def get_map(self,row,config,version=None):
  202. # label = config['label'] if 'label' in config else None
  203. handler = Formatters()
  204. if 'map' not in config and hasattr(handler,config['apply']):
  205. pointer = getattr(handler,config['apply'])
  206. object_value = pointer(row)
  207. return object_value
  208. #
  209. # Pull the goto configuration that skips rows
  210. #
  211. omap = config['map'] if not version or version not in config else config[version]
  212. anchors = config['anchors'] if 'anchors' in config else []
  213. rewrite = config['rewrite'] if 'rewrite' in config else {}
  214. if type(row[0]) == str:
  215. object_value = {}
  216. for key in omap :
  217. index = omap[key]
  218. if anchors and set(anchors) & set(row):
  219. _key = list(set(anchors) & set(row))[0]
  220. aindex = row.index(_key)
  221. index = aindex + index
  222. if index < len(row) :
  223. value = row[index]
  224. if 'cast' in config and key in config['cast'] and value.strip() != '' :
  225. if config['cast'][key] in ['float','int'] :
  226. value = eval(config['cast'][key])(value)
  227. elif hasattr(handler,config['cast'][key]):
  228. pointer = getattr(handler,config['cast'][key])
  229. value = pointer(value)
  230. else:
  231. print ("Missing Pointer ",key,config['cast'])
  232. if type(value) == dict :
  233. for objkey in value :
  234. if type(value[objkey]) == dict :
  235. continue
  236. if 'syn' in config and value[objkey] in config['syn'] :
  237. # value[objkey] = config['syn'][ value[objkey]]
  238. pass
  239. if key in rewrite :
  240. _key = rewrite[key]
  241. if _key in value :
  242. value = value[_key]
  243. else:
  244. value = ""
  245. value = {key:value} if key not in value else value
  246. else:
  247. if 'syn' in config and value in config['syn'] :
  248. # value = config['syn'][value]
  249. pass
  250. if type(value) == dict :
  251. object_value = dict(object_value, **value)
  252. else:
  253. object_value[key] = value
  254. else:
  255. #
  256. # we are dealing with a complex object
  257. object_value = []
  258. for row_item in row :
  259. value = self.get.value(row_item,config,version)
  260. object_value.append(value)
  261. #
  262. # We need to add the index of the object it matters in determining the claim types
  263. #
  264. # object_value.append( list(get_map(row_item,config,version)))
  265. # object_value = {label:object_value}
  266. return object_value
  267. def apply(self,content,_code) :
  268. """
  269. :content content of a file i.e a segment with the envelope
  270. :_code 837 or 835 (helps get the appropriate configuration)
  271. """
  272. util = Formatters()
  273. # header = default_value.copy()
  274. value = {}
  275. for row in content[:] :
  276. row = util.split(row.replace('\n','').replace('~',''))
  277. _info = util.get.config(self.config[_code][0],row)
  278. if self._custom_config and _code in self._custom_config:
  279. _cinfo = util.get.config(self._custom_config[_code],row)
  280. else:
  281. _cinfo = {}
  282. if 'SV3' in row :
  283. print (row)
  284. print (_info)
  285. if _info or _cinfo:
  286. try:
  287. _info = jsonmerge.merge(_info,_cinfo)
  288. tmp = self.get.value(row,_info)
  289. if not tmp :
  290. continue
  291. if 'label' in _info :
  292. label = _info['label']
  293. if type(tmp) == list :
  294. value[label] = tmp if label not in value else value[label] + tmp
  295. else:
  296. # if 'DTM' in row :
  297. # print ([label,tmp,label in value])
  298. if label not in value :
  299. value[label] = []
  300. value[label].append(tmp)
  301. # if label not in value:
  302. # value[label] = [tmp]
  303. # else:
  304. # value[label].append(tmp)
  305. tmp['_index'] = len(value[label]) -1
  306. elif 'field' in _info :
  307. name = _info['field']
  308. # value[name] = tmp
  309. value = jsonmerge.merge(value,{name:tmp})
  310. else:
  311. value = dict(value,**tmp)
  312. pass
  313. except Exception as e :
  314. print (e.args[0])
  315. # print ('__',(dir(e.args)))
  316. pass
  317. return value if value else {}
  318. def get_default_value(self,content,_code):
  319. util = Formatters()
  320. TOP_ROW = content[1].split('*')
  321. CATEGORY= content[2].split('*')[1].strip()
  322. VERSION = content[1].split('*')[-1].replace('~','').replace('\n','')
  323. SUBMITTED_DATE = util.parse.date(TOP_ROW[4])
  324. SENDER_ID = TOP_ROW[2]
  325. row = util.split(content[3])
  326. _info = util.get_config(self.config[_code][0],row)
  327. value = self.get.value(row,_info,VERSION) if _info else {}
  328. value['category'] = {"setid": CATEGORY,"version":'X'+VERSION.split('X')[1],"id":VERSION.split('X')[0].strip()}
  329. value["submitted"] = SUBMITTED_DATE
  330. # value['version'] = VERSION
  331. # if _code== '835' :
  332. # value['receiver_id'] = SENDER_ID
  333. # else:
  334. # value['provider_id'] = SENDER_ID
  335. # pass
  336. value['sender_id'] = SENDER_ID
  337. #
  338. # Let's parse this for default values
  339. return jsonmerge.merge(value,self.apply(content,_code))
  340. def read(self,filename) :
  341. """
  342. :formerly get_content
  343. This function returns the of the EDI file parsed given the configuration specified. it is capable of identifying a file given the content
  344. :section loop prefix (HL, CLP)
  345. :config configuration with formatting rules, labels ...
  346. :filename location of the file
  347. """
  348. # section = section if section else config['SECTION']
  349. logs = []
  350. claims = []
  351. try:
  352. file = open(filename.strip(),errors='ignore')
  353. INITIAL_ROWS = list(islice(file,4)) #.readlines(4)
  354. _code = "unknown"
  355. if len(INITIAL_ROWS) == 1 :
  356. file = INITIAL_ROWS[0].split('~')
  357. INITIAL_ROWS = file[:4]
  358. if len(INITIAL_ROWS) < 3 :
  359. return None,[{"name":filename,"completed":False}],None
  360. _code = INITIAL_ROWS[2].split('*')[1].strip()
  361. section = self.config[_code][0]['SECTION'].strip()
  362. #
  363. # adjusting the
  364. DEFAULT_VALUE = self.get.default_value(INITIAL_ROWS,_code)
  365. DEFAULT_VALUE['name'] = filename.strip()
  366. #
  367. # In the initial rows, there's redundant information (so much for x12 standard)
  368. # index 1 identifies file type i.e CLM for claim and CLP for remittance
  369. segment = []
  370. index = 0;
  371. _toprows = []
  372. _default = None
  373. for row in file :
  374. row = row.replace('\r','')
  375. if not segment and not row.startswith(section):
  376. _toprows += [row]
  377. if row.startswith(section) and not segment:
  378. segment = [row]
  379. continue
  380. elif segment and not row.startswith(section):
  381. if not _default :
  382. _default = (self.apply(_toprows,_code))
  383. DEFAULT_VALUE = dict(DEFAULT_VALUE,**_default)
  384. segment.append(row)
  385. if len(segment) > 1 and row.startswith(section):
  386. #
  387. # process the segment somewhere (create a thread maybe?)
  388. #
  389. _claim = self.apply(segment,_code)
  390. if _claim :
  391. _claim['index'] = index #len(claims)
  392. # claims.append(dict(DEFAULT_VALUE,**_claim))
  393. claims.append( jsonmerge.merge(DEFAULT_VALUE,_claim))
  394. segment = [row]
  395. index += 1
  396. pass
  397. #
  398. # Handling the last claim found
  399. if segment[0].startswith(section) :
  400. default_claim = dict({"name":index},**DEFAULT_VALUE)
  401. claim = self.apply(segment,_code)
  402. if claim :
  403. claim['index'] = len(claims)
  404. schema = [key for key in claim.keys() if type(claim[key]) == list]
  405. if schema :
  406. schema = {"properties":dict.fromkeys(schema,{"mergeStrategy":"append"})}
  407. else:
  408. schema = {}
  409. merger = jsonmerge.Merger(schema)
  410. top_row_claim = self.apply(_toprows,_code)
  411. claim = merger.merge(claim,self.apply(_toprows,_code))
  412. # claims.append(dict(DEFAULT_VALUE,**claim))
  413. claims.append(merger.merge(DEFAULT_VALUE,claim))
  414. if type(file) != list :
  415. file.close()
  416. # x12_file = open(filename.strip(),errors='ignore').read().split('\n')
  417. except Exception as e:
  418. logs.append ({"parse":_code,"completed":False,"name":filename,"msg":e.args[0]})
  419. return [],logs,None
  420. rate = 0 if len(claims) == 0 else (1 + index)/len(claims)
  421. logs.append ({"parse":"claims" if _code == '837' else 'remits',"completed":True,"name":filename,"rate":rate})
  422. # self.finish(claims,logs,_code)
  423. return claims,logs,_code
  424. def run(self):
  425. if self.emit.pre :
  426. self.emit.pre()
  427. for filename in self.files :
  428. content,logs,_code = self.read(filename)
  429. self.finish(content,logs,_code)
  430. def finish(self,content,logs,_code) :
  431. args = self.store
  432. _args = json.loads(json.dumps(self.store))
  433. if args['type'] == 'mongo.MongoWriter' :
  434. args['args']['doc'] = 'claims' if _code == '837' else 'remits'
  435. _args['args']['doc'] = 'logs'
  436. else:
  437. args['args']['table'] = 'claims' if _code == '837' else 'remits'
  438. _args['args']['table'] = 'logs'
  439. if content :
  440. writer = transport.factory.instance(**args)
  441. writer.write(content)
  442. writer.close()
  443. if logs :
  444. logger = transport.factory.instance(**_args)
  445. logger.write(logs)
  446. logger.close()
  447. if self.emit.post :
  448. self.emit.post(content,logs)