hojichar.utils.io_iter

View Source

 1import io
 2import sys
 3from typing import Iterator, TextIO
 4
 5
 6def stdin_iter() -> Iterator[str]:
 7    """
 8    Encodes standard output in utf-8 and iterates line by line.
 9
10    Wrapping sys.stdin as being opened file, for below reasons:
11    - Specifying encoding.
12    - Processing against invalid unicode inputs.
13
14    Yields:
15        Iterator[str]: Line of stdin stream.
16    """
17    stdin = io.TextIOWrapper(
18        buffer=sys.stdin.buffer,
19        encoding="utf-8",
20        errors="replace",
21    )
22    return (line.rstrip("\n") for line in stdin)
23
24
25def stdout_from_iter(iter: Iterator[str]) -> None:
26    """
27    Write iterators to standard output.
28
29    Wrapping sys.out as being opened file for below reasons:
30    - Specifying encoding.
31    - Enforces line buffering for proper behavior in the program subsequently piped.
32        - Line buffering is necessary for a case:
33            <stdout program> | head -n 1
34        If line buffering is not forced, the program runs with full buffering.
35        The output is not passed to the subsequent pipe until the buffer is full.
36        Thus, the result of the `head` command will not be output immediately.
37
38        - Piped commands sometimes interrupt in the middle of output,
39        as is the case with head, less, etc.
40        These raise a BrokenPipeError exception and must be handled properly.
41    """
42    stdout = io.TextIOWrapper(
43        buffer=sys.stdout.buffer,
44        encoding="utf-8",
45        line_buffering=True,
46    )
47    try:
48        for line in iter:
49            stdout.write(line + "\n")
50    except BrokenPipeError:
51        sys.exit(1)
52
53
54def fileout_from_iter(fp: TextIO, iter: Iterator[str]) -> None:
55    for line in iter:
56        fp.write(line + "\n")

def stdin_iter() -> Iterator[str]: View Source

 7def stdin_iter() -> Iterator[str]:
 8    """
 9    Encodes standard output in utf-8 and iterates line by line.
10
11    Wrapping sys.stdin as being opened file, for below reasons:
12    - Specifying encoding.
13    - Processing against invalid unicode inputs.
14
15    Yields:
16        Iterator[str]: Line of stdin stream.
17    """
18    stdin = io.TextIOWrapper(
19        buffer=sys.stdin.buffer,
20        encoding="utf-8",
21        errors="replace",
22    )
23    return (line.rstrip("\n") for line in stdin)

Encodes standard output in utf-8 and iterates line by line.

Wrapping sys.stdin as being opened file, for below reasons:

Specifying encoding.
Processing against invalid unicode inputs.

Yields: Iterator[str]: Line of stdin stream.

def stdout_from_iter(iter: Iterator[str]) -> None: View Source

26def stdout_from_iter(iter: Iterator[str]) -> None:
27    """
28    Write iterators to standard output.
29
30    Wrapping sys.out as being opened file for below reasons:
31    - Specifying encoding.
32    - Enforces line buffering for proper behavior in the program subsequently piped.
33        - Line buffering is necessary for a case:
34            <stdout program> | head -n 1
35        If line buffering is not forced, the program runs with full buffering.
36        The output is not passed to the subsequent pipe until the buffer is full.
37        Thus, the result of the `head` command will not be output immediately.
38
39        - Piped commands sometimes interrupt in the middle of output,
40        as is the case with head, less, etc.
41        These raise a BrokenPipeError exception and must be handled properly.
42    """
43    stdout = io.TextIOWrapper(
44        buffer=sys.stdout.buffer,
45        encoding="utf-8",
46        line_buffering=True,
47    )
48    try:
49        for line in iter:
50            stdout.write(line + "\n")
51    except BrokenPipeError:
52        sys.exit(1)

Write iterators to standard output.

Wrapping sys.out as being opened file for below reasons:

Specifying encoding.
Enforces line buffering for proper behavior in the program subsequently piped.
- Line buffering is necessary for a case: | head -n 1 If line buffering is not forced, the program runs with full buffering. The output is not passed to the subsequent pipe until the buffer is full. Thus, the result of the head command will not be output immediately.

- Piped commands sometimes interrupt in the middle of output,
as is the case with head, less, etc.
These raise a BrokenPipeError exception and must be handled properly.

def fileout_from_iter(fp: <class 'TextIO'>, iter: Iterator[str]) -> None: View Source

55def fileout_from_iter(fp: TextIO, iter: Iterator[str]) -> None:
56    for line in iter:
57        fp.write(line + "\n")