### Discover Devices and Process Manually

Source: https://github.com/devbis/aiopppp/blob/main/README.md

Implement a device discovery mechanism and manually process found devices. This example shows how to create a `Discovery` instance, define a callback for found devices, and start the discovery process.

```python
import asyncio
from aiopppp import Discovery, JsonSession

def on_disconnect():
    print("Disconnected from the device")

def on_device_found(device):
    print(f"Found device: {device}")
    session = JsonSession(device, on_disconnect=on_disconnect)
    session.start()

async def main():
    discovery = Discovery(remote_addr='255.255.255.255')
    await discovery.discover(on_device_found)

    
asyncio.run(main())

```

--------------------------------

### CLI Usage Examples

Source: https://context7.com/devbis/aiopppp/llms.txt

Examples of how to use the aiopppp command-line interface for various configurations.

```APIDOC
## CLI Usage Examples

### Specify camera IP directly, custom credentials, debug logging
```bash
python -m aiopppp \
    --addr 192.168.1.50 \
    --username admin \
    --password 6666 \
    --log-level DEBUG
```

### Custom local discovery port
```bash
python -m aiopppp --addr 192.168.1.255 --local-discovery-port 34567
```

### Accessing the camera stream and control API
```
# Then open in browser:
# http://localhost:4000
# Each camera appears with:
#   - Live MJPEG stream at /<dev_id>/v
#   - Control commands via POST /<dev_id>/c/<cmd>
```
```

--------------------------------

### Install aiopppp Library

Source: https://github.com/devbis/aiopppp/blob/main/README.md

Install the library using pip. Ensure you have Python 3.7 or higher.

```bash
pip install aiopppp
```

--------------------------------

### High-Level Async Camera Control with Device Class

Source: https://context7.com/devbis/aiopppp/llms.txt

The `Device` class offers a high-level interface for camera control, supporting both `async with` for automatic connection management and manual `connect()`/`close()` calls. After connecting, camera properties like firmware version and signal strength are available in `device.properties`. This example demonstrates starting and stopping video streams and retrieving frames.

```python
import asyncio
from aiopppp import Device
from aiopppp.exceptions import AuthError, NotConnectedError, TimeoutError

async def main():
    # Context manager — auto connects and closes
    try:
        async with Device("192.168.1.50", username="admin", password="6666") as device:
            print("Connected:", device.is_connected)
            print("Properties:", device.properties)
            # {
            #   'tz': -3, 'time': 3950400351, 'icut': 0,
            #   'batValue': 90, 'batStatus': 1,
            #   'sysver': 'HQLS_HQT66DP_20240925 11:06:42',
            #   'mcuver': '1.1.1.1', 'sensor': 'GC0329',
            #   'signal': 100, 'lamp': 1
            # }

            await device.start_video()
            for _ in range(5):
                frame = await device.get_video_frame()
                print(f"Frame #{frame.idx}: {len(frame.data)} bytes")
            await device.stop_video()

    except AuthError as e:
        print("Bad credentials:", e)
    except TimeoutError:
        print("Camera unreachable")
    except NotConnectedError as e:
        print("Lost connection:", e)

asyncio.run(main())
```

--------------------------------

### Run aiopppp CLI Test Server

Source: https://context7.com/devbis/aiopppp/llms.txt

Start the built-in aiohttp HTTP server by running `python -m aiopppp`. This server discovers cameras, connects to them, and provides a web-based MJPEG viewer with controls.

```bash
# Basic usage — broadcasts to 255.255.255.255 with default credentials
python -m aiopppp
```

--------------------------------

### Built-in CLI test server

Source: https://context7.com/devbis/aiopppp/llms.txt

Starts an aiohttp HTTP server on port 4000 serving a browser-based MJPEG viewer with PTZ, light, IR, resolution, and reboot controls.

```APIDOC
## Built-in CLI test server (`python -m aiopppp`)

Discovers cameras on the local network, connects to all found devices, and starts an aiohttp HTTP server on port 4000 serving a browser-based MJPEG viewer with PTZ, light, IR, resolution, and reboot controls.

```bash
# Basic usage — broadcasts to 255.255.255.255 with default credentials
python -m aiopppp
```
```

--------------------------------

### Start, Get, and Stop Video Stream with aiopppp

Source: https://context7.com/devbis/aiopppp/llms.txt

Initiates a video stream from the device, captures a specified number of frames, and then stops the stream. Ensure the device IP address and credentials are correct. The session automatically re-requests video if no frames arrive for 5 seconds and disconnects after 10 seconds of stale video.

