Realizing our perception of time can change drastically when we are drowning in work is a rather funny experience. Reading the publishing date of the last article nearly gave me a good shock when I realized it was done half a year ago. Yet, it felt recent, and the follow-up to the project is still sitting in the filesystem awaiting planning and drafting.
Actor Models in Pygame: Lessons in Serialization and State Isolation
Cute illustration generated by Gemini
Unfortunately, that story has to wait for a bit, as we are side-tracking to a different territory today. Definitely not a reflection of my current job situation where I am struggling outside of my comfort zone. Alrighty, let’s move on from this bad self-deprecating joke. Instead, we are abruptly revisiting Python subprocess management with AsyncIO as if nothing happened.
The Limits of Shell Scripting
Photo by Charles Asselin on Unsplash
There are times where we utilize shell scripting to automate some tasks, but to what extent should we stick with just a scripting language? With LLM getting so much more capable these days, does the choice of tool still matter?
Our last attempt was about passing data from one utility to another, from a video encoder to a streamer (both ffmpeg executed with different arguments). There’s no data to be passed between processes this time, and we are just dealing with one process at a time. Sounds like a trivial shell script to write, an LLM chatbot would be more than enough for the job, that’s what I thought.
Video data IO through ffmpeg subprocess
Fortunately, Gemini flash did deliver.
Some additional background to the problem is likely warranted before we dive deeper. It is indeed true that we are not passing data between processes, but the process we are interested in managing does require some user interaction, both directly and indirectly through the command line interface. Another annoyance is that it would just silently terminate on failure and that affects my well-being.
Because when it terminates, I allow my partner to punch me in the face.
It was clearly a joke.
But you get the point.
Getting the utility in question to re-run on failure is then Gemini’s first attempt to turn the work into a shell script written in bash.
Modern Prototyping with AI Autocomplete
Photo by Barbara Zandoval on Unsplash
Writing the script in Bash is a fine choice, the utility does work as expected. Gemini gave me a script that looks maintainable, and like many sane, lazy coders (please disregard the adjective combination), I started piling more requirements to the original problem.
This experience of treating an LLM chatbot as my Santa does make me wonder — with the overly primitive constructs and toolchain, am I expecting bash to do too much?
Welcome to the show, where I get Gemini to turn my laziness into cryptic bash code.
First request was simple, we all like easy things. Though not a needy person, I do want to make myself available to my partner after certain hours. Dear chatbot, do not retry after dinner time.
The code was generated; Gemini obliged.
But there were times I needed it for a one-off fling, and the script wouldn’t co-operate. My partner continued to enjoy his victory over my oversight — until another revision to let it simulate a do-while loop, to run at least once and do the I-want-my-me-time guard check after the first iteration.
Besides interacting with it through the command line interface, the execution of the utility also halts temporarily when it is awaiting input through my cellphone. Think of it as a wellness check procedure; imagine I am preparing to shop for an expensive toy through a website, and it automatically asks for my partner’s approval.
I feel like someone just said no from a distance, as I am typing this.
Not just in my head, I hope.
Before I knew it, the bash script was already spawning subshells, keeping track of the process ID and trapping SIGTERM in case the user wants to jump into his partner’s arms. Getting it to quit when I fail to respond to my cellphone in time turns the script into gibberish that I can no longer comprehend.
Oh, the generated code was also broken, and I sincerely do not blame Gemini.
I would whack the user if I were the chatbot.
Pay for Pro subscription!
Building a Non-Blocking Process Manager in Python
Photo by Raja Patel on Unsplash
My incompetency with Bash eventually led to a script that I cannot personally maintain. Paying for Pro subscription over this is dumb, and that made me rethink my choices in life. What if I had done this in Python from the very beginning, where it has all the bells and whistles for the job.
Looking at the monstrosity, I got fed up with getting Gemini to fix it further. Just to be very frank, I couldn’t even pin-point where things were done incorrectly. That was when I reminded myself of the video encoding project I did earlier.
Essentially, the problem can be written in a coroutine function similar to
async def connect(exit_event: asyncio.Event, password: str, timeout : int = KILL_TIMEOUT):
logger = logging.getLogger('Connect')
while not exit_event.is_set():
logger.info("Running utility")
await interact(
await asyncio.create_subprocess_exec(
'some-utility',
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
stdin=asyncio.subprocess.PIPE
),
password,
timeout,
exit_event
)
logger.info("Process exited, waiting before retrying")
await asyncio.sleep(RETRY_TIMEOUT)
if is_past_off_hour():
logger.info("Not repeating connection, exiting loop")
break
logger.info("Exiting connection loop")
exit_event.set()
Being increasingly more capable, LLM chatbots took over 70–80% of my programming work. As a result, I forgot that I had autocompletion configured in my editor. After drafting the core coroutine above, I started typing the function signature for all the interaction, and suddenly I was greeted with a somewhat usable draft.
async def interact(process: asyncio.subprocess.Process, password: str, timeout : int, exit_event: asyncio.Event):
logger = logging.getLogger('Command output')
is_success = asyncio.Event()
tasks = (
asyncio.create_task(interact_timeout(process, timeout, is_success, exit_event)),
asyncio.create_task(interact_kill(process, exit_event))
)
process.stdin.write(f"{password}\n\n".encode('utf-8'))
await process.stdin.drain()
while process.returncode is None:
line_bytes = await process.stdout.readline()
if not line_bytes:
break
line = line_bytes.decode('utf-8').strip()
logger.info("%s", line)
if is_success.is_set() and re.search(r"PARTNER APPROVED", line):
is_success.set()
for task in tasks:
if task.done() is False:
task.cancel()
await process.wait()
Unlike the main coroutine that manages the overall retry-until-me-time setup, the interact coroutine manages each session with the process created by executing the utility. As mentioned, it expects some interaction via the command line interface through STDIN, hence the process.stdin.write call to input a sequence of characters expected by the utility.
Capturing the success state is also crucial, and we do that by checking the execution output for a certain string (The string PARTNER APPROVED in the mock example). After the corresponding event is set, we cancel the interaction timeout, as shown below
async def interact_timeout(process: asyncio.subprocess.Process, timeout: int, has_dns: asyncio.Event, has_route: asyncio.Event, exit_event: asyncio.Event):
logger = logging.getLogger('Timeout killer')
await asyncio.sleep(timeout)
if not is_success.is_set():
logger.error("Timeout waiting for connection, killing process and sending event to exit")
process.terminate()
exit_event.set()
return
Lastly, we need to be able to terminate the process whenever desired — such as when we send a SIGTERM to the script.
async def interact_kill(process: asyncio.subprocess.Process, exit_event: asyncio.Event):
logger = logging.getLogger('Exit killer')
await exit_event.wait()
logger.info("Exit event set, killing process")
process.terminate()
Trapping the SIGTERM event can be done through the snippet shown below, adapted from the snippet we discussed previously
AsyncIO Task Management: A Hands-On Scheduler Project
@dataclass
class ShutdownHandler:
exit_event: asyncio.Event
def __call__ (
self, signum: int | None = None, frame: FrameType | None = None
) -> None:
logger = logging.getLogger('ShutdownHandler')
logger.info("Sending exit event to all tasks in pool")
self.exit_event.set()
for task in asyncio.all_tasks():
if task is not asyncio.current_task():
task.cancel()
exit_event = asyncio.Event()
shutdown_handler = ShutdownHandler(exit_event)
for sig in (signal.SIGINT, signal.SIGTERM):
asyncio.get_event_loop().add_signal_handler(sig, shutdown_handler)
Seeing the LLM autocompletion feature in action is both a magical and creepy experience. Most of the time it made the work a breeze, especially when it correctly predicted what I had in mind. On the other hand it really felt like someone is watching over my shoulder constantly.
Developing the script took about an hour, and that included testing to ensure things work as expected. AsyncIO proved to be the right call, as the non-blocking nature ensured the process would terminate if I failed to respond on my cellphone in time. The orchestration would be significantly more complex should we do this synchronously.
Not like asynchronous code is any easier to read and comprehend anyway.
Pick your poison.
The Verdict: Architecture, Nix, and Portability
Photo by Niek Doup on Unsplash
I didn’t put in too much effort once I got it running. All I did was taking ideas from my past projects and assemble them together for the use case. No test suite was written, and yet, it worked fine. I guess it is time to answer the core question: was rewriting this in Python the right call?
It was definitely written to fit my own unique use-case, so there’s no concern about distributing or sharing it with others. Considering it is one of those execute-and-forget (with mostly minimal intervention) programs, having it written in a language I am more comfortable with is crucial. I am sure given enough time and effort, I could get Gemini to generate a usable bash equivalent, but I would rather spend that time on things that matter.
Like hugging my partner.
One peculiar discovery during the rewrite was realizing macOS Tahoe is still shipping with Python 3.9. Despite trying to not rely on modern tooling for this humble project (e.g. skipping uv for project management and ruff for linting), the new language constructs brought by 3.10 were too good to miss. Fortunately I already manage my environment with Nix’s home-manager, so setting up 3.10 by default was a no-brainer.
Beyond rbenv and pyenv: Achieving Reproducible Dev Setups with Nix
The implementation of timeout to terminate if I fail to respond to my phone in time is also done deliberately. Such interaction is usually time-sensitive, so if I miss it at first and return after a while it would have been expired anyway. Terminating the process on an assumption that I am absent is therefore a sensible choice.
As a conclusion to this side-project, yes, rewriting the small humble script in Python is the right call, purely because it makes it maintainable for myself. If you are reading this, the takeaway of this article is hopefully to show how much my partner tolerates my erratic behaviour.
I meant how AsyncIO can be helpful in such scenario.
While I didn’t publish the script in full, the snippets shown aren’t that much different from what we discussed in the past. Do let me know if you need help on managing a long-running process through AsyncIO.
- Understanding Awaitables: Coroutines, Tasks, and Futures in Python
- AsyncIO Task Management: A Hands-On Scheduler Project
- Concurrency vs. Parallelism: Achieving Scalability with ProcessPoolExecutor
Thanks again for reading this far, and I hope to write again soon and get back to Pygame programming
Finally, I’d like to acknowledge the editorial support provided by Gemini. While the code, architecture, and real-world trial and error are entirely my own, Gemini served as an essential second set of eyes — auditing sentence-to-sentence flow, refining tense consistency, and helping me balance the narrative arc between technical mechanics and personal storytelling. It was a true collaborative effort to ensure this chaotic development journey translates clearly on the page.
Top comments (0)