import copy from . import util import transport import numpy as np import time import pandas as pd from multiprocessing import Process import json from healthcareio.logger import X12Logger def build (**_args): """ This function will build SQL statements to create a table (perhaps not needed) :plugins loaded plugins :x12 837|835 file types """ _plugins=_args['plugins'] _x12 = _args['x12'] _template = util.template(plugins=_plugins)[_x12] _primaryKey = util.getPrimaryKey(plugins=_plugins,x12=_x12) _tables = [] _main = {} for _name in _template : _item = _template[_name] #copy.deepcopy(_template[_name]) if _primaryKey not in _item and type(_item) == dict: _item[_primaryKey] = '' _tables.append({_name:_item}) else: _main[_name] = '' _name = getContext(_x12) _tables += [{_name:_main}] _template[_name] = _main return _template #_tables def getContext(_x12) : return 'claims' if _x12 == '837' else 'remits' def format(**_args) : """ :rows rows for the :primary_key primary_key field name :x12 file format """ # _name = _args['table'] _rows = _args['rows'] _primary_key = _args['primary_key'] _x12 = _args['x12'] _mainTableName = getContext(_x12) _tables = {_mainTableName:[]} for _claim in _rows : # # # # Turn the claim into a relational model ... # # _main = {} _pkvalue = None if _primary_key in _claim : _pkvalue = _claim[_primary_key] for _attrName in _claim : _item = _claim[_attrName] _item = update(_item,_primary_key,_pkvalue) # # We have a casting problem, with relational data-store and JSON objects # if type(_item) == str and (_item.startswith("{") or _item.startswith("{")) : try: _item = json.loads(_item) except Exception as ee : # print (ee) pass if _attrName not in _tables and type(_item) in [dict,list]: _tables[_attrName] = [] if type(_item) in [dict,list] : _tables[_attrName] += _item if type(_item) == list else [_item] pass else: # # This section suggests we found a main table attribute _main[_attrName] = _item _tables[_mainTableName].append(_main) return _tables def update (_item,key,value): if type(_item) not in [dict,list] : return _item if type(_item) == dict : _item[key] = value else: # # List, we will go through every item and update accordingly _index = 0 for _row in _item : if type(_row) == dict : _row['_index'] = _index _row[key] = value return _item def init(**_args): """ This function will kick off the export process provided claims/remits and the loaded plugins (not sure why) It requires the data it is pulling to be consistently formatted (otherwise nothing can be done) :plugins :store data store information i.e {source,target} specifications for data-transport :x12 file type i.e 837|835 """ _file_type = _args['x12'] _plugins = _args['plugins'] _store = _args['store'] _default = build(plugins=_plugins,x12=_file_type) _logger = X12Logger(store = _store['source']) _df = read(store = _store['source'],x12=_file_type) # # @LOG : if _logger : _logger.log(module='init',action='export-init',data={'rows':_df.shape[0],'attributes':list(_df.columns)}) _pkey = util.getPrimaryKey(plugins = _plugins, x12=_file_type) SEGMENTS = 4 # arbitrary choice _indexes = np.array_split(np.arange(_df.shape[0]),SEGMENTS) jobs = [] _tables = {} for _ii in _indexes : try: _data = format(rows= _df.iloc[_ii].to_dict(orient='records'),x12=_file_type,primary_key=_pkey) _thread = Process(target=post,args=({'store':_store['target'],'data':_data,'default':_default,'x12':_file_type},)) _thread.start() jobs.append(_thread) _tables = list(_data.keys()) except Exception as e: # # Log: sigment, print (e) pass # # @LOG : if _logger : _logger.log(module='init',action='export-wait',data={'jobs':len(jobs),'tables':_tables}) if jobs : # jobs[0].start() # jobs[0].join() while jobs : jobs = [thread for thread in jobs if thread.is_alive()] time.sleep(1) def read (**_args): _store = copy.copy(_args['store']) _x12 = _args['x12'] _store['table'] = getContext(_x12) #'claims' if _x12 == '837' else 'remits' _store['context'] = 'read' reader = transport.factory.instance(**_store) # # @TODO: reading should support streaming (for scalability) _df = reader.read() return _df def post(_args): _data = _args['data'] _store = _args['store'] _store['context'] = 'write' _default = _args['default'] _prefix = 'clm_' if _args['x12'] == '837' else 'rem_' # if 'claims' in _data or 'remits' in _data : # _key = 'claims' if 'claims' in _data else 'remits' # _data = _data[_key] for _name in _data : _tablename = _prefix+_name _store['table'] = _tablename if _name not in ['remits','claims'] else _name _store['context']='write' _store['lock'] = True writer = transport.factory.instance(**_store) if len(_data[_name]) == 0 and _name in _default and not writer.has(table=_tablename): _rows = [_default[_name]] else: _rows = _data[_name] writer.write(pd.DataFrame(_rows).fillna('')) if hasattr(writer,'close') : writer.close() # # Have a logger here to log what's going on ... # _xwriter = trasnport.factory.instance(**_store) # _xwriter.write(_df) # _info = format() pass