Module Gnumed.business.gmGenericEMRItem

GNUmed clinical business object in generic form.

license: GPL v2 or later

Functions

def generic_item_type_str(table)

Expand source code

def generic_item_type_str(table):
        try:
                return _MAP_generic_emr_item_table2type_str[table]

        except KeyError:
                return _('unmapped entry type from table [%s]') % table

def get_generic_emr_items(encounters: list = None,
episodes: list = None,
issues: list = None,
patient=None,
soap_cats: str | list[str | None] | tuple[str | None] = None,
time_range=None,
order_by: str = None,
active_encounter: cEncounter = None,
return_pks: bool = False,
exclude_soap_cats: bool = False,
item_types2exclude: list[str] = None) ‑> list[cGenericEMRItem] | list[int]

Expand source code

def get_generic_emr_items (
        encounters:list=None,
        episodes:list=None,
        issues:list=None,
        patient=None,
        soap_cats:str | list[str|None] | tuple[str|None]=None,
        time_range=None,                        # in days
        order_by:str=None,
        active_encounter:cEncounter=None,
        return_pks:bool=False,
        exclude_soap_cats:bool=False,
        item_types2exclude:list[str]=None
) -> list[cGenericEMRItem] | list[int]:
        """Return generic EMR items.

        Args:
                timerange: number of days for going back in EMR
        """
        faulty_args = (
                (patient is None) and
                (encounters is None) and
                (episodes is None) and
                (issues is None) and
                (active_encounter is None)
        )
        assert not faulty_args, 'one of <patient>, <episodes>, <issues>, <active_encounter> must not be None'

        if (patient is not None) and (active_encounter is not None):
                if patient != active_encounter['pk_patient']:
                        raise AssertionError('<patient> (%s) and <active_encounter>["pk_patient"] (%s) must match, if both given', patient, active_encounter['pk_patient'])

        if order_by is None:
                order_by = 'ORDER BY clin_when, pk_episode, scr, modified_when, src_table'
        else:
                order_by = 'ORDER BY %s' % order_by
        if (patient is None) and (active_encounter is not None):
                patient = active_encounter['pk_patient']
        where_parts = []
        args = {}
        if patient is not None:
                where_parts.append('pk_patient = %(pat)s')
                args['pat'] = patient
        if soap_cats:
                cats = list(soap_cats)
                # work around bug in psycopg2 not being able to properly
                # adapt None to NULL inside tuples
                if None in cats:
                        where_arg = '((c_vej.soap_cat = ANY(%(soap_cat)s)) OR (c_vej.soap_cat IS NULL))'
                        cats.remove(None)
                else:
                        where_arg = 'c_vej.soap_cat = ANY(%(soap_cat)s)'
                if exclude_soap_cats:
                        where_arg = 'NOT (%s)' % where_arg
                where_parts.append(where_arg)
                args['soap_cat'] = cats
        if item_types2exclude:
                where_parts.append('NOT (src_table = ANY(%(item_types)s))')
                args['item_types'] = item_types2exclude
        if time_range is not None:
                where_parts.append("clin_when > (now() - '%s days'::interval)" % time_range)
        if encounters is not None:
                where_parts.append("pk_encounter = ANY(%(encs)s)")
                args['encs'] = encounters
        if episodes is not None:
                where_parts.append("pk_episode = ANY(%(epis)s)")
                args['epis'] = episodes
        if issues is not None:
                where_parts.append("pk_health_issue = ANY(%(issues)s)")
                args['issues'] = issues
        SQL_journal = _SQL_get_generic_emr_items
        if len(where_parts) > 0:
                SQL_journal += '\nWHERE\n\t'
                SQL_journal += '\n\t\tAND '.join(where_parts)
        SQL = SQL_journal + '\n' + order_by
#       if active_encounter is None:
#               cmd = SQL_journal + '\n' + order_by
#       else:
#               args['pk_enc'] = active_encounter['pk_encounter']
#               args['enc_start'] = active_encounter['started']
#               args['enc_last_affirmed'] = active_encounter['last_affirmed']
#               args['enc_type'] = active_encounter['l10n_type']
#               args['enc_pat'] = active_encounter['pk_patient']
#               cmd = __SQL_union % (
#                       SQL_journal,
#                       _SQL_get_hints_as_generic_emr_items
#               ) + '\n' + order_by
        rows = gmPG2.run_ro_query(sql = SQL, args = args)
        if return_pks:
                return [ {
                        'src_table': r['src_table'],
                        'src_pk': r['src_pk']
                } for r in rows ]

        return [ cGenericEMRItem(row = {
                'data': r,
                'pk_obj': {'src_table': r['src_table'], 'src_pk': r['src_pk']}
        } ) for r in rows ]

