Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI, Request | |
| from pytanque import Pytanque, PetanqueError | |
| import subprocess, os, tempfile | |
| import time | |
| app = FastAPI() | |
| PET_PORT = 8080 | |
| MATHCOMP_PATH = "/home/mathcomp/" | |
| pet = None | |
| pet_process = None | |
| dict_mathcomp = {} | |
| def startup_event(): | |
| global pet, pet_process, dict_mathcomp | |
| pet_process = subprocess.Popen(["pet-server", "--port", str(PET_PORT)]) | |
| time.sleep(1) | |
| pet = Pytanque("127.0.0.1", PET_PORT) | |
| pet.connect() | |
| print("PET server started") | |
| for root, _, files in os.walk(MATHCOMP_PATH): | |
| for file in files: | |
| if file.endswith(".v"): | |
| abs_path = os.path.join(root, file) | |
| rel_path = os.path.relpath(abs_path, MATHCOMP_PATH) | |
| rel_path = rel_path[:-2].replace('/', '.') | |
| with open(abs_path, 'r') as file: | |
| content = file.read() | |
| dict_mathcomp[rel_path] = {"path": abs_path, "content": content} | |
| print(f"Indexed {len(dict_mathcomp)} .v files in MathComp") | |
| def shutdown_event(): | |
| if pet_process: | |
| pet_process.terminate() | |
| print("PET server stopped") | |
| async def check(): | |
| return {"status": "ok", "message": "Rocq backend running"} | |
| async def sources(): | |
| return list(dict_mathcomp.keys()) | |
| async def all_sources(req: Request): | |
| return dict_mathcomp | |
| async def toc(req: Request): | |
| data = await req.json() | |
| source = data['source'] | |
| try: | |
| toc = pet.toc(dict_mathcomp[source]) | |
| result = {} | |
| for name, details in toc: | |
| if len(details) == 1: | |
| info = details[0] | |
| if info['detail'] in ['Lemma', 'Theorem']: | |
| result[name] = {"range": info['range'], "kind": info['detail']} | |
| return result | |
| except Exception as e: | |
| return {"error": str(e)} | |