Skip to content

tapo_manager.py — Cellar Light Control Module

Overview

tapo_manager.py provides local network control of the cellar illumination system used by the Wine Platform vision pipeline.

The module controls a Tapo P110 smart plug which powers the cellar light.
Lighting control is necessary to ensure stable illumination conditions for camera-based bottle detection.

The module is designed for deterministic and reliable operation inside the automated pipeline.

Key design characteristics:

  • fixed-IP connection (no network discovery)
  • synchronous and asynchronous APIs
  • centralized configuration via pince_shelf.ini
  • verification of device state after switching
  • full integration with the stock detection pipeline

Role in Vision Pipeline

The module is used by the pipeline runtime orchestrator:

stock_runtime.py

Its responsibility is to control illumination before and after image acquisition.

Vision Pipeline Context

flowchart LR

subgraph VisionPipeline
SR["stock_runtime.py"]
SNAP["take_clean_snapshot.py"]
CROP["crop_with_json.py"]
ONNX["onnx_batch.py"]
DBC["DB_compare.py"]
end

subgraph Hardware
CAM["Tapo Camera (RTSP)"]
PLUG["Tapo P110 Smart Plug"]
LIGHT["Cellar Light"]
end

SR --> PLUG
PLUG --> LIGHT

SR --> SNAP
SNAP --> CROP
CROP --> ONNX
ONNX --> DBC

CAM --> SNAP

The light must be turned ON before snapshot acquisition and OFF after pipeline completion.

Design Principles

The module follows several design constraints.

1. Fixed IP Control

Instead of using network discovery, the plug is accessed using a known IP address.

Advantages:

  • faster connection
  • deterministic behaviour
  • avoids UDP broadcast discovery
  • works reliably in isolated networks

2. Configuration Driven

All parameters are loaded from:

pince_shelf.ini

via:

load_pince_config()

This avoids command-line arguments or runtime parameters.

3. Safe Device State Verification

After each state change:

device.update()

is called and the final state is verified.

If the state does not match the requested state:

RuntimeError

is raised.

Module Dependencies

External Libraries

python-kasa

Used for controlling TP-Link / Tapo devices.

Main classes used:

Credentials
Discover
KasaException

Project Modules

pince_shelf.config.settings
pince_shelf.utils.paths

Used for loading configuration.

Configuration Parameters

The following values are expected inside pince_shelf.ini.

Parameter Purpose
tapo_plug_ip IP address of Tapo plug
tapo_username Tapo account username
tapo_password Tapo account password
tapo_alias optional alias for logging
tapo_stabil illumination stabilization delay

Example:

[tapo]
plug_ip = 192.168.1.50
username = user@email.com
password = password
alias = cellar_light
stabil = 2

Module Initialization

The module loads configuration immediately:

ini_path = CONF_DIR / "pince_shelf.ini"
cfg = load_pince_config(ini_path)

This provides a global configuration instance.

Functions accept an optional configuration parameter to support dependency injection.

Configuration Helper

_get_cfg()

def _get_cfg(cfg_in: Optional[PinceConfig]) -> PinceConfig

Purpose:

Return configuration provided by caller or fallback to global config.

This enables:

  • unit testing
  • alternative runtime configurations
  • modular integration

Stabilization Delay

get_light_stabilize_seconds()

Purpose:

Return the configured stabilization delay after the light is switched on.

Implementation:

return max(0, int(c.tapo_stabil or 0))

This prevents negative delays.

Credential Builder

_build_credentials()

Purpose:

Construct authentication credentials for the Tapo device.

Logic:

if username AND password exist
    return Credentials
else
    return None

The device can sometimes be accessed without credentials depending on firmware.

Device Connection

_connect_plug()

Async function responsible for establishing communication with the smart plug.

Connection Method

The function uses:

Discover.try_connect_all()

with a specific host IP.

This avoids the UDP discovery mechanism.

Connection Flow

flowchart TD

START["Connect to Tapo plug"]

CHECKIP["Check IP configured"]
CREDS["Build credentials"]
CONNECT["try_connect_all()"]
VERIFY["device.update()"]

START --> CHECKIP
CHECKIP --> CREDS
CREDS --> CONNECT
CONNECT --> VERIFY
VERIFY --> SUCCESS["Device ready"]

Error Handling

Possible failures:

Failure Exception
missing IP ValueError
connection failure RuntimeError
device update failure RuntimeError

Light State Control

_set_light_state()

Primary async function controlling the device state.

Signature:

_set_light_state(target_on: bool)

Execution Steps

flowchart TD

START["Set light state"]

CONNECT["Connect plug"]
CHECK["Read current state"]

CHANGE["Send ON/OFF command"]

UPDATE["Device update"]

VERIFY["Verify state"]

END["Return state"]

Logging

Before change:

Tapo light request:
alias=...
ip=...
from=<previous_state>
to=<target_state>

After change:

Tapo light changed:
alias=...
ip=...
state=<final_state>

State Verification

After switching:

if after_state != target_on
    raise RuntimeError

This ensures the hardware actually changed state.

Public Async API

switch_light_on_async()

Turns light ON.

Returns:

True if device is ON

switch_light_off_async()

Turns light OFF.

Returns:

True if device is OFF

Public Sync API

The pipeline runtime is synchronous.

Therefore wrapper functions exist.

switch_light_on()

Implementation:

asyncio.run(switch_light_on_async())

Returns:

True if light ON

switch_light_off()

Implementation:

asyncio.run(switch_light_off_async())

Returns:

True if light OFF

Integration with Vision Pipeline

Inside the pipeline runtime:

switch_light_on(cfg)
sleep(stabilization_seconds)
take_snapshot()
switch_light_off(cfg)

Runtime Interaction

sequenceDiagram

participant Runtime
participant TapoPlug
participant Light

Runtime->>TapoPlug: connect
TapoPlug-->>Runtime: device ready

Runtime->>TapoPlug: turn_on()
TapoPlug->>Light: power ON

Runtime->>Runtime: wait stabilization

Runtime->>TapoPlug: turn_off()
TapoPlug->>Light: power OFF

Example Runtime Flow

Typical pipeline execution:

  1. pipeline starts
  2. light switched ON
  3. wait stabilization delay
  4. capture snapshot
  5. process images
  6. turn light OFF

Logging Example

2026-03-09 02:00:01 INFO Tapo light request alias=cellar ip=192.168.1.50 from=False to=True
2026-03-09 02:00:01 INFO Tapo light changed alias=cellar ip=192.168.1.50 state=True

Failure Handling

Failures are propagated as:

RuntimeError

The pipeline runtime handles these failures as non-fatal events.

Example:

light failed -> pipeline continues

This ensures detection still runs even if lighting control fails.

Manual Testing Mode

The module includes a simple test entry point.

python tapo_manager.py

Execution flow:

switch_light_on()
sleep(stabilization)
switch_light_off()

This is useful for:

  • hardware testing
  • network verification
  • debugging credentials

Reliability Considerations

The module includes several reliability safeguards:

Mechanism Purpose
fixed IP connection faster device connection
state verification ensure correct state
protocol.close() release network resources
exception wrapping clear error reporting
logging operational visibility

Summary

tapo_manager.py provides robust lighting control for the Wine Platform vision system.

Responsibilities:

  • connect to Tapo smart plug
  • change light state
  • verify hardware response
  • provide async and sync interfaces
  • integrate with automated pipeline runtime

The module ensures that camera snapshots are taken under stable lighting conditions, which is critical for accurate bottle detection.