```python
import asyncio
from aiopppp import Device

async def save_frames(ip: str, count: int = 10):
    async with Device(ip, username="admin", password="6666") as device:
        await device.start_video()
        print(f"Video started, is_video_requested={device.is_video_requested}")

        for i in range(count):
            frame = await device.get_video_frame()
            filename = f"frame_{i:04d}.jpg"
            with open(filename, "wb") as f:
                f.write(frame.data)
            print(f"Saved {filename} ({len(frame.data)} bytes, idx={frame.idx})")

        await device.stop_video()
        print("Video stopped")

asyncio.run(save_frames("192.168.1.50", count=10))
```

--------------------------------

### Device.start_video() / Device.stop_video() / Device.get_video_frame()

Source: https://context7.com/devbis/aiopppp/llms.txt

Control live video streaming from the camera. Start the stream, retrieve individual JPEG frames, and stop the stream. The library handles frame buffering and session management.

```APIDOC
## `Device.start_video()` / `Device.stop_video()` / `Device.get_video_frame()` — Live video streaming

`start_video()` sends the video-request command to the camera and begins accumulating ordered JPEG frames in an internal `SharedFrameBuffer`. `get_video_frame()` awaits the next complete frame. `stop_video()` flushes all queued data and sends the stop command. The session automatically re-requests video if no frames arrive for 5 seconds and disconnects after 10 seconds of stale video.

### Example Usage
```python
import asyncio
from aiopppp import Device

async def save_frames(ip: str, count: int = 10):
    async with Device(ip, username="admin", password="6666") as device:
        await device.start_video()
        print(f"Video started, is_video_requested={device.is_video_requested}")

        for i in range(count):
            frame = await device.get_video_frame()
            filename = f"frame_{i:04d}.jpg"
            with open(filename, "wb") as f:
                f.write(frame.data)
            print(f"Saved {filename} ({len(frame.data)} bytes, idx={frame.idx})")

        await device.stop_video()
        print("Video stopped")

asyncio.run(save_frames("192.168.1.50", count=10))
```
```

--------------------------------

### Run Test Web Server

Source: https://github.com/devbis/aiopppp/blob/main/README.md

Execute a Python module as a command-line interface to run a test web server. This server streams camera feeds and automatically discovers cameras. Use flags for authentication if required.

```bash
python -m aiopppp -u admin -p 6666

```

--------------------------------

### Run aiopppp CLI with custom IP and credentials

Source: https://context7.com/devbis/aiopppp/llms.txt

Use the aiopppp command-line interface to connect to a specific camera IP address with custom username and password, enabling debug logging.

```bash
python -m aiopppp \
    --addr 192.168.1.50 \
    --username admin \
    --password 6666 \
    --log-level DEBUG
```

--------------------------------

### Connect and Control Device (High-Level)

Source: https://github.com/devbis/aiopppp/blob/main/README.md

Use the high-level `Device` class for simplified connection and control. Ensure the device is within the same network. The `async with` statement handles connection and disconnection.

```python
import asyncio
from aiopppp import Device

async def main():
    async with Device("192.168.1.2") as device:
        print("Connected to the device")
        print("Device info:", device.properties)
        await device.start_video()
        await asyncio.sleep(10)
        await device.stop_video()
    print("Disconnected from the device")
        
    # or 
    
    device = Device("192.168.1.2")
    await device.connect()
    print("Device info:", device.properties)
    await device.close()
    
    
asyncio.run(main())

```

--------------------------------

### Run aiopppp CLI with custom local discovery port

Source: https://context7.com/devbis/aiopppp/llms.txt

Configure the aiopppp CLI to use a custom port for local network discovery of cameras.

```bash
python -m aiopppp --addr 192.168.1.255 --local-discovery-port 34567
```

--------------------------------

### Connect and Control Device (Low-Level Session)

Source: https://github.com/devbis/aiopppp/blob/main/README.md

Utilize low-level session connections for more granular control. This approach involves finding the device, creating a session, and managing its lifecycle. It includes handling device disconnections.

```python
import asyncio
from aiopppp import find_device
from aiopppp.device import make_session
from contextlib import suppress

async def main():
    device = await find_device("192.168.1.2", timeout=20)
    disconnected = asyncio.Event()
    session = make_session(device, on_device_lost=lambda lost_device: disconnected.set())
    session.start()
    await asyncio.wait([session.device_is_ready.wait(), session.main_task], return_when=asyncio.FIRST_COMPLETED)
    if session.main_task.done():
        await session.main_task
        return 
    print("Connected to the device")
    print("Device info:", session.dev_properties)
    session.stop()
    with suppress(asyncio.CancelledError):
        await session.main_task
    print("Disconnected from the device")
    
    
asyncio.run(main())

```