Return generic EMR items.

Args

-----=

timerange

number of days for going back in EMR

Classes

class cGenericEMRItem (aPK_obj: int | dict = None, row: dict = None, link_obj=None)

Expand source code

class cGenericEMRItem(gmBusinessDBObject.cBusinessDBObject):
        """Represents an entry in clin.v_emr_journal."""

        _cmd_fetch_payload:str = _SQL_get_generic_emr_items + "WHERE src_table = %(src_table)s AND src_pk = %(src_pk)s"
        _cmds_store_payload:list = []
        _updatable_fields:list = ['']

        #--------------------------------------------------------
        def format(self, eol=None):
                lines = self.formatted_header
                lines.append(gmTools.u_box_horiz_4dashes * 40)
                lines.extend(self._payload['narrative'].strip().split('\n'))
                if eol is None:
                        return lines

                return eol.join(lines)

        #--------------------------------------------------------
        def format_tech_info(self) -> list[str]:
                lines = []
                lines.append('%s [%s]' % (self.item_type_str, self._payload['real_soap_cat']))
                lines.append(' ' + _('Table: %s##tx: EMR item source (database table, mostly)') % self._payload['src_table'])
                lines.append(' ' + _('Version: %s (%s)') % (
                        self._payload['row_version'],
                        _('Original##tx: ... of clinical record item') if (self._payload['row_version'] == 1) else _('Revision##tx: ... of clinical record entry')
                ))
                lines.append(' ' + _('Last modified:'))
                lines.append('  %s (%s)' % (
                        self._payload['modified_when'].timestamp(),
                        _('UNIX timestamp')
                ))
                format = '%Y-%m-%d (%b, %a) %H:%M:%S:%f %Z'
                lines.append('  %s %s' % (
                        gmTools.u_arrow2right,
                        self._payload['modified_when'].astimezone(tz = pydt.timezone.utc).strftime(format)
                ))
                lines.append('  %s %s (%s)' % (
                        gmTools.u_arrow2right,
                        self._payload['modified_when'].strftime(format),
                        _('local time')
                ))
                lines.append(' ' + _('Modified by: %s') % self._payload['modified_by'])
                lines.append(' %s PK: %s' % (
                        self._payload['src_table'],
                        self._payload['src_pk']
                ))
                lines.append(' ' + _('Encounter PK: %s') % self._payload['pk_encounter'])
                lines.append(' ' + _('Episode PK: %s') % self._payload['pk_episode'])
                lines.append(' ' + _('Health Issue PK: %s') % self._payload['pk_health_issue'])
                lines.append(' ' + _('Patient PK: %s') % self._payload['pk_patient'])
                return lines

        formatted_tech_info = property(fget = format_tech_info)

        #--------------------------------------------------------
        def format_header(self, eol=None):
                lines = []
                lines.append('%s (%s)' % (self.item_type_str, self.i18n_soap_cat))
                lines.append('')
                if self._payload['health_issue'] is None:
                        issue_info = gmTools.u_diameter
                else:
                        issue_info = '%s%s' % (
                                self._payload['health_issue'],
                                gmTools.bool2subst(self._payload['issue_active'], ' (' + _('active') + ')', ' (' + _('inactive') + ')', '')
                        )
                lines.append(_('Health issue: %s') % issue_info)
                if self._payload['episode'] is None:
                        episode_info = gmTools.u_diameter
                else:
                        episode_info = '%s%s' % (
                                self._payload['episode'],
                                gmTools.bool2subst(self._payload['episode_open'], ' (' +  _('open') + ')', ' (' +  _('closed') + ')', '')
                        )
                lines.append(_('Episode: %s') % episode_info)
                if self._payload['encounter_started'] is None:
                        enc_info = gmTools.u_diameter
                else:
                        enc_info = '%s - %s (%s)' % (
                                self._payload['encounter_started'].strftime('%Y %b %d  %H:%M'),
                                self._payload['encounter_last_affirmed'].strftime('%H:%M'),
                                self._payload['l10n_encounter_type']
                        )
                lines.append(_('Encounter: %s') % enc_info)
                lines.append(_('Event: %s') % self._payload['clin_when'].strftime('%Y %b %d  %H:%M'))
                if eol is None:
                        return lines

                return eol.join(lines)

        formatted_header = property(format_header)

        #--------------------------------------------------------
        def __get_item_type_str(self):
                try:
                        tmp = _MAP_generic_emr_item_table2type_str[self._payload['src_table']]
                except KeyError:
                        tmp = '[%s:%s]' % (
                                self._payload['src_table'],
                                self._payload['src_pk']
                        )
                return tmp

        item_type_str = property(__get_item_type_str)

        #--------------------------------------------------------
        def __get_i18n_soap_cat(self):
                return gmSoapDefs.soap_cat2l10n[self._payload['soap_cat']]

        i18n_soap_cat = property(__get_i18n_soap_cat)

        #--------------------------------------------------------
        def __get_specialized_item(self):
                item_class = _MAP_generic_emr_item_table2class[self._payload['src_table']]
                return item_class(aPK_obj = self._payload['src_pk'])

        specialized_item = property(__get_specialized_item)

        #--------------------------------------------------------
        def get_revision_history(self, link_obj:gmPG2._TLnkObj=None):
                schema, table = self._payload['src_table'].split('.', 1)
                hx = gmPG2.get_row_history (
                        link_obj = link_obj,
                        schema = schema,
                        table = table,
                        pk = self._payload['src_pk']
                )
                return hx

        #--------------------------------------------------------
        def get_formatted_revision_history(self, link_obj:gmPG2._TLnkObj=None):
                hx = self.get_revision_history(link_obj = link_obj)
                if not hx:
                        return ['%s (PK %s): %s' % (
                                self._payload['src_table'],
                                self._payload['src_pk'],
                                _('no previous versions')
                        )]

                lines = ['%s, PK %s: %s version(s) before current state (= v%s)' % (
                        self._payload['src_table'],
                        self._payload['src_pk'],
                        len(hx),
                        self._payload['row_version']
                )]
                # loop over _all_ rows because the audit table can have chaged over time :(
                column_labels = [ 'v%s (%s)' % (
                        h['row_version'], h['version_logged_when'].strftime('%Y %b %d %H:%M')
                ) for h in hx ]
                lines.extend (gmTools.dicts2table_columns (
                        hx,
                        left_margin = 1,
                        eol = None,
                        keys2ignore = ['src_row_pk_audit', 'version_logged_when', 'version_logged_why', 'version_logged_by', 'row_version', 'src_table_oid', 'pk_audit_trail'],
                        show_only_changes = True,
                        column_labels = column_labels,
                        date_format = '%Y %b %d %H:%M',
                        equality_value = gmTools.u_arrow2right
                ))
                return lines

