68 lines
3.3 KiB
Python
68 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import re
|
|
import datetime
|
|
import util
|
|
import autoproc
|
|
|
|
units = {
|
|
'P': 'bpm',
|
|
'R': 'bpm',
|
|
'B/P': 'mmHg',
|
|
'Body Mass Index': 'kg/m²'
|
|
}
|
|
|
|
local_tzinfo = datetime.datetime.now(datetime.timezone.utc).astimezone().tzinfo
|
|
async def cmd_entries(proc, mrn, alpha, omega):
|
|
"""Fetch measurements"""
|
|
async with proc.sendline, autoproc.expect_async(proc) as expect:
|
|
proc.sendline('^Results Reporting Menu')
|
|
async for prompt, response in expect.promptmatches(
|
|
(
|
|
(' Press return to continue ', None),
|
|
('Select Patient(s): ', mrn, True),
|
|
('Select Patient: ', mrn, True),
|
|
),
|
|
('Select Item(s): ', '7'), # Vitals Cumulative Report
|
|
('Start Date [Time]: T// ', util.vista_strftime(alpha)),
|
|
(re.compile(r'^Ending Date \[Time\] \(inclusive\): (.*?)// $'), util.vista_strftime(omega)),
|
|
('DEVICE: HOME// ', 'HOME;;1023'),
|
|
timeout_settle=2, throw=True):
|
|
proc.sendline(response)
|
|
assert await expect.earliest(' HOME(CRT)\r\n')
|
|
pages = []
|
|
async for prompt, response in expect.promptmatches((
|
|
('Press return to continue "^" to escape ', None),
|
|
('Press RETURN to continue or \'^\' to exit: ', None),
|
|
('Select Clinician Menu Option: ', None, True),
|
|
), throw=True):
|
|
proc.sendline(response)
|
|
if prompt.index == 0 or prompt.index == 1:
|
|
pages.append(re.sub(r'^\x1b\[H\x1b\[J\x1b\[2J\x1b\[H\r\n[^\r\n]+? Cumulative Vitals\/Measurements Report[ ]+Page \d+\r\n\r\n-{10,}\r\n(?:\d{2}\/\d{2}\/\d{2} \(continued\)\r\n\r\n)?|\r\n\r\n\*\*\*[^\r\n]+\r\n\r\n[^\r\n]+?VAF 10-7987j\r\nUnit:[^\r\n]+\r\nDivision:[^\r\n]+(?:\r\n)?$', '', prompt.before))
|
|
assert await expect.endswith('\r\nSelect Patient Information and OE/RR Option: ', '\r\nSelect Patient Information and OE/RR <TEST ACCOUNT> Option: ')
|
|
expect.clear()
|
|
for m_date in re.finditer(r'^(?P<date>\d{2}\/\d{2}\/\d{2})\r\n(?P<body>.*?\r\n)(?:(?=\d{2}\/)|\r\n|$)', '\r\n'.join(pages), re.DOTALL|re.MULTILINE):
|
|
g_date = m_date.group('date')
|
|
for m_time in re.finditer(r'^(?P<time>\d{2}:\d{2})\r\n(?P<body>.*?\r\n)(?:(?=\d{2}:)|\r\n|$)', m_date.group('body'), re.DOTALL|re.MULTILINE):
|
|
dt = datetime.datetime.strptime(g_date + ' ' + m_time.group('time'), '%m/%d/%y %H:%M').replace(tzinfo=local_tzinfo)
|
|
for m_entry in re.finditer(r'^[ ]{4}\b(?P<name>[^:]+):[ ]+(?P<value_american>[\d\.\/%]+)(?P<flag>\*)?(?: (?P<unit_american>\w+))?(?: \((?P<value_metric>\d\S*) (?P<unit_metric>\S+)\))?(?: \((?P<method>[^\)\r\n]+)\))?(?P<comment>(?:\r\n[ ]{9}\S[^\r\n]*)*)', m_time.group('body'), re.DOTALL|re.MULTILINE):
|
|
m_entry = m_entry.groupdict()
|
|
m_entry['time'] = dt
|
|
if m_entry['value_american'].endswith('%'):
|
|
m_entry['value_american'] = m_entry['value_american'][:-1]
|
|
m_entry['unit_american'] = '%'
|
|
if m_entry['value_metric']:
|
|
m_entry['value'] = m_entry['value_metric']
|
|
m_entry['unit'] = m_entry['unit_metric']
|
|
else:
|
|
m_entry['value'] = m_entry['value_american']
|
|
m_entry['unit'] = m_entry['unit_american']
|
|
if m_entry['unit'] is None:
|
|
m_entry['unit'] = units.get(m_entry['name'])
|
|
m_entry['comment'] = re.sub(r'\s+', ' ', m_entry['comment'].strip()) if m_entry['comment'] else None
|
|
yield m_entry
|
|
if m_entry['name'] == 'B/P':
|
|
a, b = m_entry['value'] .split('/', 1)
|
|
yield dict(m_entry, name='SBP', value=a)
|
|
yield dict(m_entry, name='DBP', value=b)
|