Update 2.2.3

- Optimising the asynchronous version of Pixels
- Couple of fixes

.gitignore

@@ -3,6 +3,9 @@ __pycache__/
 *.py[cod]
 *$py.class
 
+# My Ignored Type
+*__hidden*
+
 # C extensions
 *.so
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ripix"
-version = "2.2.2"
+version = "2.2.3"
 description = "A Rich-compatible library for writing pixel images and ASCII art to the terminal."
 authors = ["Darren Burns <[email protected]>", "Romanin <[email protected]>"]
 repository = "https://github.com/romanin-rf/ripix"

ripix/__init__.py

@@ -1,3 +1,3 @@
 from .pixel import Pixels, AsyncPixels
 
-__all__ = [ "Pixels" ]
+__all__ = [ "Pixels", "AsyncPixels" ]

ripix/functions.py

@@ -1,20 +1,7 @@
 import asyncio
 
 async def aiter(it):
-    for item in it:
-        yield item
-        await asyncio.sleep(0)
+    for _ in it: await asyncio.sleep(0) ; yield _
 
 async def arange(*args, **kwargs) -> int:
-    for i in range(*args, **kwargs):
-        yield i
-        await asyncio.sleep(0)
-
-async def run_in_executor(loop, executor, func, *args, **kwargs):
-    if loop is None: loop = asyncio.get_running_loop()
-    return await loop.run_in_executor(executor, lambda: func(*args, **kwargs))
-
-def wrapper_run_in_executor(loop, executor, func):
-    async def wrapped_func(*args, **kwargs):
-        return await run_in_executor(loop, executor, func, *args, **kwargs)
-    return wrapped_func
+    for _ in range(*args, **kwargs): await asyncio.sleep(0) ; yield _

ripix/functions.pyi

@@ -1,16 +1,12 @@
-from asyncio import AbstractEventLoop
-from typing import TypeVar, Iterable, AsyncGenerator, overload, SupportsIndex, Callable, Optional, Any, Coroutine
+from typing import TypeVar, Iterable, AsyncGenerator, overload, SupportsIndex
 
+# ! Types
 T = TypeVar("T")
 
-
+# ! Functions
 async def aiter(it: Iterable[T]) -> AsyncGenerator[T]: ...
 
 @overload
 async def arange(__stop: SupportsIndex) -> AsyncGenerator[int]: ...
 @overload
 async def arange(__start: SupportsIndex, __stop: SupportsIndex, __step: SupportsIndex=...) -> AsyncGenerator[int]: ...
-
-async def run_in_executor(loop: Optional[AbstractEventLoop], executor: Optional[Any], func: Callable[..., T], *args, **kwargs) -> T: ...
-
-def wrapper_run_in_executor(loop: Optional[AbstractEventLoop], executor, func: Callable[..., T]) -> Callable[..., Coroutine[Any, Any, T]]: ...

ripix/pixel.py

@@ -2,8 +2,6 @@
 # > Standard Modules
 from pathlib import Path, PurePath
 from typing import Iterable, Mapping, Tuple, Union, Optional, List
-# > Asynchronous
-from asyncio import get_running_loop, AbstractEventLoop
 # > Graphics
 from PIL import Image as PILImageModule
 from PIL.Image import Image
@@ -12,7 +10,7 @@
 from rich.segment import Segment, Segments
 from rich.style import Style
 # > Local Imports
-from .functions import arange, aiter, wrapper_run_in_executor, run_in_executor
+from .functions import arange, aiter
 
 # ! Just Pixels
 class Pixels:
@@ -22,34 +20,40 @@ def __init__(self) -> None:
     @staticmethod
     def from_image(
         image: Image,
-        resize: Optional[Tuple[int, int]] = None
-    ):
-        segments = Pixels._segments_from_image(image, resize)
+        resize: Optional[Tuple[int, int]] = None,
+        resample: Optional[Resampling] = None
+    ) -> Pixels:
+        resample = resample or Resampling.NEAREST
+        segments = Pixels._segments_from_image(image, resize, resample)
         return Pixels.from_segments(segments)
 
     @staticmethod
     def from_image_path(
         path: Union[PurePath, str],
-        resize: Optional[Tuple[int, int]] = None
+        resize: Optional[Tuple[int, int]] = None,
+        resample: Optional[Resampling] = None
     ) -> Pixels:
         """Create a Pixels object from an image. Requires 'image' extra dependencies.
 
         Args:
             path: The path to the image file.
             resize: A tuple of (width, height) to resize the image to.
         """
+        resample = resample or Resampling.NEAREST
+
         with PILImageModule.open(Path(path)) as image:
