[WIP] aio.util.decolorize: Add util for decolorizing logs

aio.util.decolorize/BUILD

@@ -0,0 +1,2 @@
+
+pytooling_package("aio.api.bazel")

aio.util.decolorize/README.rst

@@ -0,0 +1,5 @@
+
+aio.util.decolorize
+===================
+
+Decolorize terminal output.

aio.util.decolorize/VERSION

@@ -0,0 +1 @@
+0.0.1-dev

aio.util.decolorize/aio/util/decolorize/BUILD

@@ -0,0 +1,10 @@
+
+pytooling_library(
+    "aio.util.decolorize",
+    dependencies=[
+        "//deps:abstracts",
+        "//deps:aio.core",
+        "//deps:aio.run.runner",
+        "//deps:aiohttp",
+    ],
+)

aio.util.decolorize/aio/util/decolorize/__init__.py

@@ -0,0 +1,14 @@
+"""aio.api.bazel."""
+
+from .decolorize import (
+    DecolorizeRunner, )
+from .exceptions import (
+    DecolorizeError, )
+from .decolorize_cmd import decolorize_cmd
+from . import decolorize, exceptions
+
+
+__all__ = (
+    "DecolorizeRunner",
+    "decolorize",
+    "decolorize_cmd")

aio.util.decolorize/aio/util/decolorize/decolorize.py

@@ -0,0 +1,82 @@
+
+import asyncio
+import pathlib
+import sys
+from functools import cached_property
+
+import aiohttp
+
+from aio.core.functional import async_property
+from aio.run import runner
+
+
+class ContentFromPath:
+
+    def __init__(self, path):
+        self._path = path
+
+    @cached_property
+    def path(self):
+        if self.path_type == "fs":
+            return pathlib.Path(self._path)
+        return self._path
+
+    @cached_property
+    def path_type(self):
+        if self._path.startswith("http://"):
+            return "http"
+        if self._path.startswith("https://"):
+            return "https"
+        return "fs"
+
+    def __await__(self):
+        return self.fetch().__await__()
+
+    async def fetch(self):
+        if self.path_type == "fs":
+            return self.path.read_text()
+        async with aiohttp.ClientSession() as session:
+            response = await session.get(self.path)
+            return await response.read()
+
+
+class DecolorizeRunner(runner.Runner):
+
+    @async_property(cache=True)
+    async def input(self):
+        if self.args.input:
+            return await ContentFromPath(self.args.input)
+        reader = await self.reader
+        incoming = await reader.read()
+        return incoming.decode()
+
+    @async_property(cache=True)
+    async def reader(self):
+        await self.loop.connect_read_pipe(
+            lambda: self.stream_protocol,
+            sys.stdin)
+        return self.stream_reader
+
+    @cached_property
+    def stream_protocol(self) -> asyncio.StreamReaderProtocol:
+        return asyncio.StreamReaderProtocol(self.stream_reader)
+
+    @cached_property
+    def stream_reader(self) -> asyncio.StreamReader:
+        return asyncio.StreamReader()
+
+    def add_arguments(self, parser):
+        parser.add_argument("input", nargs="?", default="")
+        parser.add_argument("-o", "--output")
+        super().add_arguments(parser)
+
+    async def decolorize(self, incoming):
+        proc = await asyncio.create_subprocess_exec(
+            "sed", "-r", "s/[[:cntrl:]]\[([0-9]{1,3};)*[0-9]{1,3}m//g",
+            stdin=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE)
+        print((await proc.communicate(incoming))[0].decode())
+
+    async def run(self):
+        await self.decolorize(await self.input)

aio.util.decolorize/aio/util/decolorize/decolorize_cmd.py

@@ -0,0 +1,17 @@
+
+import sys
+from typing import Optional
+
+from .decolorize import DecolorizeRunner
+
+
+def main(*args: str) -> Optional[int]:
+    return DecolorizeRunner(*args)()
+
+
+def decolorize_cmd() -> None:
+    sys.exit(main(*sys.argv[1:]))
+
+
+if __name__ == "__main__":
+    decolorize_cmd()

