parser/healthcareio/__main__.py

#!/usr/bin/env 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 :
    # parse {x12}
    healthcare-io parse <x12_folder>

    # export {x12}
    healthcare-io export 

        - 
    Embedded    :

"""
# import healthcareio
import typer
from typing import Optional
from typing_extensions import Annotated
import uuid
import os
import meta
import json
import time
from healthcareio import x12
from healthcareio.x12.parser import X12Parser
import requests

# 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') :
    """\b
    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    
    
    :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":meta.__version__,
            "copyright":meta.__author__

        }
    }
    #
    # Let create or retrieve a user's key/token to make sure he/she has access to features they need
    # This would also allow us to allow the users to be informed of new versions
    #
    try:
        host = "https://healthcareio.the-phi.com" if 'PARSER_HOST_URL' not in os.environ else os.environ['PARSER_HOST']
        url = f"{host}/api/users/signup"
        _body = {"email":email,"version":meta.__version__}
        _headers = {"content-type":"application/json"}
        resp  = requests.post(url,headers=_headers,data=json.dumps(_body))
        if resp.ok :
            _config['system'] = dict(_config['system'],**resp.json())
    except Exception as e:
        print (e)
        pass
    # store this on disk
    f = open(os.sep.join([CONFIG_FOLDER,'config.json']),'w')
    f.write(json.dumps(_config))
    f.close()
    _msg = f"""
        Thank you for considering using our {{x12}} parser verion {meta.__version__}
        The generatted configuration file found at {CONFIG_FOLDER}
        The database provider is {provider}

        visit {host} to learn more about the features,
    
    """

    print (_msg)
@app.command(name='about')
def copyright():

    for note in [meta.__name__,meta.__author__,meta.__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.
    
    plugin_folder   folder containing user defined plugins (default are loaded)
    
    config_path     default configuration path
    """
    
    _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
    print ("...................... FINISHED .........................")
    #
    #

@app.command(name="export")    
def publish (file_type:str,path:str):
    """
    This function will export to a different database
    file_type       values are either claims or remits 
    
    path            path to export configuration (data transport file)

    file_type       claims or remits
    """
    if file_type in ['837','claims'] :
        _type = 'claims'
    elif file_type in ['835','remits']:
        _type = 'remits'
    
        
if __name__ == '__main__' :
    
    app()