383201cacd
## v0.9.20260325_144654 ### Features - API Key Authentication System - Job Worker System - V2 Backup Versioning ### Bug Fixes - get_processor_results_by_job column mapping Co-authored-by: OpenCode
85 lines
2.0 KiB
Python
85 lines
2.0 KiB
Python
#!/opt/homebrew/bin/python3.11
|
|
"""
|
|
Object search test using PostgreSQL JSON queries
|
|
"""
|
|
|
|
import time
|
|
import psycopg2
|
|
|
|
|
|
VIDEO_UUID = "39567a0eb16f39fd"
|
|
|
|
POSTGRES_CONFIG = {
|
|
"host": "localhost",
|
|
"port": 5432,
|
|
"user": "accusys",
|
|
"password": "Test3200",
|
|
"database": "momentry",
|
|
}
|
|
|
|
|
|
def test_object_search():
|
|
"""Test object search using PostgreSQL JSON queries"""
|
|
results = {}
|
|
test_objects = ["person", "car", "clock", "tie", "chair", "bottle", "cup", "book"]
|
|
|
|
conn = psycopg2.connect(**POSTGRES_CONFIG)
|
|
cur = conn.cursor()
|
|
|
|
for obj in test_objects:
|
|
start = time.time()
|
|
|
|
# Query chunks that have this object in YOLO metadata
|
|
query = """
|
|
SELECT chunk_id, start_time, end_time
|
|
FROM chunks
|
|
WHERE uuid = %s
|
|
AND chunk_type = 'sentence'
|
|
AND metadata IS NOT NULL
|
|
AND metadata->'yolo'->'objects' ? %s
|
|
ORDER BY chunk_index
|
|
LIMIT 10
|
|
"""
|
|
cur.execute(query, (VIDEO_UUID, obj))
|
|
rows = cur.fetchall()
|
|
|
|
elapsed = (time.time() - start) * 1000
|
|
results[obj] = {
|
|
"ms": round(elapsed, 2),
|
|
"chunks": len(rows),
|
|
"sample": [
|
|
{"id": r[0], "time": f"{r[1]:.1f}-{r[2]:.1f}"} for r in rows[:3]
|
|
],
|
|
}
|
|
print(f"Object '{obj}': {elapsed:.2f}ms, {len(rows)} chunks")
|
|
|
|
cur.close()
|
|
conn.close()
|
|
|
|
return results
|
|
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print("Object Search Test (Priority c)")
|
|
print("=" * 60)
|
|
|
|
results = test_object_search()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("Summary")
|
|
print("=" * 60)
|
|
print(f"\n{'Object':<20} | {'Time (ms)':<12} | {'Chunks'}")
|
|
print("-" * 50)
|
|
for obj, data in results.items():
|
|
print(f"{obj:<20} | {data['ms']:<12.1f} | {data['chunks']}")
|
|
|
|
print("\nSample results:")
|
|
for obj, data in results.items():
|
|
if data["sample"]:
|
|
print(f" {obj}: {data['sample']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|