You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
113 lines
3.3 KiB
Python
113 lines
3.3 KiB
Python
"""
|
|
(c) 2019 EDI Parser 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 x12parser <action>
|
|
|
|
action:
|
|
- parser
|
|
- create.plugin
|
|
- register.plugin
|
|
-
|
|
Embedded :
|
|
|
|
"""
|
|
# import healthcareio
|
|
import typer
|
|
from typing import Optional
|
|
from typing_extensions import Annotated
|
|
import uuid
|
|
import os
|
|
import version
|
|
import json
|
|
import time
|
|
from healthcareio import x12
|
|
from healthcareio.x12.parser import X12Parser
|
|
|
|
# import healthcareio
|
|
# import healthcareio.x12.util
|
|
# from healthcareio.x12.parser import X12Parser
|
|
app = typer.Typer()
|
|
CONFIG_FOLDER = os.sep.join([os.environ['HOME'],'.healthcareio'])
|
|
@app.command(name='init')
|
|
def config(email:str,provider:str='sqlite') :
|
|
"""
|
|
Generate configuration file needed with default data store. For supported data-store providers visit https://hiplab.mc.vanderbilt.edu/git/hiplab/data-transport.git
|
|
|
|
:email your email
|
|
\r:provider data store provider (visit https://hiplab.mc.vanderbilt.edu/git/hiplab/data-transport.git)
|
|
"""
|
|
_db = "healthcareio"
|
|
# _PATH = os.sep.join([os.environ['HOME'],'.healthcareio'])
|
|
|
|
if not os.path.exists(CONFIG_FOLDER) :
|
|
os.mkdir(CONFIG_FOLDER)
|
|
|
|
if provider in ['sqlite','sqlite3'] :
|
|
_db = os.sep.join([CONFIG_FOLDER,_db+'.db3'])
|
|
|
|
_config = {
|
|
"store":{
|
|
"provider":provider,"database":_db,"context":"write"
|
|
},
|
|
"plugins":None,
|
|
"system":{
|
|
"uid":str(uuid.uuid4()),
|
|
"email":email,
|
|
"version":version.__version__,
|
|
"copyright":version.__author__
|
|
|
|
}
|
|
}
|
|
#
|
|
# store this on disk
|
|
f = open(os.sep.join([CONFIG_FOLDER,'config.json']),'w')
|
|
f.write(json.dumps(_config))
|
|
f.close()
|
|
@app.command(name='about')
|
|
def copyright():
|
|
|
|
for note in [version.__name__,version.__author__,version.__license__]:
|
|
print (note)
|
|
|
|
pass
|
|
@app.command()
|
|
def parse (claim_folder:str,plugin_folder:str = None,config_path:str = None):
|
|
"""
|
|
This function will parse 837 and or 835 claims given a location of parsing given claim folder and/or plugin folder
|
|
"""
|
|
|
|
_plugins,_parents = x12.plugins.instance(path=plugin_folder)
|
|
_files = x12.util.file.Location.get(path=claim_folder,chunks=10)
|
|
_path = config_path if config_path else os.sep.join([CONFIG_FOLDER,'config.json'])
|
|
|
|
if os.path.exists(_path) :
|
|
f = open(_path)
|
|
_config = json.loads(f.read())
|
|
f.close()
|
|
_store = _config['store']
|
|
# print (len(_files))
|
|
jobs = []
|
|
for _chunks in _files:
|
|
pthread = X12Parser(plugins=_plugins,parents=_parents,files=_chunks, store=_store)
|
|
pthread.start()
|
|
jobs.append(pthread)
|
|
while jobs :
|
|
jobs = [pthread for pthread in jobs if pthread.is_alive()]
|
|
time.sleep(1)
|
|
pass
|
|
else:
|
|
pass
|
|
#
|
|
#
|
|
if __name__ == '__main__' :
|
|
|
|
app() |