Represents an entry in clin.v_emr_journal.

Call init from child classes like so:

    super().__init__(aPK_obj = aPK_obj, row = row, link_obj = link_obj)

Args

-----=

aPK_obj

retrieve data from backend

• an scalar value the ._cmd_fetch_payload WHERE condition must be a simple column: "… WHERE pk_col = %s"
• a dictionary of values the ._cmd_fetch_payload WHERE condition must consume the dictionary and produce a unique row

row

must hold the fields

• data: list of column values for the row selected by ._cmd_fetch_payload (as returned by cursor.fetchone() in the DB-API)
• pk_field: the name of the primary key column OR
• pk_obj: a dictionary suitable for being passed to cursor.execute and holding the primary key values, used for composite PKs
• for example:

  row = {
          'data': rows[0],
          'pk_field': 'pk_XXX (the PK column name)',
          'pk_obj': {'pk_col1': pk_col1_val, 'pk_col2': pk_col2_val}
  }
  
  rows = gmPG2.run_ro_queries(queries = [{'sql': cmd, 'args': args}])
  objects = [ cChildClass(row = {'data': r, 'pk_field': 'the PK column name'}) for r in rows ]
  

Ancestors

• cBusinessDBObject

Instance variables

prop formatted_header

Expand source code

def format_header(self, eol=None):
        lines = []
        lines.append('%s (%s)' % (self.item_type_str, self.i18n_soap_cat))
        lines.append('')
        if self._payload['health_issue'] is None:
                issue_info = gmTools.u_diameter
        else:
                issue_info = '%s%s' % (
                        self._payload['health_issue'],
                        gmTools.bool2subst(self._payload['issue_active'], ' (' + _('active') + ')', ' (' + _('inactive') + ')', '')
                )
        lines.append(_('Health issue: %s') % issue_info)
        if self._payload['episode'] is None:
                episode_info = gmTools.u_diameter
        else:
                episode_info = '%s%s' % (
                        self._payload['episode'],
                        gmTools.bool2subst(self._payload['episode_open'], ' (' +  _('open') + ')', ' (' +  _('closed') + ')', '')
                )
        lines.append(_('Episode: %s') % episode_info)
        if self._payload['encounter_started'] is None:
                enc_info = gmTools.u_diameter
        else:
                enc_info = '%s - %s (%s)' % (
                        self._payload['encounter_started'].strftime('%Y %b %d  %H:%M'),
                        self._payload['encounter_last_affirmed'].strftime('%H:%M'),
                        self._payload['l10n_encounter_type']
                )
        lines.append(_('Encounter: %s') % enc_info)
        lines.append(_('Event: %s') % self._payload['clin_when'].strftime('%Y %b %d  %H:%M'))
        if eol is None:
                return lines

        return eol.join(lines)

