icare-sensor-client 0.1.4

Python client library for iCare Sensor Communication Agent API

iCare Sensor Client

A Python client library for interacting with the iCare Sensor Communication Agent API.

Installation

Install the package using pip:

pip install icare-sensor-client

For development installation with testing dependencies:

pip install icare-sensor-client[dev]

Or install from source:

git clone https://github.com/icare/sensor-client.git
cd sensor-client/icare_sensor_client
pip install -e .

Quick Start

from icare_sensor_client import SensorClient

# Create a client instance
client = SensorClient("http://localhost:8000")

# Get service information
info = client.get_info()
print(f"{info['service']} v{info['version']}")

# Check sensor health
health = client.health_check()
if health['ok']:
    print(f"Sensor {health['sensor_id']} is healthy")

# Send an alert
client.send_alert(
    content="Patient fall detected",
    severity="high",
    source_type="camera",
    source_id=1
)

API Reference

SensorClient

The main client class for interacting with the sensor agent API.

Constructor

SensorClient(base_url="http://localhost:8000", timeout=10, cache_ttl=0.1)

Parameters:

• base_url (str): Base URL of the agent API (default: "http://localhost:8000")
• timeout (int): Request timeout in seconds (default: 10)
• cache_ttl (float): Cache time-to-live in seconds for stream state (default: 0.1 = 100ms). Set to 0 to disable caching.

Methods

get_info()

Get API service information.

info = client.get_info()
# Returns: {'service': 'iCare Sensor Communication Agent', 'version': '1.0.0', ...}

health_check()

Check sensor health status.

health = client.health_check()
# Returns: {'ok': True, 'sensor_id': 'sensor-123'}

Raises:

• SensorNotInitializedError: If sensor is not initialized

send_alert(content, severity, source_type, source_id)

Send an alert through the sensor component.

client.send_alert(
    content="Patient fall detected",
    severity="high",
    source_type="camera",
    source_id=1
)
# Returns: {'ok': True, 'message': 'Alert sent successfully'}

Parameters:

• content (str): Alert message content
• severity (str): Alert severity level (e.g., "low", "medium", "high", "critical")
• source_type (str): Type of source generating the alert (e.g., "camera", "sensor")
• source_id (int): Unique identifier of the source

send_coordinates(patient_coords, bed_coords, dimension)

Send coordinate data through the sensor component.

client.send_coordinates(
    patient_coords={"x": 120, "y": 180, "z": 0},
    bed_coords={"x": 50, "y": 150, "z": 0},
    dimension={"width": 500, "height": 400, "depth": 300}
)
# Returns: {'ok': True, 'message': 'Coordinates sent successfully'}

Parameters:

• patient_coords (dict): Patient coordinates with x, y, z keys
• bed_coords (dict): Bed/furniture coordinates with x, y, z keys
• dimension (dict): Room dimensions with width, height, depth keys

upload_video()

Trigger video file upload to hub.

result = client.upload_video()
# Returns: {'ok': True, 'message': 'Video upload initiated'}

get_status()

Get current sensor status.

status = client.get_status()
print(f"Sensor ID: {status.sensor_id}")
print(f"Stream State: {status.shared_variable}")

Returns: SensorStatus object with:

• sensor_id (str): Sensor identifier
• shared_variable (int): Stream state (0=stopped, 1=active, 2=timed out)
• reset_scene (bool): Scene reset flag
• update_available (bool): Update availability flag

get_stream_state()

Get the current stream state with client-side caching.

stream_state = client.get_stream_state()
if stream_state == 1:
    print("Stream is ACTIVE")

Returns: Stream state value:

• 0: Not streaming (initial state)
• 1: Streaming started/active
• 2: Stream stopped (timed out after 60 seconds)
• None: Not available

get_reset_scene()

Get the current reset scene flag with client-side caching.

reset_requested = client.get_reset_scene()
if reset_requested:
    print("Scene reset requested!")
    client.reset_scene()  # Confirm the reset

Returns: Reset scene flag:

• True: Scene reset is requested
• False: No reset requested
• None: Not available

reset_scene()

Request scene reset.

client.reset_scene()
# Returns: {'ok': True, 'message': 'Scene reset confirmed'}

close()

Close the session and cleanup resources.

