icare-sensor-client 0.1.4
Python client library for iCare Sensor Communication Agent API
License: MIT
Requires Python: >=3.8
Keywords: icare, sensor, client, api, healthcare, monitoring

# iCare Sensor Client

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

## Installation

Install the package using pip:

```bash
pip install icare-sensor-client
```

For development installation with testing dependencies:

```bash
pip install icare-sensor-client[dev]
```

Or install from source:

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

## Quick Start

```python
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

```python
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.

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

##### health_check()

Check sensor health status.

```python
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.

```python
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.

```python
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.

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

##### get_status()

Get current sensor status.

```python
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.

```python
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.

```python
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.

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

##### close()

Close the session and cleanup resources.

```python
client.close()
```

### Context Manager Support

The client supports context manager protocol for automatic resource cleanup:

```python
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:

```python
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

```python
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

```python
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

```python
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

```python
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:

```bash
# 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