prop formatted_tech_info : list[str]

Expand source code

def format_tech_info(self) -> list[str]:
        lines = []
        lines.append('%s [%s]' % (self.item_type_str, self._payload['real_soap_cat']))
        lines.append(' ' + _('Table: %s##tx: EMR item source (database table, mostly)') % self._payload['src_table'])
        lines.append(' ' + _('Version: %s (%s)') % (
                self._payload['row_version'],
                _('Original##tx: ... of clinical record item') if (self._payload['row_version'] == 1) else _('Revision##tx: ... of clinical record entry')
        ))
        lines.append(' ' + _('Last modified:'))
        lines.append('  %s (%s)' % (
                self._payload['modified_when'].timestamp(),
                _('UNIX timestamp')
        ))
        format = '%Y-%m-%d (%b, %a) %H:%M:%S:%f %Z'
        lines.append('  %s %s' % (
                gmTools.u_arrow2right,
                self._payload['modified_when'].astimezone(tz = pydt.timezone.utc).strftime(format)
        ))
        lines.append('  %s %s (%s)' % (
                gmTools.u_arrow2right,
                self._payload['modified_when'].strftime(format),
                _('local time')
        ))
        lines.append(' ' + _('Modified by: %s') % self._payload['modified_by'])
        lines.append(' %s PK: %s' % (
                self._payload['src_table'],
                self._payload['src_pk']
        ))
        lines.append(' ' + _('Encounter PK: %s') % self._payload['pk_encounter'])
        lines.append(' ' + _('Episode PK: %s') % self._payload['pk_episode'])
        lines.append(' ' + _('Health Issue PK: %s') % self._payload['pk_health_issue'])
        lines.append(' ' + _('Patient PK: %s') % self._payload['pk_patient'])
        return lines

prop i18n_soap_cat

Expand source code

def __get_i18n_soap_cat(self):
        return gmSoapDefs.soap_cat2l10n[self._payload['soap_cat']]

prop item_type_str

Expand source code

def __get_item_type_str(self):
        try:
                tmp = _MAP_generic_emr_item_table2type_str[self._payload['src_table']]
        except KeyError:
                tmp = '[%s:%s]' % (
                        self._payload['src_table'],
                        self._payload['src_pk']
                )
        return tmp

prop specialized_item

Expand source code

def __get_specialized_item(self):
        item_class = _MAP_generic_emr_item_table2class[self._payload['src_table']]
        return item_class(aPK_obj = self._payload['src_pk'])

Methods

def format_header(self, eol=None)

Expand source code

def format_header(self, eol=None):
        lines = []
        lines.append('%s (%s)' % (self.item_type_str, self.i18n_soap_cat))
        lines.append('')
        if self._payload['health_issue'] is None:
                issue_info = gmTools.u_diameter
        else:
                issue_info = '%s%s' % (
                        self._payload['health_issue'],
                        gmTools.bool2subst(self._payload['issue_active'], ' (' + _('active') + ')', ' (' + _('inactive') + ')', '')
                )
        lines.append(_('Health issue: %s') % issue_info)
        if self._payload['episode'] is None:
                episode_info = gmTools.u_diameter
        else:
                episode_info = '%s%s' % (
                        self._payload['episode'],
                        gmTools.bool2subst(self._payload['episode_open'], ' (' +  _('open') + ')', ' (' +  _('closed') + ')', '')
                )
        lines.append(_('Episode: %s') % episode_info)
        if self._payload['encounter_started'] is None:
                enc_info = gmTools.u_diameter
        else:
                enc_info = '%s - %s (%s)' % (
                        self._payload['encounter_started'].strftime('%Y %b %d  %H:%M'),
                        self._payload['encounter_last_affirmed'].strftime('%H:%M'),
                        self._payload['l10n_encounter_type']
                )
        lines.append(_('Encounter: %s') % enc_info)
        lines.append(_('Event: %s') % self._payload['clin_when'].strftime('%Y %b %d  %H:%M'))
        if eol is None:
                return lines

        return eol.join(lines)

