"""Local HTTP API and interactive demo for the three Open classifiers.""" import argparse import json import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import torch from predict_windows import load_task, make_windows, classify_windows from api_access import AccessStore, AccessError, DEFAULT_DB ROUTES = {'/v1/emotions': 'open-emotions', '/v1/morals': 'open-morals', '/v1/sensations': 'open-sensations'} MODELS = {} MODEL_PATHS = {} LOCK = threading.Lock() ACCESS = None ALLOW_ANONYMOUS = False PAGE = Path(__file__).with_name('api_demo.html') def predict(task, request): text = request.get('text') if not isinstance(text, str) or not text.strip() or len(text) > 20000: raise ValueError('text must contain 1..20000 characters.') target = request.get('target') if task == 'open-sensations': if not isinstance(target, str) or not target.strip() or len(target) > 80: raise ValueError('Sensations requires target: a word or phrase of 1..80 characters.') elif target is not None: raise ValueError('target applies only to sensations.') mode = request.get('mode', 'sliding') if mode not in ['chunks', 'sliding']: raise ValueError('mode must be chunks or sliding.') limit = 256 if task == 'open-morals' else 128 size = request.get('window_size', limit) if type(size) is not int or not 16 <= size <= limit: raise ValueError(f'window_size must be an integer in 16..{limit}.') overlap = request.get('overlap', 0 if mode == 'chunks' else size // 4) if type(overlap) is not int or overlap < 0 or overlap > size // 2: raise ValueError('overlap must be a nonnegative integer, at most half the total window size.') if mode == 'chunks' and overlap: raise ValueError('chunks requires overlap 0.') if task not in MODELS: MODELS[task] = load_task(task, MODEL_PATHS.get(task)) model, tokenizer, options = MODELS[task] if target: prefix_tokens = len(tokenizer(f'Target: {target}\nContext: ', add_special_tokens=False)['input_ids']) special_tokens = tokenizer.num_special_tokens_to_add(pair=False) if overlap > max(0, (size - prefix_tokens - special_tokens) // 2): raise ValueError('Reduce overlap to at most half the usable sensory context capacity.') windows, metadata = make_windows(tokenizer, text, size, overlap, target) if len(windows) > 128: raise ValueError('Request exceeds 128 evaluated windows. Shorten text or increase window size.') evidence, summaries = classify_windows(model, tokenizer, windows, options['global_threshold']) return dict(task=task, model=getattr(model.config,'repo_id',f'mts-the-architect/{task}-roberta-125m-fp32'), mode=mode, target=target, window_size=size, threshold=options['global_threshold'], **metadata, windows=evidence, exploratory_summary=summaries, interpretation='Per-window labels use short-input thresholds. Max/mean summaries are uncalibrated. Sensory scores are target associations, not verified sensation presence.') class Handler(BaseHTTPRequestHandler): def setup(self): super().setup() self.connection.settimeout(15) def reply(self, status, data, content_type='application/json', retry_after=0): raw = data if isinstance(data, bytes) else json.dumps(data, ensure_ascii=False).encode('utf-8') self.send_response(status) self.send_header('Content-Type', content_type + '; charset=utf-8') self.send_header('Content-Length', str(len(raw))) self.send_header('Cache-Control', 'no-store') if retry_after: self.send_header('Retry-After', str(retry_after)) self.end_headers() self.wfile.write(raw) def do_GET(self): if self.path in ['/', '/docs']: self.reply(200, PAGE.read_bytes(), 'text/html') elif self.path == '/health': self.reply(200, dict(status='ok', loaded_models=list(MODELS), authentication_required=not ALLOW_ANONYMOUS, default_requests_per_minute=10, default_requests_per_day=500)) elif self.path == '/v1/models': self.reply(200, dict(endpoints=ROUTES, limits=dict(emotions=128, morals=256, sensations=128))) else: self.reply(404, dict(error='Unknown endpoint. See /docs.')) def do_POST(self): if self.path not in ROUTES and self.path != '/v1/access-requests': self.reply(404, dict(error='Unknown endpoint.')) return # No browser cross-origin inference: the demo uses this same server origin. origin = self.headers.get('Origin') if origin and origin not in [f'http://127.0.0.1:{self.server.server_port}', f'http://localhost:{self.server.server_port}']: self.reply(403, dict(error='Cross-origin browser requests are disabled.')) return try: # Peer address only; forwarded headers are not trusted by this local server. ACCESS.charge('peer:'+self.client_address[0], 60, 10000) length = int(self.headers.get('Content-Length', '0')) if not 0 < length <= 100000: raise ValueError('JSON body must be 1..100000 bytes.') body = json.loads(self.rfile.read(length).decode('utf-8')) if not isinstance(body, dict): raise ValueError('JSON object required.') if self.path == '/v1/access-requests': ACCESS.charge('registration:'+self.client_address[0], 3, 10, minute_period=3600) self.reply(201, ACCESS.request(body.get('app_name'), body.get('contact'))) return auth = self.headers.get('Authorization', '') ACCESS.authorize(auth[7:] if auth.startswith('Bearer ') else '', ALLOW_ANONYMOUS) if not LOCK.acquire(blocking=False): self.reply(503, dict(error='Inference busy; retry later.'), retry_after=1) return try: result = predict(ROUTES[self.path], body) finally: LOCK.release() self.reply(200, result) except AccessError as error: self.reply(error.status, dict(error=str(error)), retry_after=error.retry_after) except (ValueError, UnicodeDecodeError) as error: self.reply(400, dict(error=str(error))) except Exception: self.reply(500, dict(error='Inference failed; check local server/model setup.')) def main(): global ACCESS, ALLOW_ANONYMOUS parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--port', type=int, default=8766) parser.add_argument('--access-db', type=Path, default=DEFAULT_DB, help='Private approval/quota database; use separate paths for separate server instances') parser.add_argument('--allow-anonymous', action='store_true', help='Local demo only; anonymous calls share 10/minute and 500/day') parser.add_argument('--model-dir', action='append', default=[], metavar='TASK=PATH', help='Override a downloaded checkpoint directory; repeat per task') args = parser.parse_args() for item in args.model_dir: task, separator, path = item.partition('=') if not separator or task not in ROUTES.values() or not Path(path).is_dir(): parser.error('--model-dir requires open-emotions/open-morals/open-sensations=existing-directory') MODEL_PATHS[task] = Path(path) ACCESS = AccessStore(args.access_db) ALLOW_ANONYMOUS = args.allow_anonymous torch.set_num_threads(8) server = ThreadingHTTPServer(('127.0.0.1', args.port), Handler) print(f'Open Models API: http://127.0.0.1:{args.port}/docs (local only)', flush=True) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == '__main__': main()