--------------------------------

### Control camera via HTTP REST API - Toggle white light

Source: https://context7.com/devbis/aiopppp/llms.txt

Send a POST request to the camera's REST API to toggle the white light ON. Requires the device ID and base URL.

```bash
DEV_ID="DGOK-123456789-ABC"
BASE="http://localhost:4000"

# Toggle white light ON
curl -s -X POST "$BASE/$DEV_ID/c/toggle-lamp" \
     -H "Content-Type: application/json" \
     -d '{"value": 1}'
```

--------------------------------

### Control Camera with BinarySession

Source: https://context7.com/devbis/aiopppp/llms.txt

Use BinarySession for direct control of cameras using the binary protocol. It supports live video, PTZ, IR-cut, fill-light, resolution setting, and system reboot.

```python
import asyncio
from aiopppp import find_device
from aiopppp.session import make_session, BinarySession

async def main():
    device = await find_device("192.168.1.51", timeout=15)

    session: BinarySession = make_session(
        device,
        on_device_lost=lambda dev: print(f"Lost {dev.dev_id}"),
        login="admin",
        password="admin",
    )
    session.start()

    await asyncio.wait_for(session.device_is_ready.wait(), timeout=15)
    print("Connected, properties:", session.dev_properties)
    # {
    #   'tz': 'UTC+0', 'uptime': 4096, 'devName': 'PTZA-Cam',
    #   'sdStatus': 0, 'p2pStatus': 1, 'batLevel': 100,
    #   'ipAddr': '192.168.1.51', 'mac': '00:11:22:33:44:55',
    #   'mcuver': '1.1.1.1', 'icut': 0
    # }

    # Set video resolution to HD (640x480)
    from aiopppp.const import VideoParamType, VideoResolution
    await session.set_video_param("resolution", VideoResolution.VIDEO_RESOLUTION_HD)

    # Start video stream
    await session.start_video()
    frame = await session.frame_buffer.get()
    print(f"Frame: {len(frame.data)} bytes")
    await session.stop_video()

    # PTZ step
    await session.step_rotate("left")

    # Reboot
    await session.reboot()
    session.stop()

asyncio.run(main())
```

--------------------------------

### Session Factory with make_session

Source: https://context7.com/devbis/aiopppp/llms.txt

Use make_session to automatically instantiate either JsonSession or BinarySession based on the device's protocol. This is the recommended way to create a session.

```python
import asyncio
from aiopppp import find_device
from aiopppp.session import make_session

async def main():
    device = await find_device("192.168.1.50", timeout=10)

    # Automatically picks JsonSession or BinarySession
    session = make_session(
        device=device,
        on_device_lost=lambda dev: print(f"Device {dev.dev_id} lost"),
        login="admin",
        password="6666",
    )
    print(f"Session type: {type(session).__name__}")  # JsonSession or BinarySession

    session.start()
    await asyncio.wait_for(session.device_is_ready.wait(), timeout=15)
    print("Ready:", session.dev_properties)
    session.stop()

asyncio.run(main())
```

--------------------------------

### Exception Handling with aiopppp

Source: https://context7.com/devbis/aiopppp/llms.txt

Demonstrates how to handle various exceptions raised by the aiopppp library during device operations.

```APIDOC
## Exception handling

aiopppp defines a hierarchy of exceptions under `PPPPError` for structured error handling across authentication, connectivity, and command failures.

```python
import asyncio
from aiopppp import Device
from aiopppp.exceptions import (
    PPPPError,           # base class
    AuthError,           # wrong username / password
    NotConnectedError,   # operation on a disconnected device
    AlreadyConnectedError,  # connect() called when already connected
)
from aiopppp.exceptions import CommandResultError  # camera returned error for a command

async def robust_connect(ip: str):
    device = Device(ip, username="admin", password="wrong_pass")
    try:
        await device.connect(timeout=10)
    except AuthError as e:
        print(f"Authentication failed: {e}")
        return
    except TimeoutError:
        print("Camera did not respond in time")
        return
    except AlreadyConnectedError:
        print("Already connected")

    try:
        frame = await device.get_video_frame()
    except NotConnectedError:
        print("Lost connection before frame arrived")
    except PPPPError as e:
        print(f"Generic PPPP error: {e}")
    finally:
        await device.close()

asyncio.run(robust_connect("192.168.1.50"))
# Output (wrong password):
# Authentication failed: Login failed: {'result': 1, ...}
```
```