-            segments = Pixels._segments_from_image(image, resize)
+            segments = Pixels._segments_from_image(image, resize, resample)
 
         return Pixels.from_segments(segments)
 
     @staticmethod
     def _segments_from_image(
         image: Image,
-        resize: Optional[Tuple[int, int]] = None
+        resize: Optional[Tuple[int, int]] = None,
+        resample: Optional[Resampling] = None
     ) -> List[Segment]:
-        if resize:
-            image = image.resize(resize, resample=Resampling.NEAREST)
+        resample = resample or Resampling.NEAREST
+        if resize: image = image.resize(resize, resample=resample)
 
         width, height = image.width, image.height
         rgba_image = image.convert("RGBA")
@@ -69,7 +73,6 @@ def _segments_from_image(
 
             row_append(Segment("\n", null_style))
 
-            # TODO: Double-check if this is required - I've forgotten...
             segments += this_row
 
         return segments
@@ -116,7 +119,6 @@ def __rich_console__(
     ) -> RenderResult:
         yield self._segments or ""
 
-
 # ! Asynchronous Pixels
 class AsyncPixels:
     def __init__(self) -> None:
@@ -126,21 +128,17 @@ def __init__(self) -> None:
     async def from_image(
         image: Image,
         resize: Optional[Tuple[int, int]] = None,
-        resample: Optional[Resampling] = None,
-        loop: Optional[AbstractEventLoop] = None
+        resample: Optional[Resampling] = None
     ) -> AsyncPixels:
-        if loop is None: loop = get_running_loop()
         resample = resample or Resampling.NEAREST
-
-        segments = await AsyncPixels._segments_from_image(image, resize, resample, loop)
+        segments = await AsyncPixels._segments_from_image(image, resize, resample)
         return await AsyncPixels.from_segments(segments)
 
     @staticmethod
     async def from_image_path(
         path: Union[PurePath, str],
         resize: Optional[Tuple[int, int]] = None,
-        resample: Optional[Resampling] = None,
-        loop: Optional[AbstractEventLoop] = None
+        resample: Optional[Resampling] = None
     ) -> AsyncPixels:
         """Create a Pixels object from an image. Requires 'image' extra dependencies.
 
@@ -150,44 +148,34 @@ async def from_image_path(
         """
         resample = resample or Resampling.NEAREST
 
-        if loop is None: loop = get_running_loop()
-        pilopen = wrapper_run_in_executor(loop, None, PILImageModule.open)
-
-        with await pilopen(Path(path)) as image:
-            segments = await AsyncPixels._segments_from_image(image, resize, resample, loop)
+        with PILImageModule.open(Path(path)) as image:
+            segments = await AsyncPixels._segments_from_image(image, resize, resample)
 
         return await AsyncPixels.from_segments(segments)
 
     @staticmethod
     async def _segments_from_image(
         image: Image,
         resize: Optional[Tuple[int, int]] = None,
-        resize_resample: Optional[Resampling] = None,
-        loop: Optional[AbstractEventLoop] = None
+        resize_resample: Optional[Resampling] = None
     ) -> List[Segment]:
         resample = resize_resample or Resampling.NEAREST
-        if loop is None: loop = get_running_loop()
-        if resize is not None: image = await run_in_executor(loop, None, image.resize, resize, resample=resample)
+        if resize is not None: image = image.resize(resize, resample=resample)
 
         width, height = image.width, image.height
-        rgba_image = await run_in_executor(loop, None, image.convert, "RGBA") if image.mode != "RGBA" else image
-        get_pixel = wrapper_run_in_executor(loop, None, rgba_image.getpixel)
-        parse_style = wrapper_run_in_executor(loop, None, Style.parse)
+        rgba_image = image.convert("RGBA")
         null_style = Style.null()
         segments = []
 
         async for y in arange(height):
-            this_row: List[Segment] = []
-            row_append = this_row.append
-
+            this_row = []
             async for x in arange(width):
-                r, g, b, a = await get_pixel((x, y))
-                style = await parse_style(f"on rgb({r},{g},{b})") if (a > 0) else null_style
-                row_append(Segment(" ", style))
-            row_append(Segment("\n", null_style))
-
+                r, g, b, a = rgba_image.getpixel((x, y))
+                style = Style.parse(f"on rgb({r},{g},{b})") if (a > 0) else null_style
+                this_row.append(Segment(" ", style))
+            this_row.append(Segment("\n", null_style))
             segments += this_row
-
+        
         return segments
 
     @staticmethod
@@ -232,6 +220,7 @@ def __rich_console__(
     ) -> RenderResult:
         yield self._segments or ""
 
+
 # * Start
 if __name__ == "__main__":
     console = Console()