File size: 2,578 Bytes
33962ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | """Read Laya metadata; --synthetic additionally sends one generated test image.
Never captures a camera, reads personal media, or prints credentials.
"""
import argparse
import asyncio
import base64
import io
import json
import os
from pathlib import Path
import sys
import time
sys.path.insert(0,str(Path(__file__).resolve().parent.parent))
import httpx
from PIL import Image
from config.schema import Provider
from vrag.egress import transport_for
from vrag.providers import Adapter, MAX_RESPONSE, destination
async def check(synthetic):
p=Provider(adapter='laya-vision',base_url_env='LAYA_BASE_URL',
api_key_secret='LAYA_API_KEY' if os.environ.get('LAYA_API_KEY') else None,
model='thaitea/laya-vision',capabilities=['image_understanding','visual_decision'],
trust='approved_sensitive',timeout_ms=60000,concurrency=1)
base=await destination(p,False)
headers={'Authorization':'Bearer '+os.environ[p.api_key_secret]} if p.api_key_secret else {}
metadata={}
async with httpx.AsyncClient(transport=transport_for(base),follow_redirects=False,timeout=15) as client:
for path in ['/health','/ready','/v1/info']:
async with client.stream('GET',base+path,headers=headers) as r:
if r.status_code!=200:raise RuntimeError(f'{path}: HTTP {r.status_code}')
body=bytearray()
async for chunk in r.aiter_bytes():
body.extend(chunk)
if len(body)>MAX_RESPONSE:raise RuntimeError('metadata response too large')
metadata[path]=json.loads(body)
output={'metadata':metadata,'authorization_header_sent':bool(p.api_key_secret),'synthetic_inference_executed':False}
if synthetic:
image=io.BytesIO();Image.new('RGB',(64,64),'white').save(image,format='JPEG')
start=time.monotonic()
result=await Adapter().call('laya',p,{
'image':'data:image/jpeg;base64,'+base64.b64encode(image.getvalue()).decode(),
'text':'Synthetic Visual Mentor release connection test. No camera or personal data.',
'frame_id':0,'camera_id':'synthetic'})
output.update(synthetic_inference_executed=True,elapsed_s=round(time.monotonic()-start,3),output=result.model_dump())
return output
if __name__=='__main__':
parser=argparse.ArgumentParser(description=__doc__)
parser.add_argument('--synthetic',action='store_true',help='Explicitly send one generated image for inference')
args=parser.parse_args()
print(json.dumps(asyncio.run(check(args.synthetic)),indent=2))
|