--------------------------------

### Control camera via HTTP REST API - Set video resolution

Source: https://context7.com/devbis/aiopppp/llms.txt

Send a POST request to the camera's REST API to set the video resolution. Available values include QVGA, VGA, HD, FD, UD. Requires the device ID and base URL.

```bash
# Set video resolution
curl -s -X POST "$BASE/$DEV_ID/c/set-video-param" \
     -H "Content-Type: application/json" \
     -d '{"name": "resolution", "value": "HD"}'
```

--------------------------------

### Control camera via HTTP REST API - Reboot camera

Source: https://context7.com/devbis/aiopppp/llms.txt

Send a POST request to the camera's REST API to initiate a camera reboot. Requires the device ID and base URL.

```bash
# Reboot camera
curl -s -X POST "$BASE/$DEV_ID/c/reboot" \
     -H "Content-Type: application/json" \
     -d '{}'
```

--------------------------------

### find_device(ip_address, timeout=20)

Source: https://context7.com/devbis/aiopppp/llms.txt

Discovers a single PPPP camera by IP address. It broadcasts PPPP LAN-search packets and returns the first DeviceDescriptor that responds within the specified timeout. Raises TimeoutError if no response is received.

```APIDOC
## `find_device(ip_address, timeout=20)` — Discover a single PPPP camera by IP

Broadcasts PPPP LAN-search packets to the given IP (or subnet broadcast) and returns the first `DeviceDescriptor` that responds within the timeout. Raises `TimeoutError` if no response arrives. This is the preferred way to resolve a camera's device ID and encryption type before opening a session manually.

### Parameters
#### Path Parameters
- **ip_address** (string) - Required - The IP address or subnet broadcast to search.
- **timeout** (integer) - Optional - The maximum time in seconds to wait for a response. Defaults to 20.

### Response
#### Success Response
- **DeviceDescriptor** - An object containing device information such as `dev_id`, `addr`, `port`, `is_json`, and `encryption`.

### Request Example
```python
import asyncio
from aiopppp import find_device

async def main():
    try:
        device = await find_device("192.168.1.50", timeout=10)
        print(f"Found: {device.dev_id}")
        print(f"Address: {device.addr}:{device.port}")
        print(f"JSON protocol: {device.is_json}")
        print(f"Encryption: {device.encryption}")
    except TimeoutError:
        print("Camera not found within timeout")

asyncio.run(main())
```

### Response Example
```
Found: DevID(DGOK-123456789-ABC)
Address: 192.168.1.50:32108
JSON protocol: True
Encryption: Encryption.XOR1
```
```

--------------------------------

### Python exception handling for camera connection and operations

Source: https://context7.com/devbis/aiopppp/llms.txt

Demonstrates robust connection and operation handling using aiopppp's custom exceptions like AuthError and NotConnectedError within an asyncio application.

```python
import asyncio
from aio ગpppp import Device
from aiopppp.exceptions import (
    PPPPError,           # base class
    AuthError,           # wrong username / password
    NotConnectedError,   # operation on a disconnected device
    AlreadyConnectedError,  # connect() called when already connected
)
from aiopppp.exceptions import CommandResultError  # camera returned error for a command

async def robust_connect(ip: str):
    device = Device(ip, username="admin", password="wrong_pass")
    try:
        await device.connect(timeout=10)
    except AuthError as e:
        print(f"Authentication failed: {e}")
        return
    except TimeoutError:
        print("Camera did not respond in time")
        return
    except AlreadyConnectedError:
        print("Already connected")

    try:
        frame = await device.get_video_frame()
    except NotConnectedError:
        print("Lost connection before frame arrived")
    except PPPPError as e:
        print(f"Generic PPPP error: {e}")
    finally:
        await device.close()

asyncio.run(robust_connect("192.168.1.50"))
# Output (wrong password):
# Authentication failed: Login failed: {'result': 1, ...}
```

--------------------------------

### make_session

Source: https://context7.com/devbis/aiopppp/llms.txt

Session factory function that inspects device.is_json and returns either a JsonSession or BinarySession. This is the recommended way to instantiate a session.

```APIDOC
## `make_session(device, on_device_lost, login, password)` — Session factory

Inspects `device.is_json` and returns either a `JsonSession` or `BinarySession`. This is the canonical way to instantiate a session without hard-coding the protocol type. Used internally by `Device` and the built-in CLI server.

```python
import asyncio
from aiopppp import find_device
from aiopppp.session import make_session

