init
This commit is contained in:
122
scripts/client_example.py
Normal file
122
scripts/client_example.py
Normal file
@ -0,0 +1,122 @@
|
||||
#!/usr/bin/env python3
|
||||
# /// script
|
||||
# requires-python = ">=3.10"
|
||||
# dependencies = [
|
||||
# "httpx",
|
||||
# ]
|
||||
# ///
|
||||
"""
|
||||
Example client for the Human Pose Estimation server.
|
||||
|
||||
Usage:
|
||||
python client_example.py <image_path> [--url URL]
|
||||
uv run client_example.py <image_path> [--url URL]
|
||||
|
||||
Examples:
|
||||
python client_example.py photo.jpg
|
||||
uv run client_example.py photo.png --url https://api.pose.weihua-iot.cn/hpe
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
def detect_poses(
|
||||
image_path: Path, url: str = "https://api.pose.weihua-iot.cn/hpe"
|
||||
) -> dict | None:
|
||||
"""
|
||||
Send an image to the HPE server and return pose detection results.
|
||||
|
||||
Args:
|
||||
image_path: Path to the image file (JPEG or PNG)
|
||||
url: HPE server endpoint URL
|
||||
|
||||
Returns:
|
||||
Dictionary with pose detection info, or None if no poses detected
|
||||
"""
|
||||
# Determine content type from file extension
|
||||
suffix = image_path.suffix.lower()
|
||||
content_type_map = {
|
||||
".jpg": "image/jpeg",
|
||||
".jpeg": "image/jpeg",
|
||||
".png": "image/png",
|
||||
}
|
||||
|
||||
content_type = content_type_map.get(suffix)
|
||||
if content_type is None:
|
||||
raise ValueError(f"Unsupported image format: {suffix}. Use JPEG or PNG.")
|
||||
|
||||
# Read image bytes
|
||||
image_bytes = image_path.read_bytes()
|
||||
|
||||
# Send request
|
||||
with httpx.Client(timeout=60.0) as client:
|
||||
response = client.post(
|
||||
url,
|
||||
content=image_bytes,
|
||||
headers={"Content-Type": content_type},
|
||||
)
|
||||
|
||||
# Handle response
|
||||
response.raise_for_status()
|
||||
|
||||
if response.status_code == 200 and len(response.content) == 0:
|
||||
# No poses detected
|
||||
return None
|
||||
|
||||
return response.json()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Human Pose Estimation client example")
|
||||
parser.add_argument(
|
||||
"image",
|
||||
type=Path,
|
||||
help="Path to the image file (JPEG or PNG)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--url",
|
||||
default="https://api.pose.weihua-iot.cn/hpe",
|
||||
help="HPE server endpoint URL (default: https://api.pose.weihua-iot.cn/hpe)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
image_path: Path = args.image
|
||||
|
||||
if not image_path.exists():
|
||||
print(f"Error: Image file not found: {image_path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
result = detect_poses(image_path, args.url)
|
||||
except httpx.HTTPStatusError as e:
|
||||
print(f"HTTP Error: {e.response.status_code}", file=sys.stderr)
|
||||
try:
|
||||
error_detail = e.response.json()
|
||||
print(
|
||||
f" {error_detail.get('error')}: {error_detail.get('detail')}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
except Exception:
|
||||
print(f" {e.response.text}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except httpx.RequestError as e:
|
||||
print(f"Request Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except ValueError as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if result is None:
|
||||
print("No poses detected in the image.")
|
||||
else:
|
||||
import json
|
||||
|
||||
print(json.dumps(result, indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user