pnex-publisher 0.1.1
Client for publishing data to PNEX platform
License: MIT
Requires Python: >=3.8

# PNEX Publisher

Python client for publishing sensor data to PNEX IoT platform with ChaCha20 encryption.

## Features

- **Easy-to-use API**: Simple interface for publishing sensor data
- **ChaCha20 Encryption**: End-to-end encryption matching PNEX Server implementation
- **Automatic Reconnection**: Handles connection drops with configurable retry logic
- **Heartbeat Support**: PING/PONG for device status tracking
- **Flexible Configuration**: Environment variables, config files, or constructor arguments
- **Async Support**: Built on asyncio for high-performance I/O
- **CLI Tool**: Command-line interface for quick testing and scripting

## Installation

### Using UV (recommended)

```bash
# Install uv if not already installed
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install pnex-publisher
uv pip install pnex-publisher
```

### Using pip

```bash
pip install pnex-publisher
```

## Quick Start

### Basic Usage

```python
import asyncio
from pnex_publisher import PNEXPublisher

async def main():
    # Create publisher with your credentials
    publisher = PNEXPublisher(
        device_token="your-device-token",
        device_id="sensor-001",
        encryption_key="base64-32-byte-key",
        host="localhost:8000",  # or your PNEX server
    )

    # Connect to server
    await publisher.connect()

    # Publish sensor data
    await publisher.publish("temperature", 23.5)
    await publisher.publish("humidity", 65.2)

    # Disconnect
    await publisher.disconnect()

asyncio.run(main())
```

### Using Context Manager

```python
import asyncio
from pnex_publisher import PNEXPublisher

async def main():
    async with PNEXPublisher(
        device_token="your-token",
        device_id="sensor-001",
        encryption_key="your-key",
    ) as publisher:
        # Publisher automatically connects here
        await publisher.publish("temperature", 23.5)
        await publisher.publish("humidity", 65.2)
        # Publisher automatically disconnects here

asyncio.run(main())
```

### Batch Publishing

```python
async def publish_multiple(publisher):
    measurements = {
        "temperature": 23.5,
        "humidity": 65.2,
        "pressure": 1013.2,
        "battery": 87.5,
    }

    results = await publisher.publish_batch(measurements)
    print(f"Published {sum(results.values())}/{len(results)} measurements")
```

### Continuous Publication (Main Usage Pattern)

For IoT applications, continuous data publication is the primary use case. Here's an example that publishes sensor data every 2 seconds:

```python
import asyncio
import random
from pnex_publisher import PNEXPublisher

async def continuous_publication():
    """Continuous sensor data publication every 2 seconds."""
    async with PNEXPublisher(
        device_token="your-device-token",
        device_id="sensor-001",
        encryption_key="base64-32-byte-key",
        host="localhost:8000",
        auto_reconnect=True,  # Enable automatic reconnection
        enable_heartbeat=True,  # Keep connection alive
    ) as publisher:

        print("Starting continuous data publication (every 2 seconds)...")
        print("Press Ctrl+C to stop")

        try:
            while True:
                # Simulate sensor readings
                temperature = 20.0 + random.uniform(-5.0, 5.0)
                humidity = 50.0 + random.uniform(-10.0, 10.0)
                battery = 100.0 - random.uniform(0.0, 0.1)

                # Publish sensor data
                await publisher.publish("temperature", temperature)
                await publisher.publish("humidity", humidity)
                await publisher.publish("battery", battery)

                print(f"Published: temp={temperature:.1f}°C, humidity={humidity:.1f}%, battery={battery:.1f}%")

                # Wait 2 seconds before next publication
                await asyncio.sleep(2)

        except asyncio.CancelledError:
            print("Publication stopped by user")
        except KeyboardInterrupt:
            print("Publication interrupted")
        finally:
            print("Continuous publication ended")

# Run the continuous publication
asyncio.run(continuous_publication())
```

**Key Features for Continuous Publication:**
- `auto_reconnect=True`: Automatically reconnects if connection drops
- `enable_heartbeat=True`: Maintains connection with PING/PONG heartbeats
- `asyncio.sleep(2)`: Publishes data every 2 seconds
- Context manager (`async with`): Ensures proper connection cleanup
- Error handling: Gracefully handles interruptions

## Configuration

### Environment Variables

All configuration can be set via environment variables with `PNEX_` prefix:

```bash
export PNEX_DEVICE_TOKEN="your-device-token"
export PNEX_DEVICE_ID="sensor-001"
export PNEX_ENCRYPTION_KEY="base64-32-byte-key"
export PNEX_HOST="localhost:8000"
export PNEX_USE_TLS="false"
export PNEX_PING_INTERVAL="5"
```

### Configuration File

Create a `.env` file:

```env
PNEX_DEVICE_TOKEN=your-device-token
PNEX_DEVICE_ID=sensor-001
PNEX_ENCRYPTION_KEY=base64-32-byte-key
PNEX_HOST=localhost:8000
PNEX_USE_TLS=false
PNEX_PING_INTERVAL=5
```