async def main():
    device = await find_device("192.168.1.50", timeout=10)

    # Automatically picks JsonSession or BinarySession
    session = make_session(
        device=device,
        on_device_lost=lambda dev: print(f"Device {dev.dev_id} lost"),
        login="admin",
        password="6666",
    )
    print(f"Session type: {type(session).__name__}")  # JsonSession or BinarySession

    session.start()
    await asyncio.wait_for(session.device_is_ready.wait(), timeout=15)
    print("Ready:", session.dev_properties)
    session.stop()

asyncio.run(main())
# Output:
# Session type: JsonSession
# Ready: {'tz': -3, 'signal': 100, 'lamp': 1, ...}
```
```

--------------------------------

### Control camera via HTTP REST API - Toggle IR filter

Source: https://context7.com/devbis/aiopppp/llms.txt

Send a POST request to the camera's REST API to toggle the IR filter. Requires the device ID and base URL.

```bash
curl -s -X POST "$BASE/$DEV_ID/c/toggle-ir" \
     -H "Content-Type: application/json" \
     -d '{"value": 1}'
```

--------------------------------

### Control camera via HTTP REST API - Set bitrate

Source: https://context7.com/devbis/aiopppp/llms.txt

Send a POST request to the camera's REST API to set the video bitrate, ranging from 0 to 100. Requires the device ID and base URL.

```bash
# Set bitrate (0–100)
curl -s -X POST "$BASE/$DEV_ID/c/set-video-param" \
     -H "Content-Type: application/json" \
     -d '{"name": "bitrate", "value": 50}'
```

--------------------------------

### Control camera via HTTP REST API - Step rotate

Source: https://context7.com/devbis/aiopppp/llms.txt

Send a POST request to the camera's REST API to perform a step rotation in a specified direction (LEFT, RIGHT, UP, DOWN). Requires the device ID and base URL.

```bash
# Step rotate (single move + stop)
curl -s -X POST "$BASE/$DEV_ID/c/rotate" \
     -H "Content-Type: application/json" \
     -d '{"value": "LEFT"}'
```

--------------------------------

### Control Camera with JsonSession

Source: https://context7.com/devbis/aiopppp/llms.txt

Use JsonSession for direct control of cameras using the JSON-based protocol. Requires a DeviceDescriptor and default credentials are admin/6666.

```python
import asyncio
from contextlib import suppress
from aiopppp import find_device, JsonSession
from aiopppp.session import make_session

async def main():
    # Discover the device
    device = await find_device("192.168.1.50", timeout=15)

    disconnected = asyncio.Event()
    session: JsonSession = make_session(
        device,
        on_device_lost=lambda dev: disconnected.set(),
        login="admin",
        password="6666",
    )
    session.start()

    # Wait until ready
    done, _ = await asyncio.wait(
        [asyncio.ensure_future(session.device_is_ready.wait()), *session.running_tasks()],
        return_when=asyncio.FIRST_COMPLETED,
        timeout=15,
    )

    print("Session state:", session.state)
    print("Camera properties:", session.dev_properties)

    # Pan-tilt-zoom: single step right
    await session.step_rotate("RIGHT")

    # Toggle white light ON, then OFF
    await session.toggle_whitelight(True)
    await asyncio.sleep(2)
    await session.toggle_whitelight(False)

    # Toggle IR cut filter
    await session.toggle_ir(True)

    # Stream video for 3 seconds
    await session.start_video()
    await asyncio.sleep(3)
    frame = await session.frame_buffer.get()
    print(f"Video frame: idx={frame.idx}, size={len(frame.data)} bytes")
    await session.stop_video()

    # Graceful disconnect
    session.stop()
    with suppress(asyncio.CancelledError):
        await session.main_task

asyncio.run(main())
```

--------------------------------

### BinarySession

Source: https://context7.com/devbis/aiopppp/llms.txt

Low-level binary-protocol session control for cameras. Supports live video, PTZ, IR-cut, fill-light, resolution setting, and system reboot. Uses struct-packed payloads with a 4-byte authentication ticket.

```APIDOC
## `BinarySession` — Low-level binary-protocol session

Direct session control for cameras using the binary protocol (devices with prefix PTZA, FTYC, BATE). Supports live video, PTZ, IR-cut, fill-light, resolution setting, and system reboot. The binary protocol uses struct-packed payloads with a 4-byte authentication ticket obtained during login.

```python
import asyncio
from aiopppp import find_device
from aiopppp.session import make_session, BinarySession

