FM / supervisor

/usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/supervisor/

/usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/supervisor

Name	Size	Mode	Actions
medusa/	-	0755	rm
scripts/	-	0755	rm
skel/	-	0755	rm
tests/	-	0755	rm
ui/	-	0755	rm
__pycache__/	-	0755	rm
childutils.py	2561	0644	edit dl rm
compat.py	5692	0644	edit dl rm
confecho.py	243	0644	edit dl rm
datatypes.py	13267	0644	edit dl rm
dispatchers.py	19293	0644	edit dl rm
events.py	7235	0644	edit dl rm
http.py	31882	0644	edit dl rm
http_client.py	7030	0644	edit dl rm
loggers.py	13311	0644	edit dl rm
options.py	88291	0644	edit dl rm
pidproxy.py	2006	0644	edit dl rm
poller.py	6703	0644	edit dl rm
process.py	39444	0644	edit dl rm
rpcinterface.py	37823	0644	edit dl rm
socket_manager.py	3095	0644	edit dl rm
states.py	1791	0644	edit dl rm
supervisorctl.py	54885	0644	edit dl rm
supervisord.py	15208	0644	edit dl rm
templating.py	47081	0644	edit dl rm
version.txt	6	0644	edit dl rm
web.py	24318	0644	edit dl rm
xmlrpc.py	21438	0644	edit dl rm
__init__.py	20	0644	edit dl rm

Edit: /usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/supervisor/childutils.py (2561B)

import sys
import time

from supervisor.compat import xmlrpclib
from supervisor.compat import long
from supervisor.compat import as_string

from supervisor.xmlrpc import SupervisorTransport
from supervisor.events import ProcessCommunicationEvent
from supervisor.dispatchers import PEventListenerDispatcher

def getRPCTransport(env):
    u = env.get('SUPERVISOR_USERNAME', '')
    p = env.get('SUPERVISOR_PASSWORD', '')
    return SupervisorTransport(u, p, env['SUPERVISOR_SERVER_URL'])

def getRPCInterface(env):
    # dumbass ServerProxy won't allow us to pass in a non-HTTP url,
    # so we fake the url we pass into it and always use the transport's
    # 'serverurl' to figure out what to attach to
    return xmlrpclib.ServerProxy('http://127.0.0.1', getRPCTransport(env))

def get_headers(line):
    return dict([ x.split(':') for x in line.split() ])

def eventdata(payload):
    headerinfo, data = payload.split('\n', 1)
    headers = get_headers(headerinfo)
    return headers, data

def get_asctime(now=None):
    if now is None: # for testing
        now = time.time() # pragma: no cover
    msecs = (now - long(now)) * 1000
    part1 = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now))
    asctime = '%s,%03d' % (part1, msecs)
    return asctime

class ProcessCommunicationsProtocol:
    def send(self, msg, fp=sys.stdout):
        fp.write(ProcessCommunicationEvent.BEGIN_TOKEN)
        fp.write(msg)
        fp.write(ProcessCommunicationEvent.END_TOKEN)
        fp.flush()

def stdout(self, msg):
        return self.send(msg, sys.stdout)

def stderr(self, msg):
        return self.send(msg, sys.stderr)

pcomm = ProcessCommunicationsProtocol()

class EventListenerProtocol:
    def wait(self, stdin=sys.stdin, stdout=sys.stdout):
        self.ready(stdout)
        line = stdin.readline()
        headers = get_headers(line)
        payload = stdin.read(int(headers['len']))
        return headers, payload

def ready(self, stdout=sys.stdout):
        stdout.write(as_string(PEventListenerDispatcher.READY_FOR_EVENTS_TOKEN))
        stdout.flush()

def ok(self, stdout=sys.stdout):
        self.send('OK', stdout)

def fail(self, stdout=sys.stdout):
        self.send('FAIL', stdout)

def send(self, data, stdout=sys.stdout):
        resultlen = len(data)
        result = '%s%s\n%s' % (as_string(PEventListenerDispatcher.RESULT_TOKEN_START),
                               str(resultlen),
                               data)
        stdout.write(result)
        stdout.flush()

listener = EventListenerProtocol()

import sys import time from supervisor.compat import xmlrpclib from supervisor.compat import long from supervisor.compat import as_string from supervisor.xmlrpc import SupervisorTransport from supervisor.events import ProcessCommunicationEvent from supervisor.dispatchers import PEventListenerDispatcher def getRPCTransport(env): u = env.get('SUPERVISOR_USERNAME', '') p = env.get('SUPERVISOR_PASSWORD', '') return SupervisorTransport(u, p, env['SUPERVISOR_SERVER_URL']) def getRPCInterface(env): # dumbass ServerProxy won't allow us to pass in a non-HTTP url, # so we fake the url we pass into it and always use the transport's # 'serverurl' to figure out what to attach to return xmlrpclib.ServerProxy('http://127.0.0.1', getRPCTransport(env)) def get_headers(line): return dict([ x.split(':') for x in line.split() ]) def eventdata(payload): headerinfo, data = payload.split('\n', 1) headers = get_headers(headerinfo) return headers, data def get_asctime(now=None): if now is None: # for testing now = time.time() # pragma: no cover msecs = (now - long(now)) * 1000 part1 = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now)) asctime = '%s,%03d' % (part1, msecs) return asctime class ProcessCommunicationsProtocol: def send(self, msg, fp=sys.stdout): fp.write(ProcessCommunicationEvent.BEGIN_TOKEN) fp.write(msg) fp.write(ProcessCommunicationEvent.END_TOKEN) fp.flush() def stdout(self, msg): return self.send(msg, sys.stdout) def stderr(self, msg): return self.send(msg, sys.stderr) pcomm = ProcessCommunicationsProtocol() class EventListenerProtocol: def wait(self, stdin=sys.stdin, stdout=sys.stdout): self.ready(stdout) line = stdin.readline() headers = get_headers(line) payload = stdin.read(int(headers['len'])) return headers, payload def ready(self, stdout=sys.stdout): stdout.write(as_string(PEventListenerDispatcher.READY_FOR_EVENTS_TOKEN)) stdout.flush() def ok(self, stdout=sys.stdout): self.send('OK', stdout) def fail(self, stdout=sys.stdout): self.send('FAIL', stdout) def send(self, data, stdout=sys.stdout): resultlen = len(data) result = '%s%s\n%s' % (as_string(PEventListenerDispatcher.RESULT_TOKEN_START), str(resultlen), data) stdout.write(result) stdout.flush() listener = EventListenerProtocol()