parser/edi/__main__.py

106 lines
3.8 KiB
Python

"""
(c) 2019 Claims Toolkit,
Health Information Privacy Lab, Vanderbilt University Medical Center
Steve L. Nyemba <steve.l.nyemba@vanderbilt.edu>
Khanhly Nguyen <khanhly.t.nguyen@gmail.com>
This code is intended to process and parse healthcare x12 837 (claims) and x12 835 (remittances) into human readable JSON format.
The claims/outpout can be forwarded to a NoSQL Data store like couchdb and mongodb
Usage :
Commandline :
python edi --scope --config <path> --folder <path> --store <[mongo|disk|couch]> --<db|path]> <id|path>
with :
--scope <claims|remits>
--config path of the x12 to be parsed i.e it could be 835, or 837
--folder location of the files (they must be decompressed)
--store data store could be disk, mongodb, couchdb
--db|path name of the folder to store the output or the database name
Embedded in Code :
import edi.parser
import json
file = '/data/claim_1.x12'
conf = json.loads(open('config/837.json').read())
edi.parser.get_content(filename,conf)
"""
from params import SYS_ARGS
from transport import factory
from parser import *
import os
import json
import sys
if __name__ == '__main__' :
"""
The program was called from the command line thus we are expecting
parse in [claims,remits]
config os.sep.path.exists(path)
folder os.sep.path.exists(path)
store store ()
"""
p = len( set(['store','config','folder']) & set(SYS_ARGS.keys())) == 3 and ('db' in SYS_ARGS or 'path' in SYS_ARGS)
TYPE = {
'mongo':'mongo.MongoWriter',
'couch':'couch.CouchWriter',
'disk':'disk.DiskWriter'
}
INFO = {
'837':{'scope':'claims','section':'HL'},
'835':{'scope':'remits','section':'CLP'}
}
if p :
args = {}
scope = SYS_ARGS['config'][:-5].split(os.sep)[-1]
CONTEXT = INFO[scope]['scope']
#
# @NOTE:
# improve how database and data stores are handled.
if SYS_ARGS['store'] == 'couch' :
args = {'url': SYS_ARGS['url'] if 'url' in SYS_ARGS else 'http://localhost:5984'}
args['dbname'] = SYS_ARGS['db']
elif SYS_ARGS ['store'] == 'mongo':
args = {'host':SYS_ARGS['host']if 'host' in SYS_ARGS else 'localhost:27017'}
if SYS_ARGS['store'] in ['mongo','couch']:
args['dbname'] = SYS_ARGS['db'] if 'db' in SYS_ARGS else 'claims_outcomes'
args['doc'] = CONTEXT
TYPE = TYPE[SYS_ARGS['store']]
writer = factory.instance(type=TYPE,args=args)
if SYS_ARGS['store'] == 'disk':
writer.init(path = 'output-claims.json')
logger = factory.instance(type=TYPE,args= dict(args,**{"doc":"logs"}))
files = os.listdir(SYS_ARGS['folder'])
CONFIG = json.loads(open(SYS_ARGS['config']).read())
SECTION = INFO[scope]['section']
for file in files :
if 'limit' in SYS_ARGS and files.index(file) == int(SYS_ARGS['limit']) :
break
else:
filename = os.sep.join([SYS_ARGS['folder'],file])
try:
content,logs = get_content(filename,CONFIG,SECTION)
except Exception as e:
if sys.version_info[0] > 2 :
logs = [{"filename":filename,"msg":e.args[0]}]
else:
logs = [{"filename":filename,"msg":e.message}]
content = None
if content :
writer.write(content)
if logs:
logger.write(logs)
pass
else:
print (__doc__)