async def main():
    device = await find_device("192.168.1.51", timeout=15)

    session: BinarySession = make_session(
        device,
        on_device_lost=lambda dev: print(f"Lost {dev.dev_id}"),
        login="admin",
        password="admin",
    )
    session.start()

    await asyncio.wait_for(session.device_is_ready.wait(), timeout=15)
    print("Connected, properties:", session.dev_properties)
    # {
    #   'tz': 'UTC+0', 'uptime': 4096, 'devName': 'PTZA-Cam',
    #   'sdStatus': 0, 'p2pStatus': 1, 'batLevel': 100,
    #   'ipAddr': '192.168.1.51', 'mac': '00:11:22:33:44:55',
    #   'mcuver': '1.1.1.1', 'icut': 0
    # }

    # Set video resolution to HD (640x480)
    from aiopppp.const import VideoParamType, VideoResolution
    await session.set_video_param("resolution", VideoResolution.VIDEO_RESOLUTION_HD)

    # Start video stream
    await session.start_video()
    frame = await session.frame_buffer.get()
    print(f"Frame: {len(frame.data)} bytes")
    await session.stop_video()

    # PTZ step
    await session.step_rotate("left")

    # Reboot
    await session.reboot()
    session.stop()

asyncio.run(main())
```
```

--------------------------------

### Device Class

Source: https://context7.com/devbis/aiopppp/llms.txt

The high-level async camera controller. It wraps discovery, session creation, authentication, and property retrieval. Supports both the async with context manager pattern and manual connect()/close() lifecycle.

```APIDOC
## `Device` — High-level async camera controller

The recommended entry point for most use cases. Wraps discovery, session creation, authentication, and property retrieval behind a clean async API. Supports both the `async with` context manager pattern and manual `connect()` / `close()` lifecycle. After `connect()`, `device.properties` holds camera metadata (firmware version, battery, signal, timezone, etc.).

### Initialization
- **ip_address** (string) - Required - The IP address of the camera.
- **username** (string) - Optional - The username for authentication. Defaults to 'admin'.
- **password** (string) - Optional - The password for authentication. Defaults to '6666'.
- **timeout** (integer) - Optional - Timeout for connection and operations. Defaults to 20.

### Methods
- **connect()**: Manually establishes a connection to the camera.
- **close()**: Closes the connection to the camera.
- **start_video()**: Starts the video stream.
- **get_video_frame()**: Retrieves a video frame.
- **stop_video()**: Stops the video stream.

### Properties
- **is_connected** (bool) - True if the device is currently connected.
- **properties** (dict) - A dictionary containing camera metadata.

### Request Example (Context Manager)
```python
import asyncio
from aiopppp import Device
from aiopppp.exceptions import AuthError, NotConnectedError, TimeoutError

async def main():
    try:
        async with Device("192.168.1.50", username="admin", password="6666") as device:
            print("Connected:", device.is_connected)
            print("Properties:", device.properties)
            await device.start_video()
            for _ in range(5):
                frame = await device.get_video_frame()
                print(f"Frame #{frame.idx}: {len(frame.data)} bytes")
            await device.stop_video()

    except AuthError as e:
        print("Bad credentials:", e)
    except TimeoutError:
        print("Camera unreachable")
    except NotConnectedError as e:
        print("Lost connection:", e)

asyncio.run(main())
```

### Response Example (Properties)
```json
{
  "tz": -3,
  "time": 3950400351,
  "icut": 0,
  "batValue": 90,
  "batStatus": 1,
  "sysver": "HQLS_HQT66DP_20240925 11:06:42",
  "mcuver": "1.1.1.1",
  "sensor": "GC0329",
  "signal": 100,
  "lamp": 1
}
```

### Response Example (Video Frame)
```
Frame #142: 18432 bytes
Frame #143: 17920 bytes
```
```

--------------------------------

### Discovery

Source: https://context7.com/devbis/aiopppp/llms.txt

Perform manual LAN device discovery to find PPPP cameras on the network. This class scans a subnet or specific IP and calls a provided callback for each device found.

```APIDOC
## `Discovery` — Manual LAN device discovery

Low-level discovery class for scanning a subnet or specific IP for PPPP cameras. Calls a user-supplied callback for each device found, continuously until cancelled. Sends all known encryption variants of the LAN-search packet so it works with both encrypted and unencrypted cameras. Uses UDP port 32108.

### Example Usage
```python
import asyncio
from aiopppp import Discovery
from aiopppp.types import DeviceDescriptor

found_devices: list[DeviceDescriptor] = []

def on_device_found(device: DeviceDescriptor):
    print(f"Discovered: {device.dev_id.dev_id} at {device.addr}:{device.port} (json={device.is_json})")
    found_devices.append(device)

