Add type hints for ExpectQ methods

README.md

@@ -1,3 +1,20 @@
 # vistassh-py
 
 Python-based web interface for VistA roll-and-scroll terminal
+
+## Getting started
+
+You will need:
+- Python 3.10+ & pip
+- Node.js & npm
+
+To install the dependencies and build the frontend, perform the following steps:
+
+```bash
+pip install -r requirements.txt
+cd frontend
+npm install
+npm run build
+```
+
+To run, invoke `main.py`.

autoproc.py

@@ -9,7 +9,7 @@ import contextlib
 import logging
 from collections import namedtuple
 
-from typing import Optional, Union, Sequence, NamedTuple, Callable
+from typing import Any, Optional, Union, Sequence, NamedTuple, Callable, AsyncGenerator
 
 logger = logging.getLogger(__name__)
 
@@ -28,7 +28,7 @@ class ExpectQ(object):
 		"""Clear or restore buffer"""
 		self.buffer = buffer
 	clear = reset
-	async def prompts(self, endl: str='\r\n', timeout_settle: Optional[float]=None, throw: bool=False):
+	async def prompts(self, endl: str='\r\n', timeout_settle: Optional[float]=None, throw: bool=False) -> AsyncGenerator[tuple[Optional[str], Optional[int]], None]:
 		len_endl = len(endl)
 		while True:
 			if (pos := self.buffer.rfind(endl)) >= 0:
@@ -43,7 +43,7 @@ class ExpectQ(object):
 					if throw:
 						raise
 					yield None, None
-	async def promptmatches(self, *mappings: Union[str, re.Pattern, tuple, list], endl: str='\r\n', timeout_settle: Optional[float]=None, throw: bool=False):
+	async def promptmatches(self, *mappings: Union[str, re.Pattern, tuple, list], endl: str='\r\n', timeout_settle: Optional[float]=None, throw: bool=False) -> AsyncGenerator[tuple[Optional[ExpectMatch], Any], Optional[bool]]:
 		for i, mapping in enumerate(mappings):
 			try:
 				match mapping:

ext_discovery.py

@@ -26,5 +26,11 @@ async def cmd_listclinics(proc):
 				proc.sendline('^')
 				break
 		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: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		expect.clear()

ext_lab.py

@@ -74,7 +74,13 @@ async def cmd_reports(proc, mrn, alpha, omega):
 					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: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		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)]

ext_measurement.py

@@ -39,7 +39,13 @@ async def cmd_entries(proc, mrn, alpha, omega):
 			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: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		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')

ext_note.py

@@ -31,15 +31,24 @@ async def cmd_reports(proc, mrn, alpha, omega):
 		proc.sendline(util.vista_strftime(omega))
 		assert await expect.endswith('\r\n                 Thru: ')
 		proc.sendline(util.vista_strftime(alpha))
-		assert await expect.endswith('\r\nDo you want WORK copies or CHART copies? CHART// ')
+		found = True
-		proc.sendline()	# default CHART
+		match await expect.endswith('\r\nDo you want WORK copies or CHART copies? CHART// ', '\r\nPrint Notes Beginning: '):
-		if await expect.endswith('\r\nDo you want to start each note on a new page? NO// '):
+			case autoproc.ExpectMatch(index=0):
-			proc.sendline()	# default NO
+				proc.sendline()	# default CHART
-		assert await expect.endswith('\r\nDEVICE: HOME// ')
+				if await expect.endswith('\r\nDo you want to start each note on a new page? NO// '):
-		proc.sendline('HOME;;1023')
+					proc.sendline()	# default NO
-		assert await expect.earliest(' HOME(CRT)\r\n')
+				assert await expect.endswith('\r\nDEVICE: HOME// ')
+				proc.sendline('HOME;;1023')
+				assert await expect.earliest(' HOME(CRT)\r\n')
+			case autoproc.ExpectMatch(index=1):
+				proc.sendline('^')
+				assert await expect.endswith('\r\nSelect PATIENT NAME: ')
+				proc.sendline('^')
+				assert await expect.endswith('\r\nSelect Progress Notes Print Options Option: ')
+				found = False
+			case _: assert False
 		pages = []
-		while True:
+		while found:
 			match m_delimiter := await expect.endswith('\r\nType <Enter> to continue or \'^\' to exit: ', '\r\nSelect PATIENT NAME: '):
 				case autoproc.ExpectMatch(index=0, before=before):
 					if isnew(before) and len(pages) > 0:
@@ -59,5 +68,11 @@ async def cmd_reports(proc, mrn, alpha, omega):
 				case _: assert False
 		proc.sendline('^')
 		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: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		expect.clear()

ext_order.py

@@ -77,7 +77,13 @@ async def cmd_entries(proc, mrn, alpha, omega):
 					break
 				case _: assert False
 		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: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		expect.clear()
 		prev = None
 		for m in re.finditer(r'\b\d{2}/\d{2}/\d{2}.*?\r\n\r\n', '\r\n'.join(pages).replace('\x1b[1m', '').replace('\x1b[m', ''), re.DOTALL):

ext_patient.py

@@ -25,7 +25,13 @@ async def cmd_lookup_patient(proc, query):
 				res.append(prompt.before[:-24] if prompt.index == 0 and prompt.before.endswith('\r\nENTER \'^\' TO STOP, OR \r\n') else prompt.before)
 				if 0 < prompt.index < 4:
 					single = True