client.close()

Context Manager Support

The client supports context manager protocol for automatic resource cleanup:

with SensorClient("http://localhost:8000") as client:
    info = client.get_info()
    # Client will be automatically closed when exiting the context

Error Handling

The library provides specific exception classes for different error scenarios:

from icare_sensor_client import (
    SensorClient,
    SensorClientError,
    SensorNotInitializedError,
    ConnectionError
)

try:
    client = SensorClient("http://localhost:8000")
    health = client.health_check()
except SensorNotInitializedError:
    print("Sensor is not initialized on the server")
except ConnectionError as e:
    print(f"Unable to connect to sensor agent: {e}")
except SensorClientError as e:
    print(f"API error: {e}")

Exception Hierarchy

• SensorClientError - Base exception for all client errors
  • SensorNotInitializedError - Sensor is not initialized on the server
  • ConnectionError - Unable to connect to the sensor agent

Examples

Basic Usage

from icare_sensor_client import SensorClient

# Create client
client = SensorClient("http://10.42.0.1:8000")

# Get service info
info = client.get_info()
print(f"Connected to {info['service']}")

# Check health
health = client.health_check()
print(f"Sensor {health['sensor_id']} status: {'OK' if health['ok'] else 'ERROR'}")

# Send alert
client.send_alert(
    content="Motion detected in room 101",
    severity="medium",
    source_type="motion_sensor",
    source_id=101
)

# Clean up
client.close()

Stream State and Reset Scene Polling

from icare_sensor_client import SensorClient
import time

# Create client with 50ms cache for high-frequency polling
client = SensorClient("http://localhost:8000", cache_ttl=0.05)

try:
    while True:
        # Get stream state (uses cache if called within 50ms)
        state = client.get_stream_state()
        
        if state == 0:
            print("Stream: STOPPED")
        elif state == 1:
            print("Stream: ACTIVE")
        elif state == 2:
            print("Stream: TIMED OUT")
        else:
            print("Stream: UNKNOWN")
        
        # Check if scene reset is requested (also cached)
        if client.get_reset_scene():
            print("Scene reset requested - confirming...")
            client.reset_scene()
        
        time.sleep(0.1)  # Poll every 100ms
        
except KeyboardInterrupt:
    print("Stopping...")
finally:
    client.close()

Sending Coordinates

from icare_sensor_client import SensorClient

client = SensorClient("http://localhost:8000")

# Send patient and bed coordinates
client.send_coordinates(
    patient_coords={"x": 120.5, "y": 180.3, "z": 0.0},
    bed_coords={"x": 50.0, "y": 150.0, "z": 0.0},
    dimension={"width": 500.0, "height": 400.0, "depth": 300.0}
)

client.close()

Error Handling

from icare_sensor_client import (
    SensorClient,
    SensorClientError,
    SensorNotInitializedError,
    ConnectionError
)

def monitor_sensor(base_url):
    try:
        with SensorClient(base_url, timeout=5) as client:
            # Check health
            health = client.health_check()
            
            if not health['ok']:
                print("Sensor health check failed")
                return False
            
            # Get status
            status = client.get_status()
            print(f"Sensor {status.sensor_id} is operational")
            
            return True
            
    except SensorNotInitializedError:
        print("ERROR: Sensor is not initialized. Please initialize the sensor first.")
        return False
        
    except ConnectionError as e:
        print(f"ERROR: Cannot connect to sensor agent at {base_url}")
        print(f"Details: {e}")
        return False
        
    except SensorClientError as e:
        print(f"ERROR: API error occurred: {e}")
        return False

# Use the function
success = monitor_sensor("http://10.42.0.1:8000")

Requirements

• Python 3.8 or higher
• requests >= 2.31.0

Development

To set up for development:

# Clone the repository
git clone https://github.com/icare/sensor-client.git
cd sensor-client/icare_sensor_client

# Install in editable mode with dev dependencies
pip install -e .[dev]

# Run tests
pytest

# Run tests with coverage
pytest --cov=icare_sensor_client tests/

License

MIT License - see LICENSE file for details.

Support

For issues and questions:

• GitHub Issues: https://github.com/icare/sensor-client/issues
• Documentation: https://github.com/icare/sensor-client#readme