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 pandas as pd
import numpy as np


# import healthcareio
# import healthcareio.x12.util
# from healthcareio.x12.parser import X12Parser
app = typer.Typer()
CONFIG_FOLDER = os.sep.join([os.environ['HOME'],'.healthcareio'])
HOST = "https://healthcareio.the-phi.com" if 'PARSER_HOST_URL' not in os.environ else os.environ['PARSER_HOST_URL']
@app.command(name='init')
def config(email:str,provider:str='sqlite',auth_file:str=None) :
    """\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"
    global CONFIG_FOLDER
    # _PATH = os.sep.join([os.environ['HOME'],'.healthcareio'])
    
    if not os.path.exists(CONFIG_FOLDER) :
        os.mkdir(CONFIG_FOLDER)
    
    #
    # NOTE:
    # if a provider is setup as an environment variable, we override the parameter
    # This is designed for developement and containers
    #
    if 'X12_DATABASE' in os.environ :
        provider = os.environ['X12_DATABASE']
    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__

        }
    }
    if auth_file and os.path.exists(auth_file) and provider not in ['sqlite','sqlite3'] :
        f = open(auth_file)
        _auth = json.loads(f.read())
        f.close()
        _config['store'] = dict(_config['store'],**_auth)
    #
    # 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:
        global HOST
        # 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():
    f"""
    This function will return information about the {meta.__name__}
    """
    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 ()
    print ("                        PARSED                           ")
    print ("...................... FINISHED .........................")
    #
    #
@app.command()
def check():
    """
    This function checks for the version running against the current version    
    """
    _info = [meta.__version__,None]
    url = f'{HOST}/api/store/version'
    try:
        resp= requests.post(url)
        _info[1] = resp.text if resp.status_code == 200 else "NA"
    except Exception as e:
        _info[1] = "NA"
        pass
    if _info[1] == "NA" :
        _msg = "Unavailable server (unreachable)"
    else:
        _msg = ""

    print ()
    _info =pd.DataFrame(_info,columns=["versions"],index=["Yours","Current"])
    print (_info)

    print (_msg)

@app.command(name="export-schema")
def export_schema (file_type:str):
    """
    This function will display the schema in JSON format of a given file/type
    """
    _plugins,_parents = x12.plugins.instance()
    if file_type not in ['835','837'] and file_type in ['claims','remits']:
        file_type = '835' if file_type == 'remits' else '837'
    _template = x12.publish.build(x12=file_type,plugins=_plugins)
    print ( json.dumps(_template))


@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 (835 or 837)
    """
    _type = None
    if file_type.strip() in ['837','claims'] :
        _type = 'claims'
        _x12 = '837'
    elif file_type.strip() in ['835','remits']:
        _type = 'remits'
        _x12 = '835'
    if _type :
        _store = {'source':os.sep.join([CONFIG_FOLDER,'config.json']),'target':path}
        for _key in _store :
            f = open(_store[_key])
            _store[_key] = json.loads(f.read())
            f.close()
        _store['source'] = _store['source']['store']
        
        _plugins,_parents = x12.plugins.instance()
        x12.publish.init(plugins=_plugins,x12=_x12,store=_store)
    else:
        print ("Can not determine type, (837 or 835)")

    print ()
    print ("                        EXPORT                           ")
    print ("...................... FINISHED .........................")
    
        
if __name__ == '__main__' :
    
    app()