You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
bazarr/libs/rich/file_proxy.py

55 lines
1.6 KiB

import io
from typing import List, Any, IO, TYPE_CHECKING
from .ansi import AnsiDecoder
from .text import Text
if TYPE_CHECKING:
from .console import Console
class FileProxy(io.TextIOBase):
"""Wraps a file (e.g. sys.stdout) and redirects writes to a console."""
def __init__(self, console: "Console", file: IO[str]) -> None:
self.__console = console
self.__file = file
self.__buffer: List[str] = []
self.__ansi_decoder = AnsiDecoder()
@property
def rich_proxied_file(self) -> IO[str]:
"""Get proxied file."""
return self.__file
def __getattr__(self, name: str) -> Any:
return getattr(self.__file, name)
def write(self, text: str) -> int:
if not isinstance(text, str):
raise TypeError(f"write() argument must be str, not {type(text).__name__}")
buffer = self.__buffer
lines: List[str] = []
while text:
line, new_line, text = text.partition("\n")
if new_line:
lines.append("".join(buffer) + line)
del buffer[:]
else:
buffer.append(line)
break
if lines:
console = self.__console
with console:
output = Text("\n").join(
self.__ansi_decoder.decode_line(line) for line in lines
)
console.print(output, markup=False, emoji=False, highlight=False)
return len(text)
def flush(self) -> None:
buffer = self.__buffer
if buffer:
self.__console.print("".join(buffer))
del buffer[:]