Add Monitor — single-thread multi-device status monitoring

[jasonacox-sam]

Monitor — Single-Thread, Multi-Device Status Monitoring

Implements the Monitor (TBD) proposal by @3735943886 from PR #649.

What it does

A Monitor class that watches any number of Tuya devices on a single OS thread using selectors (select/poll/epoll). Updates are delivered through callbacks. No asyncio, no per-device threads, no new dependencies.

Key design decisions (from the proposal)

• Only receiving needs to be non-blocking — sending already has nowait=True forms
• Per-device receive buffers with frame reassembly (handles partial frames, back-to-back frames, leading garbage)
• No duplicated protocol logic — reuses parse_header(), unpack_message(), and device._decode_payload()
• One thread owns all socket I/O — eliminates race conditions by construction
• Automatic heartbeats — Monitor sends heartbeat(nowait=True) on a timer per device
• Gateway/cid routing — sub-devices routed to matching child objects automatically
• Thread-safe command queue — mon.send(device, 'set_value', 1, True) enqueues to reactor

Usage

import tinytuya

def on_status(device, result):
    print(device.id, result.get("dps"))

mon = tinytuya.Monitor(on_status=on_status)

for cfg in my_devices:
    d = tinytuya.OutletDevice(cfg["id"], cfg["ip"], cfg["key"],
                              version=3.3, persist=True)
    mon.add(d)           # blocking connect happens here, once

mon.start()              # reactor runs on one daemon thread

# Thread-safe command from any thread
mon.send(d, "set_value", 1, True)

# ... later
mon.stop()

Or drive it from your own loop:

mon = tinytuya.Monitor(on_status=on_status)
mon.add(d)
while True:
    mon.poll(timeout=1.0)

Files changed

• tinytuya/core/Monitor.py — the Monitor class (~400 lines)
• tinytuya/__init__.py — exports Monitor
• examples/monitor_example.py — daemon-thread mode example
• examples/monitor_poll_example.py — manual-poll mode example

Open items (per proposal §7)

• Reconnect logic — when a device drops, reconnecting is blocking (connect + handshake). The current implementation fires on_disconnect but does not auto-reconnect. A reconnect worker or application-level handling is the remaining design decision.

Testing

• Import verification passes (import tinytuya; tinytuya.Monitor)
• Frame reassembly logic mirrors XenonDevice._receive() exactly
• Needs real-device testing (Jason has devices available for live testing)

---

Reference: This is an exploration/testing branch as requested by @jasonacox in PR #649 comment.

[Copilot]

Pull request overview

This PR adds a new tinytuya.Monitor facility intended to monitor many Tuya devices concurrently on a single reactor loop using selectors, delivering decoded updates via callbacks (plus two runnable examples).

Changes:

• Introduces tinytuya/core/Monitor.py implementing the Monitor reactor, per-device buffering, and callback dispatch.
• Exports Monitor from tinytuya/__init__.py and adds two example scripts (threaded and manual-poll modes).

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 7 comments.

File	Description
tinytuya/core/Monitor.py	New selector-based monitoring reactor with per-device buffers, heartbeats, and callback dispatch.
tinytuya/init.py	Exposes Monitor at the package top level (tinytuya.Monitor).
examples/monitor_example.py	Demonstrates daemon-thread reactor usage.
examples/monitor_poll_example.py	Demonstrates manual poll() loop usage.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[3735943886]

Thanks for the PR, @jasonacox-sam !

Before we move forward, I have a few questions and points I'd like to discuss regarding the implementation:

1. Class Naming
   Monitor is currently just a placeholder name for the proposal. Do you have any suggestions for a better alternative? I was thinking of terms like Listener or Streamer. However, please keep in mind that if this new class also needs to handle the send function (to prevent race conditions), Listener might not be the best fit.
   Additionally, please feel free to adjust the class method names if needed, so that they maintain consistency with the existing library conventions. I'd love to hear your thoughts on this.

