hiplab/parser
hiplab
/
parser
1
0
Fork 0
Code Issues 4 Pull Requests Packages Projects 1 Releases 1 Wiki Activity

v2.0 upgrade #10

Merged
steve merged 26 commits from v2.0 into master 2024-02-09 16:00:08 +00:00
Conversation 0 Commits 26 Files Changed 1774 +101 -39
3 changed files with 101 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 3eb5c5b568 - Show all commits

12
healthcareio/x12/plugins/default/__init__.py
View File

@ -1,5 +1,6 @@
"""
This file serves as the interface (so to speak) to the x12 parser, plugin interface
This file serves as the interface (so to speak) to the x12 parser, plugin interface, We implement default plugins that will handle parsing,
In addition to the allow custom plugins to be written/loaded and these will be given priority over the default ones+6+++++++++++++++++++++++++++++++++++++++++++++++++
@TODO:
- How to write custom plugin
- Provide interface for meta-data (expected)
@ -11,11 +12,14 @@ from . import header
from . import body
EDI = body.BODY
__version__ = '0.01'
__author__ = 'The Phi Technology'
def instance(**_args):
pass
# class Parser :
#
# defining commong functions that can/should be used accross the board
#
# # class Parser :
# def __init__(**_args):
# folder = _args['path']
# files = [ os.sep.join(_name,folder) for _name in os.listdir(folder)]

17
healthcareio/x12/plugins/default/body.py
View File

@ -21,16 +21,29 @@ class BODY (HEADER):
ref IL,40,41,82,85,PR ...
Information about entities (doctors, clearing house, provider). we should be mindful of the references
"""
# _CODE_INDEX = 1
_CODE_INDEX = 1
CONTEXT_MAP = {
'2':{'field':'payer'},
'PR':{'field':'payer'},
'41':{'field':'header'},
'IL':{'field':'patient','map':{'type':2,'first_name':4,'last_name':3}},
'P5':{'field':'plan_sponsor'},
'82':{'field':'rendering_provider','map':{'type':2,'first_name':4,'last_name':3}},
'85':{'field':'billing_provider'}
}
_args ['plugin-context'] = {'@ref':CONTEXT_MAP}
# _map = {_CODE_INDEX:{'41':'submitter','40':'receiver','PR':'payer'}}
_columns = ['type','name','id']
_indexes = [1,3,-1]
# _info = [{'index':'40','field':'receiver'},{'index':'41','field':'submitter'},{'index':'PR','field':'payer'}]
_info = self.parse(_columns,_indexes,**_args)
self.lastelement = _info
return _info
def N3 (self,**_args):
"""
"""
Expected Element N3
"""

111
healthcareio/x12/plugins/default/common.py
View File

@ -74,7 +74,7 @@ class X12DOCUMENT (Process):
_field = _config['field'] if 'field' in _config else {}
_label = _config['label'] if 'label' in _config else {}
return _field,_label
def merge(self,**_args):
def consolidate(self,**_args):
#
# This function overrides the old configuration with the new configuration specifications
#
@ -92,24 +92,28 @@ class X12DOCUMENT (Process):
if '@ref' in _config :
# _columns,_indexes = [],[]
_row = _args['row']
_ref = _config['@ref']
_ref = _config['@ref']
for _anchor in _ref:
# print ([_anchor,_anchor == _row[1].strip()])
if _anchor == _row[1].strip() :
_field,_label = self._getObjectAtributes(_ref[_anchor])
_map = _ref[_anchor]['map'] if 'map' in _ref[_anchor] else {}
if _map :
_columns,_indexes = self._getColumnsIndexes([],[],_map)
else:
# print ([_anchor,_indexes,_columns])
_map = dict(zip(_columns,_indexes))
pass
break
# _columns,_indexes = _columns + _map.keys()
return {'columns':_columns,'index':_indexes,'field':_field,'label':_label}
_out = {'columns':_columns,'index':_indexes,'field':_field,'label':_label}
return _out
def legacy(self,**_args):
#
# This function returns the legacy configuration (default parsing)
@ -128,9 +132,17 @@ class X12DOCUMENT (Process):
def __init__(self,**_args):
super().__init__()
self._mode = _args['mode'] if 'mode' in _args else 'NAMES'
self.lastelement = {} # This to store in the loop
if 'files' in _args :
self.files = _args['files']
self._config = _args['config'] if 'config' in _args else {}
#
# NM1 is a fluid type and thus will be cached in order to recreate the hierarchy
# @TODO:
# -add this to the configuration
#
self._hierarchy = {'NM1':['N1','N2','N3','N4']}
self._document = []
self._x12FileType = None
@ -218,8 +230,16 @@ class X12DOCUMENT (Process):
# _field = _field if not _refField else _refField
# _label = _label if not _refLabel else _refLabel
_outInfo = self._configHandler.merge(row=_args['row'],columns=columns,index=index,config=_config)
#
# @TODO:
# There should be a priority in parsing i.e plugin - config
#
if 'plugin-context' in _args :
_config = _args['plugin-context'] #dict(_config,**_args['plugin-context'])
_outInfo = self._configHandler.consolidate(row=_args['row'],columns=columns,index=index,config=_config)
_field,_label = _outInfo['field'],_outInfo['label']
_columns,_index = _outInfo['columns'],_outInfo['index']
@ -240,42 +260,61 @@ class X12DOCUMENT (Process):
# _element = _row[0]
_configKeys = [] #list(self._config.keys())
_configTree = [] #list(self._config.values())
if 'config' in _args :
_config = _args['config']
_configKeys = list(_config.keys())
_configTree = list(_config.values())
else:
_config = {}
# _configKeys = [] #list(self._config.keys())
# _configTree = [] #list(self._config.values())
# if 'config' in _args :
# _config = _args['config']
# _configKeys = list(_config.keys())
# _configTree = list(_config.values())
# else:
# _config = {}
_info = dict(zip(_columns,_row[_index].tolist()))
_document = _args['document'] if 'document' in _args else {}
#
# @TODO:
# Apply parsing/casting function to the object retrieved
# _apply(_info) #-- the object will be processed accordingly
#
# Extracting configuration (minimal information)
# _config = _args['config'] if 'config' in _args else {}
# _config = self._config
# if '@ref' in _config :
# print (_config['@ref'])
# _values = _config['@ref']
# print (_values)
#
# @TODO:
# The objects parsed must be augmented against the appropriate ones e.g: NM1 <- N1,N2,N3,N4
# - Find a way to drive this from a configuration ...
#
if _field :
if not _field in _document :
return {_field:_info}
_item = {_field:_info}
else:
return self.merge(_document[_field],_info)
_item = self.merge(_document[_field],_info)
elif _label :
if not _label in _document :
return {_label:[_info]}
_item = {_label:[_info]}
else:
return _document[_label] + [_info]
_item = _document[_label] + [_info]
else:
return _info
_item = _info
if _ELEMENT in self._hierarchy and _field:
# print ([_field,_item])
self.lastelement = _item
pass
else:
for key in self._hierarchy :
if _ELEMENT in self._hierarchy[key] :
_ikey = list(self.lastelement.keys())[0]
_oldinfo = self.lastelement[_ikey]
_item = {_ikey: self.merge(_oldinfo,_item)}
break
return _item
else:
#
#
print (_config)
return columns
def elements(self):
"""
@ -347,16 +386,22 @@ class X12DOCUMENT (Process):
_header = self.apply(_info['header'])
# print (json.dumps(_header))
_tmp = {}
for _content in _info['blocks'] :
_body = self.apply(_content,header=_header)
_body = self.apply(_content,header=_header)
_doc = self.merge(_header,_body)
if _doc and 'claim_id' in _doc:
# X12DOCUMENT._queue.put(_document)
_documents += [_doc]
_documents += [self.merge(_tmp,_doc)]
_tmp = {}
else:
#
# The document is being built and not yet ready
_tmp = self.merge(_tmp,_doc)
except Exception as e:
#