Update streampair to use RingIO and add async compatibility tests

- Replace deprecated 'ringbuffer' with 'RingIO' name
- Remove unused 'collections.deque' import
- Add test_select_poll_compatibility() to verify select.poll() works
- Add test_streamreader_direct_usage() to verify asyncio.StreamReader compatibility

The existing ioctl implementation already correctly returns -1 for
unsupported operations, which enables select.poll() to work with the
StreamPair Python object. This in turn allows asyncio.StreamReader to
work since it's built on top of select.poll().

These tests demonstrate that pure Python stream objects with proper
ioctl() implementation can work seamlessly with both select.poll() and
asyncio without needing C-level integration.

Author: pi-anl

micropython/streampair/streampair.py

@@ -1,7 +1,6 @@
 import io
 
-from collections import deque
-from micropython import ringbuffer, const
+from micropython import RingIO, const
 
 try:
     from typing import Union, Tuple
@@ -28,14 +27,14 @@ def streampair(buffer_size: Union[int, Tuple[int, int]]=256):
     except TypeError:
         size_a = size_b = buffer_size
 
-    a = ringbuffer(size_a)
-    b = ringbuffer(size_b)
+    a = RingIO(size_a)
+    b = RingIO(size_b)
     return StreamPair(a, b), StreamPair(b, a)
 
 
 class StreamPair(io.IOBase):
 
-    def __init__(self, own: ringbuffer, other: ringbuffer):
+    def __init__(self, own: RingIO, other: RingIO):
         self.own = own
         self.other = other
         super().__init__()

micropython/streampair/test_streampair.py

@@ -1,5 +1,6 @@
 import asyncio
 import unittest
+import select
 from streampair import streampair
 
 def async_test(f):
@@ -49,6 +50,56 @@ async def test_async_streampair(self):
         assert not b.any()
         assert not a.any()
 
+    def test_select_poll_compatibility(self):
+        """Test that streampair works with select.poll()"""
+        a, b = streampair()
+
+        # Register stream with poll
+        poller = select.poll()
+        poller.register(a, select.POLLIN)
+
+        # No data available initially
+        events = poller.poll(0)
+        assert len(events) == 0, f"Expected no events, got {events}"
+
+        # Write data to b, should be readable from a
+        b.write(b"test data")
+
+        # Should now poll as readable
+        events = poller.poll(0)
+        assert len(events) == 1, f"Expected 1 event, got {events}"
+        assert events[0][0] == a, "Event should be for stream a"
+        assert events[0][1] & select.POLLIN, "Should be readable"
+
+        # Read the data
+        data = a.read()
+        assert data == b"test data", f"Expected b'test data', got {data}"
+
+        # Should no longer poll as readable
+        events = poller.poll(0)
+        assert len(events) == 0, f"Expected no events after read, got {events}"
+
+        poller.unregister(a)
+
+    @async_test
+    async def test_streamreader_direct_usage(self):
+        """Test that streampair can be used directly with asyncio.StreamReader"""
+        a, b = streampair()
+
+        # Create StreamReader directly on the streampair object
+        reader = asyncio.StreamReader(a)
+
+        # Write data in background task
+        async def write_delayed():
+            await asyncio.sleep_ms(10)
+            b.write(b"async test\n")
+
+        asyncio.create_task(write_delayed())
+
+        # Should be able to read via StreamReader
+        data = await asyncio.wait_for(reader.readline(), 1.0)
+        assert data == b"async test\n", f"Expected b'async test\\n', got {data}"
+
 
 if __name__ == "__main__":
     unittest.main()