2. Architecture & Roadmap (Asyncio vs. Current Approach)
   As @jasonacox mentioned, the ultimate long-term goal would be an asyncio-based refactoring. However, that would require major changes to the core logic, which poses risks to backward compatibility and makes it difficult to guarantee identical behavior for synchronous operations.
   Therefore, until we transition to asyncio, I see this select-based multi-device monitoring model as a good stopgap solution for our users. Alternatively, it could even be maintained as a permanent part of the synchronous track even after the asyncio async track becomes available.

3. The Necessity of .send
   Is including the .send method in this class essential? What do you think are the pros and cons of providing a built-in send versus leaving the methods untouched and instead providing a threading.Lock for users to manage direct calls safely on their end?

Looking forward to your feedback!

[jasonacox-sam]

Thanks for the thoughtful questions! I've read through the full proposal and the implementation — here are my thoughts on each point.

---

1. Class Naming

I think Monitor is actually the right name for this class. Here's my reasoning:

• Listener implies a passive, receive-only role — but this class actively manages heartbeats, accepts queued commands via .send(), and orchestrates the reactor loop. It's more than a listener.
• Streamer suggests a continuous data flow abstraction, which isn't quite right either — users get discrete callback events, not a stream interface.
• Monitor captures the essence well: it watches multiple devices, keeps connections alive, and notifies you when something changes. The name is already used in networking contexts (like socket.monitor or select-based monitors) and feels natural alongside Device, OutletDevice, Cloud, etc.

For method naming consistency with the rest of the library, I'd suggest a few adjustments to align with TinyTuya's existing conventions:

• add() / remove() → These are fine and clear. Alternatively, register() / unregister() would also work, but add/remove feel more lightweight — consistent with the library's style.
• send() → This is clear, but consider renaming to command() or dispatch() to reduce confusion with device.send() (which is a lower-level socket method on XenonDevice). The current send(device, method_name, *args) signature could also be misread as "send data to device" rather than "queue a method call."
• start() / stop() / poll() → These are standard and consistent with Python conventions.

My recommendation: keep Monitor, but rename .send() → .command() to avoid ambiguity with the existing device.send().

---

2. Architecture & Roadmap (Asyncio vs. Current Approach)

I fully agree with your assessment. A selectors-based approach is the right interim solution for TinyTuya, for several reasons:

1. Zero disruption to existing code. The synchronous Device.receive() / set_value() API stays untouched. Users who don't need multi-device monitoring see no changes at all.
2. No dependency on asyncio. TinyTuya's user base includes a lot of hobbyists and embedded users running simple scripts. Forcing async/await propagation through their code would be a significant burden. selectors is stdlib and works everywhere select works.
3. Long-term coexistence is realistic. Even if TinyTuya eventually adds an asyncio track, the synchronous path should remain first-class. Many users (home automation scripts, simple polling loops) will never need async. The Monitor class gives those users multi-device monitoring without forcing them into an async mental model.

The one caveat I'd flag — which Copilot also caught in the review comments — is Windows compatibility. The current implementation uses os.pipe() for the self-pipe wake mechanism, and SelectSelector on Windows doesn't support non-socket file descriptors. If Windows support is a goal for Monitor, the wake mechanism should use a socket.socketpair() instead. This is a straightforward fix and doesn't change the architecture at all.

I see Monitor as a permanent part of the synchronous track, not just a stopgap. It fills a real gap in the library (multi-device monitoring without one-thread-per-device) and does so with minimal complexity.

---

3. The Necessity of .send

Yes, a built-in .send (or .command) is essential. Here's why:

The problem with "just give users a threading.Lock":

• Users would need to acquire the lock, call device.set_value(..., nowait=True), and release it — coordinating with the Monitor's reactor thread. This is error-prone and leaks internal implementation details (the reactor thread) to the user.
• If a user forgets nowait=True, the device method will attempt a blocking recv() on the same socket the reactor is watching — causing either a deadlock or corrupted data. The Monitor can't protect against this from the outside.
• The lock approach also doesn't compose: two separate code paths that both need to send commands would need to share the same lock instance. This gets messy fast in any non-trivial application.

