MitraSETI Cloud REST API

curl -H "x-api-key: sk_live_xxxxxxxxxxxx" \ https://api-dev.deepfieldlabs.dev/jobs

curl -H "Authorization: Bearer eyJraWQi..." \ https://api-dev.deepfieldlabs.dev/jobs

# Set your API key API_KEY="sk_live_xxxxxxxxxxxx" API="https://api-dev.deepfieldlabs.dev" AUTH="-H x-api-key:$API_KEY" # 1. Get presigned upload URL UPLOAD=$(curl -s -X POST "$API/upload-url" $AUTH \ -H "Content-Type: application/json" \ -d '{"filename":"observation.fil","file_size":46000000}') UPLOAD_URL=$(echo $UPLOAD | python3 -c "import sys,json; print(json.load(sys.stdin)['upload_url'])") FILE_KEY=$(echo $UPLOAD | python3 -c "import sys,json; print(json.load(sys.stdin)['file_key'])") # 2. Upload file to S3 curl -X PUT "$UPLOAD_URL" -H "Content-Type: application/octet-stream" \ --data-binary @observation.fil # 3. Submit for analysis JOB=$(curl -s -X POST "$API/analyze" $AUTH \ -H "Content-Type: application/json" \ -d "{\"file_key\":\"$FILE_KEY\",\"filename\":\"observation.fil\"}") JOB_ID=$(echo $JOB | python3 -c "import sys,json; print(json.load(sys.stdin)['job_id'])") echo "Job submitted: $JOB_ID" # 4. Poll for completion while true; do STATUS=$(curl -s "$API/jobs/$JOB_ID" $AUTH | python3 -c \ "import sys,json; print(json.load(sys.stdin)['status'])") echo "Status: $STATUS" [ "$STATUS" = "complete" ] || [ "$STATUS" = "failed" ] && break sleep 10 done # 5. Download results curl -s "$API/jobs/$JOB_ID/download?format=json" $AUTH | python3 -c \ "import sys,json; print(json.load(sys.stdin)['download_url'])"

{ "error": "Monthly limit reached", "limit": 10, "used": 10, "tier": "free" }

import requests import time import os API = "https://api-dev.deepfieldlabs.dev" API_KEY = "sk_live_xxxxxxxxxxxx" HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"} AUTH = {"x-api-key": API_KEY} FILE_PATH = "observation.fil" file_size = os.path.getsize(FILE_PATH) # 1. Get upload URL resp = requests.post(f"{API}/upload-url", json={"filename": os.path.basename(FILE_PATH), "file_size": file_size}, headers=HEADERS) upload = resp.json() file_key = upload["file_key"] # 2. Upload file if upload.get("multipart"): # Multipart upload for large files parts = [] with open(FILE_PATH, "rb") as f: for part in upload["parts"]: chunk = f.read(upload["part_size"]) r = requests.put(part["url"], data=chunk) parts.append({"partNumber": part["partNumber"], "etag": r.headers["ETag"]}) requests.post(f"{API}/complete-upload", json={"file_key": file_key, "upload_id": upload["upload_id"], "parts": parts}, headers=HEADERS) print("Multipart upload complete") else: # Single upload for small files with open(FILE_PATH, "rb") as f: requests.put(upload["upload_url"], data=f, headers={"Content-Type": "application/octet-stream"}) print("Upload complete") # 3. Start analysis resp = requests.post(f"{API}/analyze", json={"file_key": file_key, "filename": os.path.basename(FILE_PATH)}, headers=HEADERS) job_id = resp.json()["job_id"] print(f"Job submitted: {job_id}") # 4. Poll for results (recommended: 10-second intervals) while True: status = requests.get(f"{API}/jobs/{job_id}", headers=AUTH).json() print(f" Status: {status['status']}") if status["status"] in ("complete", "failed"): break time.sleep(10) # 5. Process results if status["status"] == "complete": results = status["results"] print(f"\nResults: {results['raw_hits']} raw hits, " f"{results.get('filtered_hits', 0)} filtered, " f"{results.get('candidates', 0)} candidates") print(f"Processing time: {results.get('processing_time_seconds', 0):.1f}s") for c in results.get("top_candidates", [])[:5]: freq_mhz = c["frequency_hz"] / 1e6 print(f" {freq_mhz:.3f} MHz | SNR={c['snr']:.1f} | {c['classification']}") # 6. Download results (wait for exports to be generated) time.sleep(15) for fmt in ["json", "csv", "fits", "waterfall"]: resp = requests.get(f"{API}/jobs/{job_id}/download?format={fmt}", headers=AUTH) if fmt == "csv": # CSV is returned inline with open(f"{job_id[:8]}_signals.csv", "w") as f: f.write(resp.text) print(f" Saved {fmt}: {job_id[:8]}_signals.csv") else: data = resp.json() if "download_url" in data: content = requests.get(data["download_url"]) ext = {"json": "json", "fits": "fits", "waterfall": "png"}[fmt] filename = f"{job_id[:8]}.{ext}" with open(filename, "wb") as f: f.write(content.content) print(f" Saved {fmt}: {filename}") else: print(f"Job failed: {status.get('error', 'Unknown error')}")