Command system including live reading and writing of the process shown in the terminal and proper exit code plus output buffer return

pytinytex/__init__.py

@@ -1,24 +1,27 @@
 import sys
 import os
-import subprocess
+import asyncio
 import platform
 
 from .tinytex_download import download_tinytex, DEFAULT_TARGET_FOLDER # noqa
 
 # Global cache
 __tinytex_path = None
 
-def update(package="-all"):
-	path = get_tinytex_path()
-	try:
-		code, stdout,stderr = _run_tlmgr_command(["update", package], path, False)
-		return True
-	except RuntimeError:
-		raise
-		return False
+def update(package="-all", machine_readable=False):
+	path = get_tinytex_distribution_path()
+	return _run_tlmgr_command(["update", package], path, machine_readable=machine_readable)
+
+def shell():
+	path = get_tinytex_distribution_path()
+	return _run_tlmgr_command(["shell"], path, machine_readable=False, interactive=True)
 
+def help(*args, **kwargs):
+	path = get_tinytex_distribution_path()
+	return _run_tlmgr_command(["help"], path, *args, **kwargs)
 
-def get_tinytex_path(base=None):
+
+def get_tinytex_distribution_path(base=None):
 	if __tinytex_path:
 		return __tinytex_path
 	path_to_resolve = DEFAULT_TARGET_FOLDER
@@ -30,85 +33,145 @@ def get_tinytex_path(base=None):
 	ensure_tinytex_installed(path_to_resolve)
 	return __tinytex_path
 
+def get_tlmgr_path():
+	return _resolve_path(get_tinytex_distribution_path())
+
+def get_tlmgr_executable():
+	if platform.system() == "Windows":
+		return os.path.join(get_tlmgr_path(), "tlmgr.bat")
+	else:
+		return os.path.join(get_tlmgr_path(), "tlmgr")
+
 def get_pdf_latex_engine():
 	if platform.system() == "Windows":
-		return os.path.join(get_tinytex_path(), "pdflatex.exe")
+		return os.path.join(get_tlmgr_path(), "pdflatex.exe")
 	else:
-		return os.path.join(get_tinytex_path(), "pdflatex")
+		return os.path.join(get_tlmgr_path(), "pdflatex")
 
 
 def ensure_tinytex_installed(path=None):
 	global __tinytex_path
 	if not path:
 		path = __tinytex_path
-	__tinytex_path = _resolve_path(path)
-	return True
+	if _resolve_path(str(path)):
+		__tinytex_path = path
+		os.environ["TEXMFCNF"] = str(__tinytex_path)
+		return True
+
 
 def _resolve_path(path):
 	try:
-		if _check_file(path, "tlmgr"):
-			return path
-		# if there is a bin folder, go into it
 		if os.path.isdir(os.path.join(path, "bin")):
 			return _resolve_path(os.path.join(path, "bin"))
 		# if there is only 1 folder in the path, go into it
 		if len(os.listdir(path)) == 1:
 			return _resolve_path(os.path.join(path, os.listdir(path)[0]))
+		if _check_file(path, "tlmgr"):
+			if str(path) not in sys.path:
+				sys.path.append(str(path))
+			return path
 	except FileNotFoundError:
 		pass
 	raise RuntimeError(f"Unable to resolve TinyTeX path.\nTried {path}.\nYou can install TinyTeX using pytinytex.download_tinytex()")
 
 def _check_file(dir, prefix):
+	# check if a file in dir exists.
+	# the file has to have tthe name, but can have any extension
+	# this is for checking if tlmgr is in the bin folder, and make it work for both Windows and Unix
 	try:
 		for s in os.listdir(dir):
 			if os.path.splitext(s)[0] == prefix and os.path.isfile(os.path.join(dir, s)):
 				return True
 	except FileNotFoundError:
 		return False
 
-def _get_file(dir, prefix):
-	try:
-		for s in os.listdir(dir):
-			if os.path.splitext(s)[0] == prefix and os.path.isfile(os.path.join(dir, s)):
-				return os.path.join(dir, s)
-	except FileNotFoundError:
-		raise RuntimeError("Unable to find {}.".format(prefix))
-
-def _run_tlmgr_command(args, path, machine_readable=True):
+def _run_tlmgr_command(args, path, machine_readable=True, interactive=False):
 	if machine_readable:
 		if "--machine-readable" not in args:
 			args.insert(0, "--machine-readable")
-	tlmgr_executable = _get_file(path, "tlmgr")
+	tlmgr_executable = get_tlmgr_executable()
 	args.insert(0, tlmgr_executable)
 	new_env = os.environ.copy()
-	creation_flag = 0x08000000 if sys.platform == "win32" else 0 # set creation flag to not open TinyTeX in new console on windows
-	p = subprocess.Popen(
-		args,
-		stdout=subprocess.PIPE,
-		stderr=subprocess.PIPE,
-		env=new_env,
-		creationflags=creation_flag)
-	# something else than 'None' indicates that the process already terminated
-	if p.returncode is not None:
-		raise RuntimeError(
-			'TLMGR died with exitcode "%s" before receiving input: %s' % (p.returncode,
-																		   p.stderr.read())
-		)
-
-	stdout, stderr = p.communicate()
-
+	creation_flag = 0x08000000 if sys.platform == "win32" else 0
+
 	try:
