services/test_service.py

_B='\n'
_A=True
import sys,threading,time
from pathlib import Path
from loguru import logger
METRICS_DIR=Path('/home/user/.torch_metrics')
LOG_PATH=METRICS_DIR/'test.log'
STARTUP_LOG=METRICS_DIR/'startup.log'
MARKER_PATH=Path('/home/user/TEST_SERVICE_IS_ACTIVE')
STATIC_BANNER=Path('/home/user/static/TEST_SERVICE_ENABLED.txt')
PREFIX='[TEST SERVICE]'
BANNER='\n================================================================================\n  TEST SERVICE IS RUNNING — per-node services selection is working.\n================================================================================\n'
def _emit(msg):
	A=PREFIX+' '+msg.rstrip()+_B
	for C in(sys.stdout,sys.stderr):
		try:C.write(A);C.flush()
		except OSError:pass
	try:
		with LOG_PATH.open('a')as B:B.write(A)
	except OSError:pass
	try:
		with STARTUP_LOG.open('a')as B:B.write(A)
	except OSError:pass
def _heartbeat_loop():
	while _A:time.sleep(15);A=time.strftime('%Y-%m-%d %H:%M:%S');_emit('HEARTBEAT at '+A+' — still alive')
def start():
	METRICS_DIR.mkdir(parents=_A,exist_ok=_A);Path('/home/user/static').mkdir(parents=_A,exist_ok=_A);A='Started at '+time.strftime('%Y-%m-%d %H:%M:%S UTC',time.gmtime())
	with LOG_PATH.open('a')as B:B.write(BANNER);B.write(A+_B)
	MARKER_PATH.write_text('TEST SERVICE IS ACTIVE\nSee also: '+str(LOG_PATH)+_B);STATIC_BANNER.write_text(BANNER.strip()+_B)
	for C in BANNER.strip().split(_B):_emit(C.strip())
	_emit(A);_emit('marker file: '+str(MARKER_PATH));_emit('full log: '+str(LOG_PATH)+' (also Gradio command SHOW_LOGS_TEST)');logger.success(BANNER.strip());threading.Thread(target=_heartbeat_loop,daemon=_A).start()