-		assert await expect.endswith('\r\nSelect Patient Information and OE/RR Option: ', '\r\nSelect Patient Information and OE/RR  <TEST ACCOUNT> Option: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		expect.clear()
 		if single:
 			return [re.search(r'[ ]{2}(?P<name>.+?)[ ]{2}(?:\((?P<alias>[^\)]*?)\))?[ ]{6}(?P<dob>\S+)[ ]{4}(?P<ssn>\S+(?:P \*\*Pseudo SSN\*\*)?)[ ]{5}(?P<yesno>\S+)[ ]{5}(?P<type>.+?)[ ]{6}(?P<no>[^\r\n]*)', res[0].replace('\r\n', '', 1)).groupdict()]
@@ -65,6 +71,12 @@ async def cmd_lookup_patient_ordinal(proc, query, ordinal, force=False):
 					proc.sendline(response)
 				case autoproc.ExpectMatch(index=5):
 					proc.sendline()
-		assert await expect.endswith('\r\nSelect Patient Information and OE/RR Option: ', '\r\nSelect Patient Information and OE/RR  <TEST ACCOUNT> Option: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		expect.clear()
 		return re.sub(r'\r\n\r\n(?:[^\r\n;]+);(?:\([^\)]*?\))? (?:\d+ )?(?:\d{3}-\d{2}-\d{4}P?) (?:[^\r\n]+?)[ ]*?(\r\n={10,}\r\n)\r\n', r'\1', '\r\n'.join(res))

ext_rcrs.py

@@ -58,7 +58,13 @@ async def cmd_patients(proc, alpha, omega):
 				summary.extend({k.strip(): v.strip() for k, v in row.groupdict().items()} for row in re.finditer(r'(?P<last5>[A-Z]\d{4}) (?P<name>[^\r\n]{30}) (?P<uid>[^ \r\n]+) (?P<primarySite>[^ \r\n]+) (?P<dateOfDiagnosis>\d{2}/\d{2}/\d{4})  (?P<dateCaseLastChanged>\d{2}/\d{2}/\d{4})', prompt.before))
 			elif prompt.index == 4:
 				break
-		assert await expect.endswith('\r\nSelect Patient Information and OE/RR Option: ', '\r\nSelect Patient Information and OE/RR  <TEST ACCOUNT> Option: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		expect.clear()
 		for item in parse_xml_rcrs(doc_rcrs, summary):
 			yield item

ext_scheduling.py

@@ -63,7 +63,13 @@ async def cmd_appointments(proc, clinics='NPT-HEM/ONC ATTENDING', date='T', stor
 				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: ')
+		async for prompt, response in expect.promptmatches((
+				(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+				('Select Patient Information and OE/RR Option: ',					None, True),
+				('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+			), throw=True):
+			if prompt.index == 0:
+				proc.sendline(response)
 		expect.clear()
 
 async def vista_appointment_clinics(proc, expect):

ext_session.py

@@ -23,6 +23,13 @@ async def task_smartcard(proc, config: Optional[configparser.ConfigParser]=None)
 				if certificate:
 					config.set('auth', 'certificate', certificate)
 				proc.create_task(task_keepalive(proc, True), name='@task:keepalive')
+				async for prompt, response in expect.promptmatches((
+						(re.compile(r' Press \'RETURN\' to continue, \'\^\' to stop: $'),	None),
+						('Select Patient Information and OE/RR Option: ',					None, True),
+						('Select Patient Information and OE/RR  <TEST ACCOUNT> Option: ',	None, True),
+					), throw=True):
+					if prompt.index == 0:
+						proc.sendline(response)
 				return True
 
 async def task_keepalive(proc, suppress=False):

frontend/src/lib/backend.js

@@ -1,7 +1,8 @@
 export async function get_api_appointments({ fetch, clinics = [], date = 'T' } = {}) {
 	if(clinics.constructor === Array) clinics = clinics.map(x => x.replace(/^\s+|\s+$/g, '').replace(/\s+/, ' ')).filter(x => x).join('^').replace(/\//g, '|');
 	else clinics = clinics.replace(/^\s+|\s+$/g, '').replace(/\s+/, ' ').replace(/\//g, '|');
-	return await (await (fetch || window.fetch)('/api/appointments/' + clinics + '/' + date)).json();
+	if(clinics) return await (await (fetch || window.fetch)('/api/appointments/' + clinics + '/' + date)).json();
+	else return [];
 }
 
 export async function get_api_lookup({ fetch, query, ordinal, force = false } = {}) {

frontend/src/routes/appointments/+page.js

@@ -2,7 +2,12 @@ import { get_api_appointments } from '$lib/backend.js';
 
 /** @type {import('./$types').PageLoad} */
 export async function load({ params, fetch }) {
-	let clinics = await (await fetch('/api/config/user/clinics')).json();
+	let clinics = [];
+	try {
+		clinics = await (await fetch('/api/config/user/clinics')).json();
+	} catch(ex) {
+		console.error(ex, ex.stack);
+	}
 	let appointments = await get_api_appointments({ fetch, clinics, date: 'T' });
 	appointments.sort((a, b) => a.time_scheduled < b.time_scheduled ? -1 : a.time_scheduled > b.time_scheduled ? 1 : 0);
 	return {

frontend/src/routes/clinics/+page.js

@@ -2,8 +2,12 @@
 export async function load({ params, fetch }) {
 	let clinics = await (await fetch('/api/clinic/list')).json();
 	clinics.reduce((acc, item) => (acc[item.name] = item, acc), clinics);
-	let selection = await (await fetch('/api/config/user/clinics')).json();
+	try {
-	selection.forEach(x => clinics[x] ? clinics[x].active = true : false);
+		let selection = await (await fetch('/api/config/user/clinics')).json();
+		selection.forEach(x => clinics[x] ? clinics[x].active = true : false);
+	} catch(ex) {
+		console.error(ex, ex.stack);
+	}
 	return {
 		clinics
 	};