What the built-in .send gives us:

• Safety by construction. The command is queued and executed on the reactor thread, so there's no possibility of concurrent socket access. nowait=True is enforced as the default (and as Copilot noted, we should reject nowait=False to prevent blocking the reactor).
• Clean API. mon.command(device, "set_value", 1, True) is one call. No lock management, no nowait remembering.
• Thread-safe by design. Any thread can call .command() at any time — the queue handles serialization.

The trade-off: The current .send() delegates to arbitrary device methods by name (getattr(device, method_name)), which is flexible but a bit magical. If we want a more constrained API, we could expose specific methods like .set_value(device, ...), .set_status(device, ...), etc. But the generic approach is fine for v1 — it keeps the class small and doesn't require updating when new device methods are added.

My recommendation: Keep the built-in send mechanism. Rename to .command() for clarity. Enforce nowait=True (reject nowait=False with a ValueError). This gives users the simplest, safest experience.

---

Summary

Question	Recommendation
Class name	Keep Monitor — it's the best fit
.send() naming	Rename to .command() to avoid confusion with device.send()
Asyncio roadmap	Monitor is the right synchronous-track solution; should be permanent, not just a stopgap
Built-in .send	Essential — don't offload lock management to users. Enforce nowait=True.
Windows compat	Switch os.pipe() → socket.socketpair() if Windows support is needed

I also noted several of Copilot's review comments that are worth addressing (CID routing to child devices, heartbeat implementation simplification, unused imports, and unit tests). Happy to discuss any of those as well.

— Sam ⚙️

[3735943886]

Since _process_buffer directly mirrors the framing logic of XenonDevice._receive(), having duplicated logic might lead to maintenance issues if the protocol changes in the future.

Instead, what if we keep XenonDevice's behavior exactly as is, but refactor its internal parsing logic into a separate helper or utility function? This way, we can safely reuse the code in both places, keeping it DRY while ensuring we don't break any existing functionality.

Additionally, by doing this, we can also reuse the existing unit tests for XenonDevice to verify the shared framing logic, minimizing the effort needed to write brand new tests from scratch. fixes #712 (comment)

[3735943886]

3. The Necessity of .send

Yes, a built-in .send (or .command) is essential. Here's why:

The problem with "just give users a threading.Lock":

• Users would need to acquire the lock, call device.set_value(..., nowait=True), and release it — coordinating with the Monitor's reactor thread. This is error-prone and leaks internal implementation details (the reactor thread) to the user.
• If a user forgets nowait=True, the device method will attempt a blocking recv() on the same socket the reactor is watching — causing either a deadlock or corrupted data. The Monitor can't protect against this from the outside.
• The lock approach also doesn't compose: two separate code paths that both need to send commands would need to share the same lock instance. This gets messy fast in any non-trivial application.

What the built-in .send gives us:

• Safety by construction. The command is queued and executed on the reactor thread, so there's no possibility of concurrent socket access. nowait=True is enforced as the default (and as Copilot noted, we should reject nowait=False to prevent blocking the reactor).
• Clean API. mon.command(device, "set_value", 1, True) is one call. No lock management, no nowait remembering.
• Thread-safe by design. Any thread can call .command() at any time — the queue handles serialization.

The trade-off: The current .send() delegates to arbitrary device methods by name (getattr(device, method_name)), which is flexible but a bit magical. If we want a more constrained API, we could expose specific methods like .set_value(device, ...), .set_status(device, ...), etc. But the generic approach is fine for v1 — it keeps the class small and doesn't require updating when new device methods are added.

My recommendation: Keep the built-in send mechanism. Rename to .command() for clarity. Enforce nowait=True (reject nowait=False with a ValueError). This gives users the simplest, safest experience.

Summary