aio.util.decolorize/aio/util/decolorize/exceptions.py

@@ -0,0 +1,3 @@
+
+class DecolorizeError(Exception):
+    pass

aio.util.decolorize/aio/util/decolorize/interface.py

@@ -0,0 +1,53 @@
+
+import sys
+from typing import Any, Awaitable, Callable, TextIO, Type
+
+import abstracts
+
+from aio.core import pipe
+
+
+class IBazelProcessProtocol(
+        pipe.IProcessProtocol,
+        metaclass=abstracts.Interface):
+
+    # TODO: copy this to aio.core.pipe.interface and fix type
+    @abstracts.interfacemethod
+    async def process(self, request: Any) -> Any:
+        """Process incoming items."""
+        raise NotImplementedError
+
+
+class IBazelWorker(metaclass=abstracts.Interface):
+
+    @property  # type:ignore
+    @abstracts.interfacemethod
+    def processor_class(self) -> Type["IBazelWorkerProcessor"]:
+        raise NotImplementedError
+
+    @property  # type:ignore
+    @abstracts.interfacemethod
+    def protocol_class(self) -> Type["IBazelProcessProtocol"]:
+        raise NotImplementedError
+
+
+class IBazelWorkerProcessor(
+        pipe.IStdinStdoutProcessor,
+        metaclass=abstracts.Interface):
+
+    # TODO: copy this to aio.core.pipe.interface and fix type
+    @abstracts.interfacemethod
+    def __init__(
+            self,
+            protocol: Callable[
+                ["IBazelWorkerProcessor"],
+                Awaitable[IBazelProcessProtocol]],
+            stdin: TextIO = sys.stdin,
+            stdout: TextIO = sys.stdout,
+            log: Callable[[str], None] = None) -> None:
+        raise NotImplementedError
+
+    # TODO: copy this to aio.core.pipe.interface and fix type
+    @abstracts.interfacemethod
+    def __call__(self, *args) -> Any:
+        raise NotImplementedError

aio.util.decolorize/setup.cfg

@@ -0,0 +1,58 @@
+[metadata]
+name = aio.util.decolorize
+version = file: VERSION
+author = Ryan Northey
+author_email = [email protected]
+maintainer = Ryan Northey
+maintainer_email = [email protected]
+license = Apache Software License 2.0
+url = https://bazel.com/envoyproxy/pytooling/tree/main/aio.util.decolorize
+description = Decolorize terminal output
+long_description = file: README.rst
+classifiers =
+    Development Status :: 4 - Beta
+    Framework :: Pytest
+    Intended Audience :: Developers
+    Topic :: Software Development :: Testing
+    Programming Language :: Python
+    Programming Language :: Python :: 3
+    Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.9
+    Programming Language :: Python :: 3 :: Only
+    Programming Language :: Python :: Implementation :: CPython
+    Operating System :: OS Independent
+    License :: OSI Approved :: Apache Software License
+
+[options]
+python_requires = >=3.8
+py_modules = aio.util.decolorize
+packages = find_namespace:
+install_requires =
+    abstracts>=0.0.12
+    aio.core>=0.8.9
+    aio.run.runner>=0.3.3
+
+[options.extras_require]
+test =
+    pytest
+    pytest-asyncio
+    pytest-coverage
+    pytest-patches
+lint = flake8
+types =
+    mypy
+publish = wheel
+
+[options.package_data]
+* = py.typed
+
+[options.packages.find]
+include = aio.*
+exclude =
+    build.*
+    tests.*
+    dist.*
+
+[options.entry_points]
+console_scripts =
+    decolorize = aio.util.decolorize:decolorize_cmd

aio.util.decolorize/setup.py

@@ -0,0 +1,5 @@
+#!/usr/bin/env python
+
+from setuptools import setup  # type:ignore
+
+setup()

aio.util.decolorize/tests/BUILD

@@ -0,0 +1,9 @@
+
+pytooling_tests(
+    "aio.api.bazel",
+    dependencies=[
+        "//deps:abstracts",
+        "//deps:aio.core",
+        "//deps:aio.run.runner",
+    ],
+)