108 lines
5.2 KiB
Python
108 lines
5.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
|
||
|
|
"""Appointments"""
|
||
|
|
|
||
|
|
import re
|
||
|
|
import datetime
|
||
|
|
import autoproc
|
||
|
|
|
||
|
|
import logging
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
local_tzinfo = datetime.datetime.now(datetime.timezone.utc).astimezone().tzinfo
|
||
|
|
async def cmd_appointments(proc, clinics='NPT-HEM/ONC ATTENDING', date='T', storeview=None):
|
||
|
|
"""List appointments"""
|
||
|
|
async with proc.sendline, autoproc.expect_async(proc) as expect:
|
||
|
|
proc.sendline('^Appointment List')
|
||
|
|
assert await expect.endswith('\r\nSelect division: ALL// ')
|
||
|
|
proc.sendline() # default ALL
|
||
|
|
assert await expect.endswith('\r\nCount, Non Count, or Both: C//')
|
||
|
|
proc.sendline('Both')
|
||
|
|
assert await expect.endswith('\r\nSelect clinic: ALL// ')
|
||
|
|
for line in clinics.strip().split('^'):
|
||
|
|
proc.sendline(line)
|
||
|
|
match await expect.endswith('\x07 ??\r\nSelect clinic: ALL// ', '\r\nSelect another clinic: ', '\r\nSelect clinic: ALL// ', re.compile(r'\r\nCHOOSE \d+-\d+: $'), '\r\n ...OK? Yes// '):
|
||
|
|
case autoproc.ExpectMatch(index=0):
|
||
|
|
raise RuntimeError
|
||
|
|
case autoproc.ExpectMatch(index=3):
|
||
|
|
proc.sendline('1') # choose first option
|
||
|
|
case autoproc.ExpectMatch(index=4):
|
||
|
|
proc.sendline() # accept confirmation
|
||
|
|
proc.sendline() # finish clinic selection with blank entry
|
||
|
|
assert await expect.endswith('\r\nFor date: ')
|
||
|
|
proc.sendline(date)
|
||
|
|
if storeview:
|
||
|
|
assert (m := await expect.endswith(re.compile(r'\((.*?)\)', flags=re.MULTILINE|re.DOTALL)))
|
||
|
|
date = datetime.datetime.strptime(m.group(1), '%b %d, %Y').date()
|
||
|
|
storeview.delete('date_scheduled=?', (str(date),))
|
||
|
|
assert await expect.endswith(re.compile(r'Include Primary Care assignment information in the output\? NO// ', flags=re.MULTILINE|re.DOTALL))
|
||
|
|
proc.sendline() # default NO
|
||
|
|
assert await expect.endswith('\r\nNumber of copies: 1// ')
|
||
|
|
proc.sendline() # default 1
|
||
|
|
assert await expect.endswith('\r\nDEVICE: HOME// ')
|
||
|
|
proc.sendline() # default HOME
|
||
|
|
assert await expect.earliest('Right Margin: 80// ')
|
||
|
|
proc.sendline() # default 80 (maximum 256)
|
||
|
|
async for clinic in vista_appointment_clinics(proc, expect):
|
||
|
|
body = re.split(r'(?:^|\n)([ \d]\d:\d{2} [AP]M)\n\n', clinic['body'])
|
||
|
|
for i in range(1, len(body), 2):
|
||
|
|
item = clinic.copy()
|
||
|
|
del item['body']
|
||
|
|
item['time_scheduled'] = datetime.datetime.combine(item['date_scheduled'], datetime.datetime.strptime(body[i].strip(), '%I:%M %p').time()).replace(tzinfo=local_tzinfo)
|
||
|
|
detail = re.sub(r'\r\n(\s{0,9}\S)', r'\1', body[i + 1]) # collapse hard wrap
|
||
|
|
name, detail = detail.split('\n', 1)
|
||
|
|
item['flag'] = name[:10].strip()
|
||
|
|
assert (match := re.search(r'^(?P<patient_name>\w.*?)\s+(?P<patient_last4>\d{4}).*?$', name[10:]))
|
||
|
|
item.update(match.groupdict())
|
||
|
|
match = re.search(r'^\s{41}\b(?P<location_ward>.*?)\n\s{41}\b(?P<location_bed>.*?)$', detail, re.MULTILINE)
|
||
|
|
if match:
|
||
|
|
item.update(match.groupdict())
|
||
|
|
match = re.search(r'^\s{10}Phone #: (?P<patient_phone>.*)$', detail, re.MULTILINE)
|
||
|
|
if match:
|
||
|
|
item.update(match.groupdict())
|
||
|
|
item['comment'] = '\r\n'.join(m.group(1) for m in re.finditer(r'^\s{15}(\w.*?)$', detail, re.MULTILINE))
|
||
|
|
yield item
|
||
|
|
proc.sendline('^Patient information AND OE/RR')
|
||
|
|
assert await expect.endswith('\r\nSelect Patient Information and OE/RR Option: ', '\r\nSelect Patient Information and OE/RR <TEST ACCOUNT> Option: ')
|
||
|
|
expect.clear()
|
||
|
|
|
||
|
|
async def vista_appointment_clinics(proc, expect):
|
||
|
|
"""List appointments by clinic, collecting all pages"""
|
||
|
|
item = None
|
||
|
|
while True:
|
||
|
|
m = await expect.earliest('\x07')
|
||
|
|
if m:
|
||
|
|
try:
|
||
|
|
head, body = re.split(r'\n_{10,}\n', m.before.replace('\r\n', '\n'))
|
||
|
|
except ValueError:
|
||
|
|
logger.warning('invalid page %r'%m.before)
|
||
|
|
continue
|
||
|
|
if body.strip() not in {'No activity found for this clinic date!', 'Clinic cancelled for this date!'}:
|
||
|
|
assert (m1 := re.search(r'^Date printed: (?P<time_printed>.*?)\s+Page: (?P<print_page>\d+)$', head, re.MULTILINE))
|
||
|
|
assert (m2 := re.search(r'Appointments for (?P<clinic>.+) clinic on (?P<day_scheduled>[A-Z]+) (?P<date_scheduled>(?:[A-Z]+) (?:[0-9]+), (?:[0-9]+))', head))
|
||
|
|
if int(m1.group('print_page')) > 1:
|
||
|
|
# next page of same report
|
||
|
|
assert datetime.datetime.strptime(m1.group('time_printed'), '%b %d,%Y@%H:%M') == item['time_printed']
|
||
|
|
assert m2.group('clinic') == item['clinic']
|
||
|
|
assert m2.group('day_scheduled') == item['day_scheduled']
|
||
|
|
assert datetime.datetime.strptime(m2.group('date_scheduled'), '%b %d, %Y').date() == item['date_scheduled']
|
||
|
|
item['body'] += '\r\n' + body.rstrip('\r\n') # concatenate report bodies
|
||
|
|
else:
|
||
|
|
# new report
|
||
|
|
if item is not None:
|
||
|
|
yield item
|
||
|
|
item = {
|
||
|
|
'time_printed': datetime.datetime.strptime(m1.group('time_printed'), '%b %d,%Y@%H:%M'),
|
||
|
|
'clinic': m2.group('clinic'),
|
||
|
|
'day_scheduled': m2.group('day_scheduled'),
|
||
|
|
'date_scheduled': datetime.datetime.strptime(m2.group('date_scheduled'), '%b %d, %Y').date(),
|
||
|
|
'body': body.strip('\r\n')
|
||
|
|
}
|
||
|
|
m1 = re.search(r'Appointments for (?P<clinic>.+) clinic on (?P<day>[A-Z]+) (?P<date>(?:[A-Z]+) (?:[0-9]+), (?:[0-9]+))', head)
|
||
|
|
assert await expect.endswith('\r\nPress return to continue or "^" to escape ')
|
||
|
|
proc.sendline()
|
||
|
|
else: # leftovers
|
||
|
|
if item is not None:
|
||
|
|
yield item
|
||
|
|
break
|