Question Recommendation
Class name Keep Monitor — it's the best fit
.send() naming Rename to .command() to avoid confusion with device.send()
Asyncio roadmap Monitor is the right synchronous-track solution; should be permanent, not just a stopgap
Built-in .send Essential — don't offload lock management to users. Enforce nowait=True.
Windows compat Switch os.pipe() → socket.socketpair() if Windows support is needed
I also noted several of Copilot's review comments that are worth addressing (CID routing to child devices, heartbeat implementation simplification, unused imports, and unit tests). Happy to discuss any of those as well.

— Sam ⚙️

How about keeping command(), and additionally exposing a dynamic proxy for a cleaner UX?

mon[device].set_value(...)

or when a user adds a device to the monitor, we return the proxy handle right there as the return value:

m = mon.add(device)
m.set_value(...)

Also, I’d love to get @uzlonewolf 's review on this PR. His deep technical insights and creative architectural ideas would be incredibly valuable here before we move forward.

[jasonacox-sam]

Great suggestion on the proxy pattern! I just pushed a commit that implements both approaches:

Proxy via add() return value (recommended)

handle = mon.add(device)
handle.set_value(1, True)  # thread-safe, auto-queued

Proxy via __getitem__

mon[device].set_value(1, True)

The _DeviceProxy object captures any attribute access and enqueues the call through Monitor.command() with nowait=True forced. This gives the clean UX you described while keeping all the safety guarantees.

Other changes in this push

Addressed all of Copilot's review comments:

• .send() → .command() — renamed to avoid confusion with device.send(). Old .send() preserved as a backward-compatible alias.
• nowait=False rejected — command() now raises ValueError if nowait=False is passed, preventing accidental blocking of the reactor.
• os.pipe() → socket.socketpair() — the self-pipe wake mechanism now uses sockets, which works on Windows too (SelectSelector on Windows only supports sockets).
• Removed sock.setblocking(False) — selectors work fine with blocking sockets as long as we only recv() after select() indicates readability. Keeping blocking mode avoids issues with TinyTuya's sendall/retry path.
• Simplified heartbeat — _do_heartbeat() now just calls device.heartbeat(nowait=True) instead of reimplementing encoding/sending. Much cleaner.
• Fixed CID routing — callbacks for gateway sub-devices now receive the child device object (not the gateway), and use the child's registered state when available.
• Removed unused imports — struct and TuyaMessage were unused; socket kept for socketpair().

On the _process_buffer refactor

Your inline comment about extracting the shared framing logic into a utility function is a great idea. I held off on that in this push because it requires refactoring XenonDevice._receive() in core.py — that's a more invasive change that should be a separate commit (or a follow-up PR) to keep the diff reviewable. The goal would be a shared reassemble_frame(buffer) helper that both Monitor._process_buffer() and XenonDevice._receive() call, keeping the protocol logic DRY. Happy to do that as a next step.

I've also requested @uzlonewolf's review as you suggested — his perspective on the socket handling and protocol layer would be really valuable here.

— Sam ⚙️

[3735943886]

We need to implement a reconnect mechanism without blocking. As far as I understand, the current XenonDevice only supports blocking connection initiation, correct?
To handle this, I think we have a few options:

1. Refactor XenonDevice to natively support non-blocking connections.
2. Introduce a single worker thread that processes connection requests via a queue.
3. Spawn a dedicated thread for each individual device connection.
4. Any other suggestions?

Could you please provide the pros and cons for each approach?

[jasonacox-sam]

Great question — reconnect is the last major design decision for this PR. You're right that XenonDevice._get_socket() is blocking (TCP connect + v3.4/3.5 session-key handshake). The reactor can't block for reconnects without stalling status updates and heartbeats for all other devices.

Here's my analysis of each option:

---

Option 1: Refactor XenonDevice for non-blocking connections

Pros:

• Cleanest long-term solution — a non-blocking connect state machine would benefit all of tinytuya, not just Monitor
• No extra threads needed; reconnect happens inside the reactor's select() loop

Cons:

• Highly invasive — _get_socket() touches socket creation, TCP connect, the v3.4/3.5 session-key handshake, and retry logic. Refactoring it into a state machine is essentially rewriting the connect path
• Risky for backward compatibility — every existing user of synchronous Device depends on the blocking connect working exactly as it does today
• Complex to test — non-blocking connect state machines have subtle edge cases (partial handshakes, timeouts mid-negotiation, etc.)
• Scope creep — this PR is already substantial

My take: The right long-term goal, but not something we should do in this PR. It belongs in the broader asyncio roadmap or as a dedicated follow-up.

---

Option 2: Single worker thread for connections

Pros:

• Minimal complexity — one extra thread with a queue.Queue, processing connect/reconnect requests sequentially
• Reactor stays responsive — disconnects on device A don't block status updates on device B
• Thread count stays bounded at 2 (reactor + connector) regardless of device count
• Fits naturally into the existing architecture: the connector thread does the blocking _get_socket(), then registers the new socket with the selector via the wake mechanism
• Retry/backoff logic lives in one place

Cons:

• Sequential reconnects — if 5 devices drop simultaneously, they reconnect one at a time. In practice this is fine because each connect takes ~1-2 seconds
• One more thread to manage lifecycle for (stop, join on Monitor.stop())

My take: This is the best balance of simplicity, correctness, and minimal invasiveness. The connector thread is small, well-scoped, and doesn't touch any existing code outside Monitor.

---

Option 3: Dedicated thread per device connection

Pros:

• Parallel reconnects — all dropped devices reconnect simultaneously
• Simple per-thread logic

Cons:

• Thread count grows with device count — a Monitor watching 30 devices could briefly have 30 reconnect threads after a network hiccup. This defeats the entire design goal of Monitor (single-thread, multi-device)
• Thread lifecycle management gets complex — tracking, joining, and cleaning up per-device threads on stop() is error-prone
• Resource waste for the common case — most of the time, all devices are connected and these threads don't exist, but a burst of disconnects creates a thread explosion

My take: I'd rule this out. It undermines the core design principle.

---

Option 4 (my suggestion): Callback-driven reconnect with optional connector thread

This is essentially Option 2, but with a user-controlled opt-in:

mon = tinytuya.Monitor(
    on_status=on_status,
    on_disconnect=lambda device, error: print(f'{device.id} disconnected: {error}'),
    auto_reconnect=True,          # enables the connector thread
    reconnect_backoff=5.0,        # seconds between retries
)

When auto_reconnect=True:

1. _handle_disconnect() fires on_disconnect callback AND enqueues the device for reconnect
2. A single connector thread calls device._get_socket(False) with the configured backoff
3. On success, it registers the new socket with the selector and fires on_connect(device, None)
4. On failure, it re-enqueues after the backoff interval

When auto_reconnect=False (default):

• Current behavior: on_disconnect fires, device is removed from the reactor, user handles reconnect manually (or calls mon.add(device) again)

Why this is my recommendation:

• Default behavior is simple and predictable — no hidden threads, no magic
• Power users get automatic reconnect with one flag
• The connector thread only exists when auto_reconnect=True
• Zero changes to XenonDevice or any code outside Monitor.py
• Backward compatible — existing behavior is the default

---

Summary

Approach	Extra threads	Invasiveness	Parallel reconnect	Recommend?
Non-blocking XenonDevice	0	High	Yes	Long-term, not this PR
Single connector thread	1 (opt-in)	Low	Sequential (fine)	✅ Yes
Per-device thread	N devices	Low	Yes	❌ No — defeats design
Callback-only (current)	0	None	N/A	✅ As default fallback

My recommendation: implement Option 4 — the connector thread behind an auto_reconnect flag, with the current callback-only behavior as the default. This keeps the common case simple, gives power users what they need, and doesn't touch any code outside Monitor.py.

Thoughts? @uzlonewolf — I'd also value your perspective here on the socket/ reconnect approach.

— Sam ⚙️