### Programmatic Configuration

```python
from pnex_publisher import PNEXConfig, PNEXPublisher

# Create configuration
config = PNEXConfig(
    device_token="your-token",
    device_id="sensor-001",
    encryption_key="your-key",
    host="localhost:8000",
    use_tls=False,
    ping_interval=5,
    # ... other options
)

# Create publisher with config
publisher = PNEXPublisher(config=config)
```

## Command Line Interface

The package includes a CLI tool `pnex-publish`:

### Publish Single Measurement

```bash
pnex-publish --token YOUR_TOKEN --id SENSOR-001 --key YOUR_KEY temperature 23.5
```

### Publish Multiple Measurements

```bash
pnex-publish --token YOUR_TOKEN --id SENSOR-001 --key YOUR_KEY \
  --batch "temperature=23.5,humidity=65.0,pressure=1013.2"
```

### Test Connection

```bash
pnex-publish --token YOUR_TOKEN --id SENSOR-001 --key YOUR_KEY --test
```

### Using Environment Variables

```bash
export PNEX_DEVICE_TOKEN=YOUR_TOKEN
export PNEX_DEVICE_ID=SENSOR-001
export PNEX_ENCRYPTION_KEY=YOUR_KEY

pnex-publish temperature 23.5
```

## API Reference

### PNEXPublisher Class

#### Constructor

```python
PNEXPublisher(
    device_token: str,
    device_id: str,
    encryption_key: str,
    host: str = "localhost:8000",
    use_tls: bool = False,
    predefined_device_type: str = "custom_sensor",
    ping_interval: int = 5,
    publish_timeout: float = 5.0,
    reconnect_delay: int = 5,
    max_reconnect_attempts: int = 10,
    auto_reconnect: bool = True,
    enable_heartbeat: bool = True,
    validate_key_on_start: bool = True,
    config: Optional[PNEXConfig] = None,
)
```

#### Methods

- `async connect()`: Connect to PNEX Server
- `async disconnect()`: Disconnect from server
- `async publish(measurement_name, value, timeout=None)`: Publish single measurement
- `async publish_batch(measurements, timeout=None)`: Publish multiple measurements
- `__aenter__()`, `__aexit__()`: Async context manager support

#### Properties

- `connected`: Whether publisher is connected
- `connection_info`: Dictionary with connection information
- `config`: PNEXConfig instance

### PNEXConfig Class

Configuration management with pydantic-settings support.

#### Creation Methods

- `PNEXConfig.from_env()`: Load from environment variables
- `PNEXConfig.from_dict(data)`: Load from dictionary
- Constructor: Direct initialization

#### Properties

- `ws_url`: WebSocket URL (ws:// or wss://)
- `full_websocket_url`: Complete URL with query parameters
- `websocket_params`: Dictionary of connection parameters
- `to_dict()`: Convert to dictionary

### ChaChaEncryptor Class

Encryption utilities matching PNEX Server implementation.

#### Methods

- `encrypt(plaintext, key_b64)`: Encrypt plaintext
- `decrypt(ciphertext_b64, key_b64)`: Decrypt ciphertext
- `encrypt_sensor_data(measurement_name, value, key_b64)`: Encrypt sensor measurement
- `validate_key(key_b64)`: Validate encryption key
- `generate_key()`: Generate random 32-byte key
- `is_valid_key(key_b64)`: Check if key is valid

## Examples

See the `examples/` directory for complete examples:

- `basic_usage.py`: Basic publisher usage
- `context_manager.py`: Using async context manager
- `environment_config.py`: Configuration with environment variables

## Encryption Details

The publisher uses ChaCha20 encryption (RFC 7539) matching the PNEX Server implementation:

- **Algorithm**: ChaCha20
- **Key Size**: 32 bytes (256 bits)
- **Nonce Size**: 12 bytes (96 bits)
- **Key Format**: Base64-encoded (44 characters)
- **Message Format**: `Base64([12-byte nonce] + [ChaCha20(ciphertext)])`
- **Plaintext Format**: `measurement_name=value`

## Requirements

- Python 3.8+
- websockets >= 12.0
- pycryptodome >= 3.20.0
- pydantic >= 2.0.0
- pydantic-settings >= 2.0.0

## Development

### Setup Development Environment

```bash
# Clone repository
git clone https://github.com/Pnex/py-publisher.git
cd py-publisher

# Install with dev dependencies
uv pip install -e ".[dev]"

# Run tests
pytest

# Format code
black .
ruff check --fix
```

### Running Tests

```bash
# Run all tests
pytest

# Run with coverage
pytest --cov=pnex_publisher --cov-report=html

# Run specific test file
pytest tests/test_publisher.py -v
```

## License

MIT License - see LICENSE file for details.

## Support

- GitHub Issues: https://github.com/Pnex/py-publisher/issues
- Documentation: https://pnex.github.io/pnex-publisher/
- PNEX Server: https://github.com/pnex/pnex-server

## Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.