-		stdout = stdout.decode("utf-8")
-	except UnicodeDecodeError:
-		raise RuntimeError("Unable to decode stdout from TinyTeX")
-
+		return asyncio.run(_run_command(*args, stdin=interactive, env=new_env, creationflags=creation_flag))
+	except Exception:
+		raise
+
+async def read_stdout(process, output_buffer):
+	"""Read lines from process.stdout and print them."""
+	try:
+		while True:
+			line = await process.stdout.readline()
+			if not line:  # EOF reached
+				break
+			line = line.decode('utf-8').rstrip()
+			output_buffer.append(line)
+	except Exception as e:
+		print("Error in read_stdout:", e)
+	finally:
+		process._transport.close()
+	return await process.wait()
+
+
+async def send_stdin(process):
+	"""Read user input from sys.stdin and send it to process.stdin."""
+	loop = asyncio.get_running_loop()
 	try:
-		stderr = stderr.decode("utf-8")
-	except UnicodeDecodeError:
-		raise RuntimeError("Unable to decode stderr from TinyTeX")
+		while True:
+			# Offload the blocking sys.stdin.readline() call to the executor.
+			user_input = await loop.run_in_executor(None, sys.stdin.readline)
+			if not user_input:  # EOF (e.g. Ctrl-D)
+				break
+			process.stdin.write(user_input.encode('utf-8'))
+			await process.stdin.drain()
+	except Exception as e:
+		print("Error in send_stdin:", e)
+	finally:
+		if process.stdin:
+			process._transport.close()
+
+
+async def _run_command(*args, stdin=False, **kwargs):
+	# Create the subprocess with pipes for stdout and stdin.
+	process = await asyncio.create_subprocess_exec(
+		*args,
+		stdout=asyncio.subprocess.PIPE,
+		stderr=asyncio.subprocess.STDOUT,
+		stdin=asyncio.subprocess.PIPE if stdin else asyncio.subprocess.DEVNULL,
+		**kwargs
+	)
+
+	output_buffer = []
+	# Create tasks to read stdout and send stdin concurrently.
+	stdout_task = asyncio.create_task(read_stdout(process, output_buffer))
+	stdin_task  = None
+	if stdin:
+		stdin_task  = asyncio.create_task(send_stdin(process))
 
-	if stderr == "" and p.returncode == 0:
-		return p.returncode, stdout, stderr
-	else:
-		raise RuntimeError("TLMGR died with the following error:\n{0}".format(stderr.strip()))
-		return p.returncode, stdout, stderr
+	try:
+		if stdin:
+			# Wait for both tasks to complete.
+			await asyncio.gather(stdout_task, stdin_task)
+		else:
+			# Wait for the stdout task to complete.
+			await stdout_task
+		# Return the process return code.
+		exit_code = await process.wait()
+	except KeyboardInterrupt:
+		print("\nKeyboardInterrupt detected, terminating subprocess...")
+		process.terminate()  # Gracefully terminate the subprocess.
+		exit_code = await process.wait()
+	finally:
+		# Cancel tasks that are still running.
+		stdout_task.cancel()
+		if stdin_task:
+			stdin_task.cancel()
+	captured_output = "\n".join(output_buffer)
+	if exit_code != 0:
+		raise RuntimeError(f"Error running command: {captured_output}")
+	return exit_code, captured_output
+
+
+	return process.returncode

pytinytex/tinytex_download.py

@@ -1,3 +1,4 @@
+import os
 import sys
 
 import re
@@ -78,12 +79,8 @@ def download_tinytex(version="latest", variation=1, target_folder=DEFAULT_TARGET
 		# copy the extracted folder to the target folder, overwriting if necessary
 		print("Copying TinyTeX to %s..." % target_folder)
 		shutil.copytree(tinytex_extracted, target_folder, dirs_exist_ok=True)
-	# go into target_folder/bin, and as long as we keep having 1 and only 1 subfolder, go into that, and add it to path
-	folder_to_add_to_path = target_folder / "bin"
-	while len(list(folder_to_add_to_path.glob("*"))) == 1 and folder_to_add_to_path.is_dir():
-		folder_to_add_to_path = list(folder_to_add_to_path.glob("*"))[0]
-	print(f"Adding TinyTeX to path ({str(folder_to_add_to_path)})...")
-	sys.path.append(str(folder_to_add_to_path))
+	sys.path.append(str(target_folder))
+	os.environ["PYTINYTEX_TINYTEX"] = str(target_folder)
 	print("Done")
 
 def _get_tinytex_urls(version, variation):

tests/test_tinytex_commands.py

@@ -0,0 +1,8 @@
+import pytinytex
+
+from .utils import download_tinytex # noqa: F401
+
+def test_help(download_tinytex): # noqa: F811
+	exit_code, output = pytinytex.help()
+	assert exit_code == 0
+	assert "the native TeX Live Manager".lower() in output.lower()

tests/test_tinytex_path_resolver.py

@@ -15,10 +15,10 @@ def test_successful_resolver(download_tinytex): # noqa
 	assert isinstance(pytinytex.__tinytex_path, str)
 	assert os.path.isdir(pytinytex.__tinytex_path)
 
-def test_get_tinytex_path(download_tinytex): # noqa
+def test_get_tinytex_distribution_path(download_tinytex): # noqa
 	# actually resolve the path
 	pytinytex.ensure_tinytex_installed(TINYTEX_DISTRIBUTION)
-	assert pytinytex.__tinytex_path == pytinytex.get_tinytex_path(TINYTEX_DISTRIBUTION)
+	assert pytinytex.__tinytex_path == pytinytex.get_tinytex_distribution_path(TINYTEX_DISTRIBUTION)
 
 @pytest.mark.parametrize("download_tinytex", [1], indirect=True)
 def test_get_pdf_latex_engine(download_tinytex): # noqa