Method: rich.progress.Progress.open
Calls: 2, Exceptions: 0, Paths: 1Back
Path 1: 2 calls (1.0)
'/var/folders/yp/qx0crmvd4sbck7chb52sws500000gn/T/tmp64e9b1we' (1) '/var/folders/yp/qx0crmvd4sbck7chb52sws500000gn/T/tmpaf7shnw1' (1)
'r' (2)
-1 (2)
None (2)
None (2)
None (2)
None (2)
None (2)
'Reading...' (2)
TextIOWrapper (2)
1def open(
2 self,
3 file: Union[str, "PathLike[str]", bytes],
4 mode: Union[Literal["rb"], Literal["rt"], Literal["r"]] = "r",
5 buffering: int = -1,
6 encoding: Optional[str] = None,
7 errors: Optional[str] = None,
8 newline: Optional[str] = None,
9 *,
10 total: Optional[int] = None,
11 task_id: Optional[TaskID] = None,
12 description: str = "Reading...",
13 ) -> Union[BinaryIO, TextIO]:
14 """Track progress while reading from a binary file.
15
16 Args:
17 path (Union[str, PathLike[str]]): The path to the file to read.
18 mode (str): The mode to use to open the file. Only supports "r", "rb" or "rt".
19 buffering (int): The buffering strategy to use, see :func:`io.open`.
20 encoding (str, optional): The encoding to use when reading in text mode, see :func:`io.open`.
21 errors (str, optional): The error handling strategy for decoding errors, see :func:`io.open`.
22 newline (str, optional): The strategy for handling newlines in text mode, see :func:`io.open`.
23 total (int, optional): Total number of bytes to read. If none given, os.stat(path).st_size is used.
24 task_id (TaskID): Task to track. Default is new task.
25 description (str, optional): Description of task, if new task is created.
26
27 Returns:
28 BinaryIO: A readable file-like object in binary mode.
29
30 Raises:
31 ValueError: When an invalid mode is given.
32 """
33 # normalize the mode (always rb, rt)
34 _mode = "".join(sorted(mode, reverse=False))
35 if _mode not in ("br", "rt", "r"):
36 raise ValueError("invalid mode {!r}".format(mode))
37
38 # patch buffering to provide the same behaviour as the builtin `open`
39 line_buffering = buffering == 1
40 if _mode == "br" and buffering == 1:
41 warnings.warn(
42 "line buffering (buffering=1) isn't supported in binary mode, the default buffer size will be used",
43 RuntimeWarning,
44 )
45 buffering = -1
46 elif _mode == "rt" or _mode == "r":
47 if buffering == 0:
48 raise ValueError("can't have unbuffered text I/O")
49 elif buffering == 1:
50 buffering = -1
51
52 # attempt to get the total with `os.stat`
53 if total is None:
54 total = stat(file).st_size
55
56 # update total of task or create new task
57 if task_id is None:
58 task_id = self.add_task(description, total=total)
59 else:
60 self.update(task_id, total=total)
61
62 # open the file in binary mode,
63 handle = io.open(file, "rb", buffering=buffering)
64 reader = _Reader(handle, self, task_id, close_handle=True)
65
66 # wrap the reader in a `TextIOWrapper` if text mode
67 if mode == "r" or mode == "rt":
68 return io.TextIOWrapper(
69 reader,
70 encoding=encoding,
71 errors=errors,
72 newline=newline,
73 line_buffering=line_buffering,
74 )
75
76 return reader