async def scan_network():
    # Broadcast to entire subnet
    discovery = Discovery(remote_addr='192.168.1.255')

    # Run discovery for 5 seconds, then cancel
    task = asyncio.create_task(discovery.discover(on_device_found, period=2))
    await asyncio.sleep(5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

    print(f"\nFound {len(found_devices)} device(s)")
    for d in found_devices:
        print(f"  {d.dev_id.dev_id} -> {d.addr}")

asyncio.run(scan_network())
```
```

--------------------------------

### JsonSession

Source: https://context7.com/devbis/aiopppp/llms.txt

Low-level JSON-protocol session control for cameras. Provides PTZ rotation, lamp/IR/white-light toggling, video streaming, and factory reset. Requires a DeviceDescriptor and default credentials are admin/6666.

```APIDOC
## `JsonSession` — Low-level JSON-protocol session

Direct session control for cameras using the JSON-based protocol (typically XOR1-encrypted devices with prefix DGOK). Provides PTZ rotation, lamp/IR/white-light toggling, video streaming, and factory reset. Requires a `DeviceDescriptor` from `find_device()` or `Discovery`. Default credentials are `admin` / `6666`.

```python
import asyncio
from contextlib import suppress
from aiopppp import find_device, JsonSession
from aiopppp.session import make_session

async def main():
    # Discover the device
    device = await find_device("192.168.1.50", timeout=15)

    disconnected = asyncio.Event()
    session: JsonSession = make_session(
        device,
        on_device_lost=lambda dev: disconnected.set(),
        login="admin",
        password="6666",
    )
    session.start()

    # Wait until ready
    done, _ = await asyncio.wait(
        [asyncio.ensure_future(session.device_is_ready.wait()), *session.running_tasks()],
        return_when=asyncio.FIRST_COMPLETED,
        timeout=15,
    )

    print("Session state:", session.state)
    print("Camera properties:", session.dev_properties)

    # Pan-tilt-zoom: single step right
    await session.step_rotate("RIGHT")

    # Toggle white light ON, then OFF
    await session.toggle_whitelight(True)
    await asyncio.sleep(2)
    await session.toggle_whitelight(False)

    # Toggle IR cut filter
    await session.toggle_ir(True)

    # Stream video for 3 seconds
    await session.start_video()
    await asyncio.sleep(3)
    frame = await session.frame_buffer.get()
    print(f"Video frame: idx={frame.idx}, size={len(frame.data)} bytes")
    await session.stop_video()

    # Graceful disconnect
    session.stop()
    with suppress(asyncio.CancelledError):
        await session.main_task

asyncio.run(main())
```
```

--------------------------------

### Discover PPPP Camera by IP Address

Source: https://context7.com/devbis/aiopppp/llms.txt

Use `find_device` to broadcast PPPP LAN-search packets and identify a camera by its IP address. This function returns a `DeviceDescriptor` containing the camera's ID, address, and protocol type. It raises `TimeoutError` if no response is received within the specified timeout.

```python
import asyncio
from aiopppp import find_device

async def main():
    try:
        device = await find_device("192.168.1.50", timeout=10)
        print(f"Found: {device.dev_id}")       # e.g. DevID(DGOK-123456789-ABC)
        print(f"Address: {device.addr}:{device.port}")
        print(f"JSON protocol: {device.is_json}")
        print(f"Encryption: {device.encryption}")
    except TimeoutError:
        print("Camera not found within timeout")

asyncio.run(main())
```

--------------------------------

### Control camera via HTTP REST API - Stop rotation

Source: https://context7.com/devbis/aiopppp/llms.txt

Send a POST request to the camera's REST API to stop any ongoing rotation. Requires the device ID and base URL.

```bash
# Stop any ongoing rotation
curl -s -X POST "$BASE/$DEV_ID/c/rotate-stop" \
     -H "Content-Type: application/json" \
     -d '{}'
```

--------------------------------

### Manual LAN Device Discovery with aiopppp

Source: https://context7.com/devbis/aiopppp/llms.txt

Performs a manual scan of a subnet or specific IP for PPPP cameras using UDP port 32108. It calls a user-supplied callback for each device found and sends all known encryption variants of the LAN-search packet. The discovery runs continuously until cancelled.

```python
import asyncio
from aiopppp import Discovery
from aiopppp.types import DeviceDescriptor

found_devices: list[DeviceDescriptor] = []

def on_device_found(device: DeviceDescriptor):
    print(f"Discovered: {device.dev_id.dev_id} at {device.addr}:{device.port} (json={device.is_json})")
    found_devices.append(device)

async def scan_network():
    # Broadcast to entire subnet
    discovery = Discovery(remote_addr='192.168.1.255')

    # Run discovery for 5 seconds, then cancel
    task = asyncio.create_task(discovery.discover(on_device_found, period=2))
    await asyncio.sleep(5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

    print(f"\nFound {len(found_devices)} device(s)")
    for d in found_devices:
        print(f"  {d.dev_id.dev_id} -> {d.addr}")

asyncio.run(scan_network())
```

--------------------------------

### Device.reboot()

Source: https://context7.com/devbis/aiopppp/llms.txt

Reboots the camera by sending a reboot command. This method works for both JSON-protocol and binary-protocol cameras. Note that the connection will be lost immediately after the command is sent.

```APIDOC
## `Device.reboot()` — Reboot the camera

Sends a reboot command to the camera. Works on both JSON-protocol and binary-protocol cameras. The connection will be lost immediately after the camera reboots.

### Example Usage
```python
import asyncio
from aiopppp import Device

async def main():
    async with Device("192.168.1.50", username="admin", password="6666") as device:
        print("Rebooting camera...")
        await device.reboot()
        print("Reboot command sent — camera is restarting")

asyncio.run(main())
```
```

--------------------------------

### Reboot Camera with aiopppp

Source: https://context7.com/devbis/aiopppp/llms.txt

Sends a reboot command to the camera. This function works with both JSON-protocol and binary-protocol cameras. The connection will be lost immediately after the camera reboots.

```python
import asyncio
from aiopppp import Device

async def main():
    async with Device("192.168.1.50", username="admin", password="6666") as device:
        print("Rebooting camera...")
        await device.reboot()
        print("Reboot command sent — camera is restarting")

asyncio.run(main())
```

--------------------------------

### HTTP REST Control API

Source: https://context7.com/devbis/aiopppp/llms.txt

Control camera functions via HTTP POST requests to the built-in server.

```APIDOC
## HTTP REST control API (from the built-in server)

When the built-in server is running, each discovered camera exposes a REST endpoint for sending commands. All commands are POST requests with a JSON body. The device ID is the camera's `dev_id` string (e.g., `DGOK-123456789-ABC`).

```bash
DEV_ID="DGOK-123456789-ABC"
BASE="http://localhost:4000"
```

### Toggle white light ON
```bash
curl -s -X POST "$BASE/$DEV_ID/c/toggle-lamp" \
     -H "Content-Type: application/json" \
     -d '{"value": 1}'
# {"status": "ok"}
```

### Toggle IR filter
```bash
curl -s -X POST "$BASE/$DEV_ID/c/toggle-ir" \
     -H "Content-Type: application/json" \
     -d '{"value": 1}'
```

### Step rotate (single move + stop)
```bash
curl -s -X POST "$BASE/$DEV_ID/c/rotate" \
     -H "Content-Type: application/json" \
     -d '{"value": "LEFT"}'
# value: LEFT | RIGHT | UP | DOWN
```

### Stop any ongoing rotation
```bash
curl -s -X POST "$BASE/$DEV_ID/c/rotate-stop" \
     -H "Content-Type: application/json" \
     -d '{}'
```

### Set video resolution
```bash
curl -s -X POST "$BASE/$DEV_ID/c/set-video-param" \
     -H "Content-Type: application/json" \
     -d '{"name": "resolution", "value": "HD"}'
# value: QVGA | VGA | HD | FD | UD
```

### Set bitrate (0–100)
```bash
curl -s -X POST "$BASE/$DEV_ID/c/set-video-param" \
     -H "Content-Type: application/json" \
     -d '{"name": "bitrate", "value": 50}
```

### Reboot camera
```bash
curl -s -X POST "$BASE/$DEV_ID/c/reboot" \
     -H "Content-Type: application/json" \
     -d '{}'
```

### Live MJPEG stream (open in browser or VLC)
```bash
curl -s "$BASE/$DEV_ID/v" --output stream.mjpeg
# Or open: http://localhost:4000/DGOK-123456789-ABC/v
```
```

--------------------------------

### Capture live MJPEG stream via HTTP

Source: https://context7.com/devbis/aiopppp/llms.txt

Use curl to capture the live MJPEG stream from the camera's /v endpoint or open the URL directly in a browser or media player.

```bash
# Live MJPEG stream (open in browser or VLC)
curl -s "$BASE/$DEV_ID/v" --output stream.mjpeg
# Or open: http://localhost:4000/DGOK-123456789-ABC/v
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.