28 lines
1.2 KiB
Python
28 lines
1.2 KiB
Python
import json, os, signal
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
class H(BaseHTTPRequestHandler):
|
|
def _j(self,o,s=200):
|
|
p=json.dumps(o).encode(); self.send_response(s)
|
|
self.send_header("Content-Type","application/json"); self.send_header("Content-Length",str(len(p))); self.end_headers(); self.wfile.write(p)
|
|
def do_GET(self):
|
|
if self.path=="/health": self._j({"status":"ok"})
|
|
elif self.path=="/": self._j({"name":"light-adapt-test","status":"running"})
|
|
else: self._j({"error":"not found"},404)
|
|
def do_POST(self):
|
|
if self.path=="/task":
|
|
l=int(self.headers.get("Content-Length",0)); b=self.rfile.read(l) if l else b"{}"
|
|
try: t=json.loads(b)
|
|
except: t={}
|
|
print("task",t.get("model_address",""),flush=True)
|
|
self._j({"status":"accepted","model":t.get("model_address","")})
|
|
else: self._j({"error":"not found"},404)
|
|
def log_message(self,*a): pass
|
|
run=True
|
|
def stop(*a):
|
|
global run; run=False
|
|
signal.signal(signal.SIGTERM,stop); signal.signal(signal.SIGINT,stop)
|
|
srv=ThreadingHTTPServer(("0.0.0.0",8080),H); srv.timeout=1
|
|
print("listening 8080",flush=True)
|
|
while run: srv.handle_request()
|
|
srv.server_close()
|