def format_tech_info(self) ‑> list[str]

Expand source code

def format_tech_info(self) -> list[str]:
        lines = []
        lines.append('%s [%s]' % (self.item_type_str, self._payload['real_soap_cat']))
        lines.append(' ' + _('Table: %s##tx: EMR item source (database table, mostly)') % self._payload['src_table'])
        lines.append(' ' + _('Version: %s (%s)') % (
                self._payload['row_version'],
                _('Original##tx: ... of clinical record item') if (self._payload['row_version'] == 1) else _('Revision##tx: ... of clinical record entry')
        ))
        lines.append(' ' + _('Last modified:'))
        lines.append('  %s (%s)' % (
                self._payload['modified_when'].timestamp(),
                _('UNIX timestamp')
        ))
        format = '%Y-%m-%d (%b, %a) %H:%M:%S:%f %Z'
        lines.append('  %s %s' % (
                gmTools.u_arrow2right,
                self._payload['modified_when'].astimezone(tz = pydt.timezone.utc).strftime(format)
        ))
        lines.append('  %s %s (%s)' % (
                gmTools.u_arrow2right,
                self._payload['modified_when'].strftime(format),
                _('local time')
        ))
        lines.append(' ' + _('Modified by: %s') % self._payload['modified_by'])
        lines.append(' %s PK: %s' % (
                self._payload['src_table'],
                self._payload['src_pk']
        ))
        lines.append(' ' + _('Encounter PK: %s') % self._payload['pk_encounter'])
        lines.append(' ' + _('Episode PK: %s') % self._payload['pk_episode'])
        lines.append(' ' + _('Health Issue PK: %s') % self._payload['pk_health_issue'])
        lines.append(' ' + _('Patient PK: %s') % self._payload['pk_patient'])
        return lines

def get_formatted_revision_history(self,
link_obj: psycopg2.extras.DictConnection | psycopg2.extras.DictCursor | None = None)

Expand source code

def get_formatted_revision_history(self, link_obj:gmPG2._TLnkObj=None):
        hx = self.get_revision_history(link_obj = link_obj)
        if not hx:
                return ['%s (PK %s): %s' % (
                        self._payload['src_table'],
                        self._payload['src_pk'],
                        _('no previous versions')
                )]

        lines = ['%s, PK %s: %s version(s) before current state (= v%s)' % (
                self._payload['src_table'],
                self._payload['src_pk'],
                len(hx),
                self._payload['row_version']
        )]
        # loop over _all_ rows because the audit table can have chaged over time :(
        column_labels = [ 'v%s (%s)' % (
                h['row_version'], h['version_logged_when'].strftime('%Y %b %d %H:%M')
        ) for h in hx ]
        lines.extend (gmTools.dicts2table_columns (
                hx,
                left_margin = 1,
                eol = None,
                keys2ignore = ['src_row_pk_audit', 'version_logged_when', 'version_logged_why', 'version_logged_by', 'row_version', 'src_table_oid', 'pk_audit_trail'],
                show_only_changes = True,
                column_labels = column_labels,
                date_format = '%Y %b %d %H:%M',
                equality_value = gmTools.u_arrow2right
        ))
        return lines

def get_revision_history(self,
link_obj: psycopg2.extras.DictConnection | psycopg2.extras.DictCursor | None = None)

Expand source code

def get_revision_history(self, link_obj:gmPG2._TLnkObj=None):
        schema, table = self._payload['src_table'].split('.', 1)
        hx = gmPG2.get_row_history (
                link_obj = link_obj,
                schema = schema,
                table = table,
                pk = self._payload['src_pk']
        )
        return hx

Inherited members

• cBusinessDBObject:
  • fields_as_dict
  • format
  • get_fields
  • get_patient
  • get_updatable_fields
  • is_modified
  • keys
  • patient_pk
  • refetch_payload
  • same_payload
  • save
  • save_payload
  • staff_id