I have a python application that must run only one instance of it except in two particular cases. For that reason, I'm having issues achieving the second instance of the same application to communicate (in the way I want) with the first instance of that application. Let me explain in a more "practical" way:
- I execute the program (Which is an executable that I made with Pyinstaller)
- Everything normal at this time; the app just do what it needs to do.
- I re-execute the executable file again. Here is where I'm having issues.
The application currently kills the second instance and works fine. But now I want the second instance to just simply print in the console to ask me what I want to do: Either close the program (This is kill both running instances) or just cancel the operation (Kill just the second instance which is not a problem because is what I have implemented right now).
I tried to run a server and a client. When the app is executed for the first time, it creates a simple pid (process id) file that identifies that instance of the application and then it runs the server function. When I execute for second time the program, if the pid file is detected it runs the client function. Then, the client just sends a message to the server, so it can do something, like killing itself (and the second instance will too).
I'm pretty sure this is the way to go but I'm having issues implementing this. I cannot just run in a loop the server infinitely because the program needs to continue in many other things. I tried to run the server in a different thread but I have not much experience with concurrency so, maybe that's my problem.
Here is my current main.py file. I cannot show more for some reasons, but I think this with my previous explanation is more than enough. Pay more attention to the ** init, __notify(), and __server() methods where I was thinking to implement everything.
import sys, time, json, logging, os
from api_handlers import team_viewer_api as tv_api
from utils.report_writter import ReportWritter
from config.schemas import BasicAppConfig
from utils.unique_pid import instance_listener
from config.gui_model.config_from_gui import get_config_from_gui
from config.arg_model.config_from_args import get_config_from_args
class ApplicationDeviceReporter:
def __init__(self):
logging.basicConfig(filename="teamViewerAppLog.log",
filemode="a",
level=logging.DEBUG,
format="%(levelname)s[%(asctime)s]: %(message)s",
datefmt="%d/%m/%Y %I:%M:%S %p")
self.PID_FILE = os.getcwd() + r'\pid'
single_instance = instance_listener()
is_one_instance = single_instance.detect_running_instances(self.PID_FILE)
if not is_one_instance:
self.__notify()
sys.exit()
else:
# Should I call self.__listen() in a different thread, or run the rest of the app
# in a different thread and just loop infinitely the __listen() method?
pass
self.__setup_app_configuration()
logging.info("Starting application...")
def __notify(self):
# Implement a mechanism to notify the first instance to go into standby mode
# For example, you can write a specific file, send a signal, or use IPC
pass
def __listen(self):
# Implement a mechanism to listen for notification from another instance.
# To kill this runtime or do another thing.
pass
def __setup_app_configuration(self, config_params: dict = None):
logging.info("Setting up app's configuration...")
if not config_params:
if len(sys.argv) == 1:
config_params = get_config_from_gui()
else:
config_params = get_config_from_args()
self.app_configuration = BasicAppConfig(device_id=config_params['device_id'],
token=config_params['token'],
run_once=config_params['run_once'],
wait_time=config_params['wait_time'])
logging.info("Configuration has been set succesfully.")
def run(self):
self.report_writter = ReportWritter()
self.teamviewer_api_interface = tv_api.TeamViewerApiInterface(
TOKEN=self.app_configuration.token,
DEVICE_ID=self.app_configuration.device_id,
)
while True:
app_timer_start = time.perf_counter()
logging.info(f"Getting device status...")
# try:
# data_received = self.teamviewer_api_interface.get_teamviewer_data()
# except Exception as e:
# time.sleep(60)
# continue
# logging.info("Writing report files...")
# self.report_writter.write_reports(api_results=self.teamviewer_api_interface.results)
app_timer_finish = time.perf_counter()
logging.info(f'Program execution performance: {round(app_timer_finish - app_timer_start, 3)} second(s)')
if self.app_configuration.run_once:
break
logging.info(f"Execution finished. Waiting until next execution in {round(self.app_configuration.wait_time / 3600, 1)} hours and {round((self.app_configuration.wait_time % 3600) / 60)} minutes...")
time.sleep(self.app_configuration.wait_time)
logging.info("Program finished.")
os.remove(self.PID_FILE)
sys.exit(0)
if __name__ == "__main__":
app = ApplicationDeviceReporter()
app.run()