84 lines
4.8 KiB
Python
84 lines
4.8 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
|
||
|
|
import re
|
||
|
|
import datetime
|
||
|
|
import util
|
||
|
|
import autoproc
|
||
|
|
|
||
|
|
import logging
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
local_tzinfo = datetime.datetime.now(datetime.timezone.utc).astimezone().tzinfo
|
||
|
|
def parse(text):
|
||
|
|
data = { 'body': text.rstrip() }
|
||
|
|
if match := re.search(r'\bProvider: \b(?P<practitioner>.*?)\r\n', text):
|
||
|
|
data.update(match.groupdict())
|
||
|
|
if match := re.search(r'\bSpecimen: \b(?P<specimen>.*?)\r\n', text):
|
||
|
|
data.update(match.groupdict())
|
||
|
|
if match := re.search(r'\bAccession \[UID\]: \b(?P<accession>.*?) \[(?P<uid>.*?)\]\r\n', text):
|
||
|
|
data.update(match.groupdict())
|
||
|
|
if match := re.search(r'\bReport Released: \b(?P<time_released>.*?)\r\n', text):
|
||
|
|
value = match.group(1)
|
||
|
|
data['time_released'] = datetime.datetime.strptime(value, '%b %d, %Y@%H:%M').replace(tzinfo=local_tzinfo) if '@' in value else datetime.datetime.strptime(value, '%b %d, %Y').replace(tzinfo=local_tzinfo)
|
||
|
|
if match := re.search(r'\bSpecimen Collection Date: \b(?P<time_collected>.*?)\r\n', text):
|
||
|
|
value = match.group(1)
|
||
|
|
data['time_collected'] = datetime.datetime.strptime(value, '%b %d, %Y@%H:%M').replace(tzinfo=local_tzinfo) if '@' in value else datetime.datetime.strptime(value, '%b %d, %Y').replace(tzinfo=local_tzinfo)
|
||
|
|
if match := re.match(r'\s+----(.*?)----\s+page \d+\r\n', text):
|
||
|
|
data['title'] = match.group(1)
|
||
|
|
if match := re.search(r'\bAccession \[UID\]: (?P<accession>.*?) \[(?P<uid>.*?)\]\s+Received: (?P<time_received>.*?)\r\n', text):
|
||
|
|
data.update(match.groupdict())
|
||
|
|
data['time_received'] = datetime.datetime.strptime(data['time_received'], '%b %d, %Y@%H:%M').replace(tzinfo=local_tzinfo) if '@' in data['time_received'] else datetime.datetime.strptime(data['time_received'], '%b %d, %Y').replace(tzinfo=local_tzinfo)
|
||
|
|
if match := re.search(r'\bReport Completed: \b(?P<time_completed>.*?)\r\n', text):
|
||
|
|
value = match.group(1)
|
||
|
|
data['time_completed'] = datetime.datetime.strptime(value, '%b %d, %Y@%H:%M').replace(tzinfo=local_tzinfo) if '@' in value else datetime.datetime.strptime(value, '%b %d, %Y').replace(tzinfo=local_tzinfo)
|
||
|
|
if match := re.search(r'\bCollection sample: (?P<specimen>.*?)\s+Collection date: (?P<time_collected>.*?)\r\n', text):
|
||
|
|
data.update(match.groupdict())
|
||
|
|
data['time_collected'] = datetime.datetime.strptime(data['time_collected'], '%b %d, %Y %H:%M').replace(tzinfo=local_tzinfo) if ':' in data['time_collected'] else datetime.datetime.strptime(data['time_collected'], '%b %d, %Y').replace(tzinfo=local_tzinfo)
|
||
|
|
return data
|
||
|
|
|
||
|
|
async def cmd_reports(proc, mrn, alpha, omega):
|
||
|
|
"""Fetch lab reports"""
|
||
|
|
async with proc.sendline, autoproc.expect_async(proc) as expect:
|
||
|
|
proc.sendline('^Laboratory Menu')
|
||
|
|
async for prompt, response in expect.promptmatches(
|
||
|
|
('Select HOSPITAL LOCATION NAME: ', None),
|
||
|
|
('Select Laboratory Menu Option: ', '13'), # Interim report
|
||
|
|
('Select Patient Name: ', mrn),
|
||
|
|
(
|
||
|
|
('Do you wish to continue with this patient [Yes/No]? ', 'Yes'),
|
||
|
|
('Date to START with: TODAY//', util.vista_strftime(omega), True),
|
||
|
|
),
|
||
|
|
('Date to END with: T-7//', util.vista_strftime(alpha)),
|
||
|
|
('Print address page? NO// ', None), # default NO
|
||
|
|
(
|
||
|
|
('Do you want to start each note on a new page? NO// ', None), # default NO
|
||
|
|
('DEVICE: HOME// ', 'HOME;90;1023', True),
|
||
|
|
),
|
||
|
|
throw=True):
|
||
|
|
proc.sendline(response)
|
||
|
|
assert await expect.earliest(' HOME(CRT)\r\n')
|
||
|
|
pages = []
|
||
|
|
async for prompt, response in expect.promptmatches((
|
||
|
|
(re.compile(r' PRESS \'\^\' TO STOP $'), None),
|
||
|
|
(re.compile(r' \'\^\' TO STOP$'), None),
|
||
|
|
('Select Patient Name: ', None),
|
||
|
|
('Select Laboratory Menu Option: ', None, True),
|
||
|
|
), throw=True):
|
||
|
|
proc.sendline(response)
|
||
|
|
match prompt:
|
||
|
|
case autoproc.ExpectMatch(index=(0|1), before=before):
|
||
|
|
if left := re.match(r'(?:\x1b\[H\x1b\[J\x1b\[2J\x1b\[H\r\n\r\n.+?[ ]{2,}Report date: .+?\r\n Pat ID: \d{3}-\d{2}-\d{4}[ ]{2,}SEX: \w[ ]{2,}DOB: .+?[ ]{2,}LOC: .+?\r\n)|(?:\r\n\x1b\[H\x1b\[J\x1b\[2J\x1b\[H\r\n.+?[ ]{2,}\d{3}-\d{2}-\d{4}[ ]{2,}AGE: \d+[^\r\n]*?\r\n)|(?:\r[ ]+\r\r\n.+?[ ]{2,}\d{3}-\d{2}-\d{4}[ ]{2,}AGE: \d+[ ]{2,}.+?\r\n)', before):
|
||
|
|
pages.append(before[len(left.group(0)):])
|
||
|
|
elif re.match(r'(?:\r\n)+.+?[ ]{2,}\d{3}-\d{2}-\d{4}[ ]{2,}.+?[ ]+$', before) or re.match(r'^(?:\r\n)+$', before):
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
print(repr(before))
|
||
|
|
assert False
|
||
|
|
assert await expect.endswith('\r\nSelect Patient Information and OE/RR Option: ', '\r\nSelect Patient Information and OE/RR <TEST ACCOUNT> Option: ')
|
||
|
|
expect.clear()
|
||
|
|
text = re.sub(r'\r\n\s+>> CONTINUATION OF .+? <<(?:(?:\r\n)|(?:\s+page \d+))', '', '\r\n'.join(pages))
|
||
|
|
positions = [m.start() for m in re.finditer(r'(?:(?:[ ]+----MICROBIOLOGY----[ ]+page \d+\r\n\r\n)|(?:[ ]+))Reporting Lab:', text)]
|
||
|
|
positions.append(len(text))
|
||
|
|
for i in range(len(positions) - 1):
|
||
|
|
yield parse(text[positions[i]:positions[i + 1]])
|