Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3from collections import (
4 abc,
5 defaultdict,
6)
7import csv
8from io import StringIO
9import re
10from typing import (
11 IO,
12 TYPE_CHECKING,
13 Any,
14 DefaultDict,
15 Literal,
16 cast,
17 final,
18)
19import warnings
21import numpy as np
23from pandas._libs import lib
24from pandas._typing import Scalar
25from pandas.errors import (
26 EmptyDataError,
27 ParserError,
28 ParserWarning,
29)
30from pandas.util._decorators import cache_readonly
31from pandas.util._exceptions import find_stack_level
33from pandas.core.dtypes.astype import astype_array
34from pandas.core.dtypes.common import (
35 is_bool_dtype,
36 is_extension_array_dtype,
37 is_integer,
38 is_numeric_dtype,
39 is_object_dtype,
40 is_string_dtype,
41 pandas_dtype,
42)
43from pandas.core.dtypes.dtypes import (
44 CategoricalDtype,
45 ExtensionDtype,
46)
47from pandas.core.dtypes.inference import is_dict_like
49from pandas.core import algorithms
50from pandas.core.arrays import (
51 Categorical,
52 ExtensionArray,
53)
54from pandas.core.arrays.boolean import BooleanDtype
55from pandas.core.indexes.api import Index
57from pandas.io.common import (
58 dedup_names,
59 is_potential_multi_index,
60)
61from pandas.io.parsers.base_parser import (
62 ParserBase,
63 evaluate_callable_usecols,
64 get_na_values,
65 parser_defaults,
66 validate_parse_dates_presence,
67)
69if TYPE_CHECKING:
70 from collections.abc import (
71 Hashable,
72 Iterator,
73 Mapping,
74 Sequence,
75 )
77 from pandas._typing import (
78 ArrayLike,
79 DtypeObj,
80 ReadCsvBuffer,
81 T,
82 )
84 from pandas import (
85 MultiIndex,
86 Series,
87 )
89# BOM character (byte order mark)
90# This exists at the beginning of a file to indicate endianness
91# of a file (stream). Unfortunately, this marker screws up parsing,
92# so we need to remove it if we see it.
93_BOM = "\ufeff"
96class PythonParser(ParserBase):
97 _no_thousands_columns: set[int]
99 def __init__(self, f: ReadCsvBuffer[str] | list, **kwds) -> None:
100 """
101 Workhorse function for processing nested list into DataFrame
102 """
103 super().__init__(kwds)
105 self.data: Iterator[list[str]] | list[list[Scalar]] = []
106 self.buf: list = []
107 self.pos = 0
108 self.line_pos = 0
110 self.skiprows = kwds["skiprows"]
112 if callable(self.skiprows):
113 self.skipfunc = self.skiprows
114 else:
115 self.skipfunc = lambda x: x in self.skiprows
117 self.skipfooter = _validate_skipfooter_arg(kwds["skipfooter"])
118 self.delimiter = kwds["delimiter"]
120 self.quotechar = kwds["quotechar"]
121 if isinstance(self.quotechar, str):
122 self.quotechar = str(self.quotechar)
124 self.escapechar = kwds["escapechar"]
125 self.doublequote = kwds["doublequote"]
126 self.skipinitialspace = kwds["skipinitialspace"]
127 self.lineterminator = kwds["lineterminator"]
128 self.quoting = kwds["quoting"]
129 self.skip_blank_lines = kwds["skip_blank_lines"]
131 # Passed from read_excel
132 self.has_index_names = kwds.get("has_index_names", False)
134 self.thousands = kwds["thousands"]
135 self.decimal = kwds["decimal"]
137 self.comment = kwds["comment"]
139 # Set self.data to something that can read lines.
140 if isinstance(f, list):
141 # read_excel: f is a nested list, can contain non-str
142 self.data = f
143 else:
144 assert hasattr(f, "readline")
145 # yields list of str
146 self.data = self._make_reader(f)
148 # Get columns in two steps: infer from data, then
149 # infer column indices from self.usecols if it is specified.
150 self._col_indices: list[int] | None = None
151 columns: list[list[Scalar | None]]
152 (
153 columns,
154 self.num_original_columns,
155 self.unnamed_cols,
156 ) = self._infer_columns()
158 # Now self.columns has the set of columns that we will process.
159 # The original set is stored in self.original_columns.
160 # error: Cannot determine type of 'index_names'
161 (
162 self.columns,
163 self.index_names,
164 self.col_names,
165 _,
166 ) = self._extract_multi_indexer_columns(
167 columns,
168 self.index_names,
169 )
171 # get popped off for index
172 self.orig_names: list[Hashable] = list(self.columns)
174 index_names, self.orig_names, self.columns = self._get_index_name()
175 if self.index_names is None:
176 self.index_names = index_names
178 if self._col_indices is None:
179 self._col_indices = list(range(len(self.columns)))
181 self._no_thousands_columns = self._set_no_thousand_columns()
183 if len(self.decimal) != 1:
184 raise ValueError("Only length-1 decimal markers supported")
186 @cache_readonly
187 def num(self) -> re.Pattern:
188 decimal = re.escape(self.decimal)
189 if self.thousands is None:
190 regex = rf"^[\-\+]?[0-9]*({decimal}[0-9]*)?([0-9]?(E|e)\-?[0-9]+)?$"
191 else:
192 thousands = re.escape(self.thousands)
193 regex = (
194 rf"^[\-\+]?([0-9]+{thousands}|[0-9])*({decimal}[0-9]*)?"
195 rf"([0-9]?(E|e)\-?[0-9]+)?$"
196 )
197 return re.compile(regex)
199 def _make_reader(self, f: IO[str] | ReadCsvBuffer[str]) -> Iterator[list[str]]:
200 sep = self.delimiter
202 if sep is None or len(sep) == 1:
203 if self.lineterminator:
204 raise ValueError(
205 "Custom line terminators not supported in python parser (yet)"
206 )
208 class MyDialect(csv.Dialect):
209 delimiter = self.delimiter
210 quotechar = self.quotechar
211 escapechar = self.escapechar
212 doublequote = self.doublequote
213 skipinitialspace = self.skipinitialspace
214 quoting = self.quoting
215 lineterminator = "\n"
217 dia = MyDialect
219 if sep is not None:
220 dia.delimiter = sep
221 # Skip rows at file level before csv.reader sees them
222 # prevents CSV parsing errors on lines that will be discarded
223 if self.skiprows is not None:
224 while self.skipfunc(self.pos):
225 line = f.readline()
226 if not line:
227 break
228 self.pos += 1
229 else:
230 # attempt to sniff the delimiter from the first valid line,
231 # i.e. no comment line and not in skiprows
232 line = f.readline()
233 lines = self._check_comments([[line]])[0]
234 while self.skipfunc(self.pos) or not lines:
235 self.pos += 1
236 line = f.readline()
237 lines = self._check_comments([[line]])[0]
238 lines_str = cast(list[str], lines)
240 # since `line` was a string, lines will be a list containing
241 # only a single string
242 line = lines_str[0]
244 self.pos += 1
245 self.line_pos += 1
246 sniffed = csv.Sniffer().sniff(line)
247 dia.delimiter = sniffed.delimiter
249 # Note: encoding is irrelevant here
250 line_rdr = csv.reader(StringIO(line), dialect=dia)
251 self.buf.extend(list(line_rdr))
253 # Note: encoding is irrelevant here
254 reader = csv.reader(f, dialect=dia, strict=True)
256 else:
258 def _read():
259 line = f.readline()
260 pat = re.compile(sep)
262 yield pat.split(line.strip())
264 for line in f:
265 yield pat.split(line.strip())
267 reader = _read()
269 return reader
271 def read(
272 self, rows: int | None = None
273 ) -> tuple[
274 Index | None,
275 Sequence[Hashable] | MultiIndex,
276 Mapping[Hashable, ArrayLike | Series],
277 ]:
278 try:
279 content = self._get_lines(rows)
280 except StopIteration:
281 if self._first_chunk:
282 content = []
283 else:
284 self.close()
285 raise
287 # done with first read, next time raise StopIteration
288 self._first_chunk = False
290 index: Index | None
291 columns: Sequence[Hashable] = list(self.orig_names)
292 if not content: # pragma: no cover
293 # DataFrame with the right metadata, even though it's length 0
294 # error: Cannot determine type of 'index_col'
295 names = dedup_names(
296 self.orig_names,
297 is_potential_multi_index(
298 self.orig_names,
299 self.index_col,
300 ),
301 )
302 index, columns, col_dict = self._get_empty_meta(
303 names,
304 self.dtype,
305 )
306 conv_columns = self._maybe_make_multi_index_columns(columns, self.col_names)
307 return index, conv_columns, col_dict
309 # handle new style for names in index
310 indexnamerow = None
311 if self.has_index_names and sum(
312 int(v == "" or v is None) for v in content[0]
313 ) == len(columns):
314 indexnamerow = content[0]
315 content = content[1:]
317 alldata = self._rows_to_cols(content)
318 data, columns = self._exclude_implicit_index(alldata)
320 conv_data = self._convert_data(data)
321 conv_data = self._do_date_conversions(columns, conv_data)
323 index, result_columns = self._make_index(alldata, columns, indexnamerow)
325 return index, result_columns, conv_data
327 def _exclude_implicit_index(
328 self,
329 alldata: list[np.ndarray],
330 ) -> tuple[Mapping[Hashable, np.ndarray], Sequence[Hashable]]:
331 # error: Cannot determine type of 'index_col'
332 names = dedup_names(
333 self.orig_names,
334 is_potential_multi_index(
335 self.orig_names,
336 self.index_col,
337 ),
338 )
340 offset = 0
341 if self._implicit_index:
342 offset = len(self.index_col)
344 len_alldata = len(alldata)
345 self._check_data_length(names, alldata)
347 return {
348 name: alldata[i + offset] for i, name in enumerate(names) if i < len_alldata
349 }, names
351 # legacy
352 def get_chunk(
353 self, size: int | None = None
354 ) -> tuple[
355 Index | None,
356 Sequence[Hashable] | MultiIndex,
357 Mapping[Hashable, ArrayLike | Series],
358 ]:
359 if size is None:
360 # error: "PythonParser" has no attribute "chunksize"
361 size = self.chunksize # type: ignore[attr-defined]
362 return self.read(rows=size)
364 def _convert_data(
365 self,
366 data: Mapping[Hashable, np.ndarray],
367 ) -> Mapping[Hashable, ArrayLike]:
368 # apply converters
369 clean_conv = self._clean_mapping(self.converters)
370 clean_dtypes = self._clean_mapping(self.dtype)
372 # Apply NA values.
373 clean_na_values = {}
374 clean_na_fvalues = {}
376 if isinstance(self.na_values, dict):
377 for col in self.na_values:
378 if col is not None:
379 na_value = self.na_values[col]
380 na_fvalue = self.na_fvalues[col]
382 if isinstance(col, int) and col not in self.orig_names:
383 col = self.orig_names[col]
385 clean_na_values[col] = na_value
386 clean_na_fvalues[col] = na_fvalue
387 else:
388 clean_na_values = self.na_values
389 clean_na_fvalues = self.na_fvalues
391 return self._convert_to_ndarrays(
392 data,
393 clean_na_values,
394 clean_na_fvalues,
395 clean_conv,
396 clean_dtypes,
397 )
399 @final
400 def _convert_to_ndarrays(
401 self,
402 dct: Mapping,
403 na_values,
404 na_fvalues,
405 converters=None,
406 dtypes=None,
407 ) -> dict[Any, np.ndarray]:
408 result = {}
409 parse_date_cols = validate_parse_dates_presence(self.parse_dates, self.columns)
410 for c, values in dct.items():
411 conv_f = None if converters is None else converters.get(c, None)
412 if isinstance(dtypes, dict):
413 cast_type = dtypes.get(c, None)
414 else:
415 # single dtype or None
416 cast_type = dtypes
418 if self.na_filter:
419 col_na_values, col_na_fvalues = get_na_values(
420 c, na_values, na_fvalues, self.keep_default_na
421 )
422 else:
423 col_na_values, col_na_fvalues = set(), set()
425 if c in parse_date_cols:
426 # GH#26203 Do not convert columns which get converted to dates
427 # but replace nans to ensure to_datetime works
428 mask = algorithms.isin(values, set(col_na_values) | col_na_fvalues) # pyright: ignore[reportArgumentType]
429 np.putmask(values, mask, np.nan)
430 result[c] = values
431 continue
433 if conv_f is not None:
434 # conv_f applied to data before inference
435 if cast_type is not None:
436 warnings.warn(
437 (
438 "Both a converter and dtype were specified "
439 f"for column {c} - only the converter will be used."
440 ),
441 ParserWarning,
442 stacklevel=find_stack_level(),
443 )
445 try:
446 values = lib.map_infer(values, conv_f)
447 except ValueError:
448 mask = algorithms.isin(values, list(na_values)).view(np.uint8)
449 values = lib.map_infer_mask(values, conv_f, mask)
451 cvals, na_count = self._infer_types(
452 values,
453 set(col_na_values) | col_na_fvalues,
454 cast_type is None,
455 try_num_bool=False,
456 )
457 else:
458 is_ea = is_extension_array_dtype(cast_type)
459 is_str_or_ea_dtype = is_ea or is_string_dtype(cast_type)
460 # skip inference if specified dtype is object
461 # or casting to an EA
462 try_num_bool = not (cast_type and is_str_or_ea_dtype)
464 # general type inference and conversion
465 cvals, na_count = self._infer_types(
466 values,
467 set(col_na_values) | col_na_fvalues,
468 cast_type is None,
469 try_num_bool,
470 )
472 # type specified in dtype param or cast_type is an EA
473 if cast_type is not None:
474 cast_type = pandas_dtype(cast_type)
475 if cast_type and (cvals.dtype != cast_type or is_ea):
476 if not is_ea and na_count > 0:
477 if is_bool_dtype(cast_type):
478 raise ValueError(f"Bool column has NA values in column {c}")
479 cvals = self._cast_types(cvals, cast_type, c)
481 result[c] = cvals
482 return result
484 @final
485 def _cast_types(self, values: ArrayLike, cast_type: DtypeObj, column) -> ArrayLike:
486 """
487 Cast values to specified type
489 Parameters
490 ----------
491 values : ndarray or ExtensionArray
492 cast_type : np.dtype or ExtensionDtype
493 dtype to cast values to
494 column : string
495 column name - used only for error reporting
497 Returns
498 -------
499 converted : ndarray or ExtensionArray
500 """
501 if isinstance(cast_type, CategoricalDtype):
502 known_cats = cast_type.categories is not None
504 if not is_object_dtype(values.dtype) and not known_cats:
505 # TODO: this is for consistency with
506 # c-parser which parses all categories
507 # as strings
508 values = lib.ensure_string_array(
509 values, skipna=False, convert_na_value=False
510 )
512 cats = Index(values, copy=False).unique().dropna()
513 values = Categorical._from_inferred_categories(
514 cats, cats.get_indexer(values), cast_type, true_values=self.true_values
515 )
517 # use the EA's implementation of casting
518 elif isinstance(cast_type, ExtensionDtype):
519 array_type = cast_type.construct_array_type()
520 try:
521 if isinstance(cast_type, BooleanDtype):
522 # error: Unexpected keyword argument "true_values" for
523 # "_from_sequence_of_strings" of "ExtensionArray"
524 values_str = [str(val) for val in values]
525 return array_type._from_sequence_of_strings( # type: ignore[call-arg]
526 values_str,
527 dtype=cast_type,
528 true_values=self.true_values, # pyright: ignore[reportCallIssue]
529 false_values=self.false_values, # pyright: ignore[reportCallIssue]
530 none_values=self.na_values, # pyright: ignore[reportCallIssue]
531 )
532 else:
533 return array_type._from_sequence_of_strings(values, dtype=cast_type)
534 except NotImplementedError as err:
535 raise NotImplementedError(
536 f"Extension Array: {array_type} must implement "
537 "_from_sequence_of_strings in order to be used in parser methods"
538 ) from err
540 elif isinstance(values, ExtensionArray):
541 values = values.astype(cast_type, copy=False)
542 elif issubclass(cast_type.type, str):
543 # TODO: why skipna=True here and False above? some tests depend
544 # on it here, but nothing fails if we change it above
545 # (as no tests get there as of 2022-12-06)
546 values = lib.ensure_string_array(
547 values, skipna=True, convert_na_value=False
548 )
549 else:
550 try:
551 values = astype_array(values, cast_type, copy=True)
552 except ValueError as err:
553 raise ValueError(
554 f"Unable to convert column {column} to type {cast_type}"
555 ) from err
556 return values
558 @cache_readonly
559 def _have_mi_columns(self) -> bool:
560 if self.header is None:
561 return False
563 header = self.header
564 if isinstance(header, (list, tuple, np.ndarray)):
565 return len(header) > 1
566 else:
567 return False
569 def _infer_columns(
570 self,
571 ) -> tuple[list[list[Scalar | None]], int, set[Scalar | None]]:
572 names = self.names
573 num_original_columns = 0
574 clear_buffer = True
575 unnamed_cols: set[Scalar | None] = set()
577 if self.header is not None:
578 header = self.header
579 have_mi_columns = self._have_mi_columns
581 if isinstance(header, (list, tuple, np.ndarray)):
582 # we have a mi columns, so read an extra line
583 if have_mi_columns:
584 header = [*list(header), header[-1] + 1]
585 else:
586 header = [header]
588 columns: list[list[Scalar | None]] = []
589 for level, hr in enumerate(header):
590 try:
591 line = self._buffered_line()
593 while self.line_pos <= hr:
594 line = self._next_line()
596 except StopIteration as err:
597 if 0 < self.line_pos <= hr and (
598 not have_mi_columns or hr != header[-1]
599 ):
600 # If no rows we want to raise a different message and if
601 # we have mi columns, the last line is not part of the header
602 joi = list(map(str, header[:-1] if have_mi_columns else header))
603 msg = f"[{','.join(joi)}], len of {len(joi)}, "
604 raise ValueError(
605 f"Passed header={msg}but only {self.line_pos} lines in file"
606 ) from err
608 # We have an empty file, so check
609 # if columns are provided. That will
610 # serve as the 'line' for parsing
611 if have_mi_columns and hr > 0:
612 if clear_buffer:
613 self.buf.clear()
614 columns.append([None] * len(columns[-1]))
615 return columns, num_original_columns, unnamed_cols
617 if not self.names:
618 raise EmptyDataError("No columns to parse from file") from err
620 line = self.names[:]
622 this_columns: list[Scalar | None] = []
623 this_unnamed_cols = []
625 for i, c in enumerate(line):
626 if c == "":
627 if have_mi_columns:
628 col_name = f"Unnamed: {i}_level_{level}"
629 else:
630 col_name = f"Unnamed: {i}"
632 this_unnamed_cols.append(i)
633 this_columns.append(col_name)
634 else:
635 this_columns.append(c)
637 if not have_mi_columns:
638 counts: DefaultDict = defaultdict(int)
639 # Ensure that regular columns are used before unnamed ones
640 # to keep given names and mangle unnamed columns
641 col_loop_order = [
642 i
643 for i in range(len(this_columns))
644 if i not in this_unnamed_cols
645 ] + this_unnamed_cols
647 # TODO: Use pandas.io.common.dedup_names instead (see #50371)
648 for i in col_loop_order:
649 col = this_columns[i]
650 old_col = col
651 cur_count = counts[col]
653 if cur_count > 0:
654 while cur_count > 0:
655 counts[old_col] = cur_count + 1
656 col = f"{old_col}.{cur_count}"
657 if col in this_columns:
658 cur_count += 1
659 else:
660 cur_count = counts[col]
662 if (
663 self.dtype is not None
664 and is_dict_like(self.dtype)
665 and self.dtype.get(old_col) is not None
666 and self.dtype.get(col) is None
667 ):
668 self.dtype.update({col: self.dtype.get(old_col)})
669 this_columns[i] = col
670 counts[col] = cur_count + 1
671 elif have_mi_columns:
672 # if we have grabbed an extra line, but it's not in our
673 # format so save in the buffer, and create a blank extra
674 # line for the rest of the parsing code
675 if hr == header[-1]:
676 lc = len(this_columns)
677 sic = self.index_col
678 ic = len(sic) if sic is not None else 0
679 unnamed_count = len(this_unnamed_cols)
681 # if wrong number of blanks or no index, not our format
682 if (lc != unnamed_count and lc - ic > unnamed_count) or ic == 0:
683 clear_buffer = False
684 this_columns = [None] * lc
685 self.buf = [self.buf[-1]]
687 columns.append(this_columns)
688 unnamed_cols.update({this_columns[i] for i in this_unnamed_cols})
690 if len(columns) == 1:
691 num_original_columns = len(this_columns)
693 if clear_buffer:
694 self.buf.clear()
696 first_line: list[Scalar] | None
697 if names is not None:
698 # Read first row after header to check if data are longer
699 try:
700 first_line = self._next_line()
701 except StopIteration:
702 first_line = None
704 len_first_data_row = 0 if first_line is None else len(first_line)
706 if len(names) > len(columns[0]) and len(names) > len_first_data_row:
707 raise ValueError(
708 "Number of passed names did not match "
709 "number of header fields in the file"
710 )
711 if len(columns) > 1:
712 raise TypeError("Cannot pass names with multi-index columns")
714 if self.usecols is not None:
715 # Set _use_cols. We don't store columns because they are
716 # overwritten.
717 self._handle_usecols(columns, names, num_original_columns)
718 else:
719 num_original_columns = len(names)
720 if self._col_indices is not None and len(names) != len(
721 self._col_indices
722 ):
723 columns = [[names[i] for i in sorted(self._col_indices)]]
724 else:
725 columns = [names]
726 else:
727 columns = self._handle_usecols(
728 columns, columns[0], num_original_columns
729 )
730 else:
731 ncols = len(self._header_line)
732 num_original_columns = ncols
734 if not names:
735 columns = [list(range(ncols))]
736 columns = self._handle_usecols(columns, columns[0], ncols)
737 elif self.usecols is None or len(names) >= ncols:
738 columns = self._handle_usecols([names], names, ncols)
739 num_original_columns = len(names)
740 elif not callable(self.usecols) and len(names) != len(self.usecols):
741 raise ValueError(
742 "Number of passed names did not match number of "
743 "header fields in the file"
744 )
745 else:
746 # Ignore output but set used columns.
747 columns = [names]
748 self._handle_usecols(columns, columns[0], ncols)
750 return columns, num_original_columns, unnamed_cols
752 @cache_readonly
753 def _header_line(self):
754 # Store line for reuse in _get_index_name
755 if self.header is not None:
756 return None
758 try:
759 line = self._buffered_line()
760 except StopIteration as err:
761 if not self.names:
762 raise EmptyDataError("No columns to parse from file") from err
764 line = self.names[:]
765 return line
767 def _handle_usecols(
768 self,
769 columns: list[list[Scalar | None]],
770 usecols_key: list[Scalar | None],
771 num_original_columns: int,
772 ) -> list[list[Scalar | None]]:
773 """
774 Sets self._col_indices
776 usecols_key is used if there are string usecols.
777 """
778 col_indices: set[int] | list[int]
779 if self.usecols is not None:
780 if callable(self.usecols):
781 col_indices = evaluate_callable_usecols(self.usecols, usecols_key)
782 elif any(isinstance(u, str) for u in self.usecols):
783 if len(columns) > 1:
784 raise ValueError(
785 "If using multiple headers, usecols must be integers."
786 )
787 col_indices = []
789 for col in self.usecols:
790 if isinstance(col, str):
791 try:
792 col_indices.append(usecols_key.index(col))
793 except ValueError:
794 self._validate_usecols_names(self.usecols, usecols_key)
795 else:
796 col_indices.append(col)
797 else:
798 missing_usecols = [
799 col for col in self.usecols if col >= num_original_columns
800 ]
801 if missing_usecols:
802 raise ParserError(
803 "Defining usecols with out-of-bounds indices is not allowed. "
804 f"{missing_usecols} are out-of-bounds.",
805 )
806 col_indices = self.usecols
808 columns = [
809 [n for i, n in enumerate(column) if i in col_indices]
810 for column in columns
811 ]
812 self._col_indices = sorted(col_indices)
813 return columns
815 def _buffered_line(self) -> list[Scalar]:
816 """
817 Return a line from buffer, filling buffer if required.
818 """
819 if len(self.buf) > 0:
820 return self.buf[0]
821 else:
822 return self._next_line()
824 def _check_for_bom(self, first_row: list[Scalar]) -> list[Scalar]:
825 """
826 Checks whether the file begins with the BOM character.
827 If it does, remove it. In addition, if there is quoting
828 in the field subsequent to the BOM, remove it as well
829 because it technically takes place at the beginning of
830 the name, not the middle of it.
831 """
832 # first_row will be a list, so we need to check
833 # that that list is not empty before proceeding.
834 if not first_row:
835 return first_row
837 # The first element of this row is the one that could have the
838 # BOM that we want to remove. Check that the first element is a
839 # string before proceeding.
840 if not isinstance(first_row[0], str):
841 return first_row
843 # Check that the string is not empty, as that would
844 # obviously not have a BOM at the start of it.
845 if not first_row[0]:
846 return first_row
848 # Since the string is non-empty, check that it does
849 # in fact begin with a BOM.
850 first_elt = first_row[0][0]
851 if first_elt != _BOM:
852 return first_row
854 first_row_bom = first_row[0]
855 new_row: str
857 if len(first_row_bom) > 1 and first_row_bom[1] == self.quotechar:
858 start = 2
859 quote = first_row_bom[1]
860 end = first_row_bom[2:].index(quote) + 2
862 # Extract the data between the quotation marks
863 new_row = first_row_bom[start:end]
865 # Extract any remaining data after the second
866 # quotation mark.
867 if len(first_row_bom) > end + 1:
868 new_row += first_row_bom[end + 1 :]
870 else:
871 # No quotation so just remove BOM from first element
872 new_row = first_row_bom[1:]
874 new_row_list: list[Scalar] = [new_row]
875 return new_row_list + first_row[1:]
877 def _is_line_empty(self, line: Sequence[Scalar]) -> bool:
878 """
879 Check if a line is empty or not.
881 Parameters
882 ----------
883 line : str, array-like
884 The line of data to check.
886 Returns
887 -------
888 boolean : Whether or not the line is empty.
889 """
890 return not line or all(not x for x in line)
892 def _next_line(self) -> list[Scalar]:
893 if isinstance(self.data, list):
894 while self.skipfunc(self.pos):
895 if self.pos >= len(self.data):
896 break
897 self.pos += 1
899 while True:
900 try:
901 line = self._check_comments([self.data[self.pos]])[0]
902 self.pos += 1
903 # either uncommented or blank to begin with
904 if not self.skip_blank_lines and (
905 self._is_line_empty(self.data[self.pos - 1]) or line
906 ):
907 break
908 if self.skip_blank_lines:
909 ret = self._remove_empty_lines([line])
910 if ret:
911 line = ret[0]
912 break
913 except IndexError as err:
914 raise StopIteration from err
915 else:
916 while self.skipfunc(self.pos):
917 self.pos += 1
918 next(self.data)
920 while True:
921 orig_line = self._next_iter_line(row_num=self.pos + 1)
922 self.pos += 1
924 if orig_line is not None:
925 line = self._check_comments([orig_line])[0]
927 if self.skip_blank_lines:
928 ret = self._remove_empty_lines([line])
930 if ret:
931 line = ret[0]
932 break
933 elif self._is_line_empty(orig_line) or line:
934 break
936 # This was the first line of the file,
937 # which could contain the BOM at the
938 # beginning of it.
939 if self.pos == 1:
940 line = self._check_for_bom(line)
942 self.line_pos += 1
943 self.buf.append(line)
944 return line
946 def _alert_malformed(self, msg: str, row_num: int) -> None:
947 """
948 Alert a user about a malformed row, depending on value of
949 `self.on_bad_lines` enum.
951 If `self.on_bad_lines` is ERROR, the alert will be `ParserError`.
952 If `self.on_bad_lines` is WARN, the alert will be printed out.
954 Parameters
955 ----------
956 msg: str
957 The error message to display.
958 row_num: int
959 The row number where the parsing error occurred.
960 Because this row number is displayed, we 1-index,
961 even though we 0-index internally.
962 """
963 if self.on_bad_lines == self.BadLineHandleMethod.ERROR:
964 raise ParserError(msg)
965 if self.on_bad_lines == self.BadLineHandleMethod.WARN or callable(
966 self.on_bad_lines
967 ):
968 warnings.warn(
969 f"Skipping line {row_num}: {msg}\n",
970 ParserWarning,
971 stacklevel=find_stack_level(),
972 )
974 def _next_iter_line(self, row_num: int) -> list[Scalar] | None:
975 """
976 Wrapper around iterating through `self.data` (CSV source).
978 When a CSV error is raised, we check for specific
979 error messages that allow us to customize the
980 error message displayed to the user.
982 Parameters
983 ----------
984 row_num: int
985 The row number of the line being parsed.
986 """
987 try:
988 assert not isinstance(self.data, list)
989 line = next(self.data)
990 # lie about list[str] vs list[Scalar] to minimize ignores
991 return line # type: ignore[return-value]
992 except csv.Error as e:
993 if self.on_bad_lines in (
994 self.BadLineHandleMethod.ERROR,
995 self.BadLineHandleMethod.WARN,
996 ):
997 msg = str(e)
999 if "NULL byte" in msg or "line contains NUL" in msg:
1000 msg = (
1001 "NULL byte detected. This byte "
1002 "cannot be processed in Python's "
1003 "native csv library at the moment, "
1004 "so please pass in engine='c' instead"
1005 )
1007 if self.skipfooter > 0:
1008 reason = (
1009 "Error could possibly be due to "
1010 "parsing errors in the skipped footer rows "
1011 "(the skipfooter keyword is only applied "
1012 "after Python's csv library has parsed "
1013 "all rows)."
1014 )
1015 msg += ". " + reason
1017 self._alert_malformed(msg, row_num)
1018 return None
1020 def _check_comments(self, lines: list[list[Scalar]]) -> list[list[Scalar]]:
1021 if self.comment is None:
1022 return lines
1023 ret = []
1024 for line in lines:
1025 rl = []
1026 for x in line:
1027 if (
1028 not isinstance(x, str)
1029 or self.comment not in x
1030 or x in self.na_values
1031 ):
1032 rl.append(x)
1033 else:
1034 x = x[: x.find(self.comment)]
1035 if len(x) > 0:
1036 rl.append(x)
1037 break
1038 ret.append(rl)
1039 return ret
1041 def _remove_empty_lines(self, lines: list[list[T]]) -> list[list[T]]:
1042 """
1043 Iterate through the lines and remove any that are
1044 either empty or contain only one whitespace value
1046 Parameters
1047 ----------
1048 lines : list of list of Scalars
1049 The array of lines that we are to filter.
1051 Returns
1052 -------
1053 filtered_lines : list of list of Scalars
1054 The same array of lines with the "empty" ones removed.
1055 """
1056 # Remove empty lines and lines with only one whitespace value
1057 ret = [
1058 line
1059 for line in lines
1060 if (
1061 len(line) > 1
1062 or (
1063 len(line) == 1 and (not isinstance(line[0], str) or line[0].strip())
1064 )
1065 )
1066 ]
1067 return ret
1069 def _check_thousands(self, lines: list[list[Scalar]]) -> list[list[Scalar]]:
1070 if self.thousands is None:
1071 return lines
1073 return self._search_replace_num_columns(
1074 lines=lines, search=self.thousands, replace=""
1075 )
1077 def _search_replace_num_columns(
1078 self, lines: list[list[Scalar]], search: str, replace: str
1079 ) -> list[list[Scalar]]:
1080 ret = []
1081 for line in lines:
1082 rl = []
1083 for i, x in enumerate(line):
1084 if (
1085 not isinstance(x, str)
1086 or search not in x
1087 or i in self._no_thousands_columns
1088 or not self.num.search(x.strip())
1089 ):
1090 rl.append(x)
1091 else:
1092 rl.append(x.replace(search, replace))
1093 ret.append(rl)
1094 return ret
1096 def _check_decimal(self, lines: list[list[Scalar]]) -> list[list[Scalar]]:
1097 if self.decimal == parser_defaults["decimal"]:
1098 return lines
1100 return self._search_replace_num_columns(
1101 lines=lines, search=self.decimal, replace="."
1102 )
1104 def _get_index_name(
1105 self,
1106 ) -> tuple[Sequence[Hashable] | None, list[Hashable], list[Hashable]]:
1107 """
1108 Try several cases to get lines:
1110 0) There are headers on row 0 and row 1 and their
1111 total summed lengths equals the length of the next line.
1112 Treat row 0 as columns and row 1 as indices
1113 1) Look for implicit index: there are more columns
1114 on row 1 than row 0. If this is true, assume that row
1115 1 lists index columns and row 0 lists normal columns.
1116 2) Get index from the columns if it was listed.
1117 """
1118 columns: Sequence[Hashable] = self.orig_names
1119 orig_names = list(columns)
1120 columns = list(columns)
1122 line: list[Scalar] | None
1123 if self._header_line is not None:
1124 line = self._header_line
1125 else:
1126 try:
1127 line = self._next_line()
1128 except StopIteration:
1129 line = None
1131 next_line: list[Scalar] | None
1132 try:
1133 next_line = self._next_line()
1134 except StopIteration:
1135 next_line = None
1137 # implicitly index_col=0 b/c 1 fewer column names
1138 implicit_first_cols = 0
1139 if line is not None:
1140 # leave it 0, #2442
1141 # Case 1
1142 index_col = self.index_col
1143 if index_col is not False:
1144 implicit_first_cols = len(line) - self.num_original_columns
1146 # Case 0
1147 if (
1148 next_line is not None
1149 and self.header is not None
1150 and index_col is not False
1151 ):
1152 if len(next_line) == len(line) + self.num_original_columns:
1153 # column and index names on diff rows
1154 self.index_col = list(range(len(line)))
1155 self.buf = self.buf[1:]
1157 for c in reversed(line):
1158 columns.insert(0, c)
1160 # Update list of original names to include all indices.
1161 orig_names = list(columns)
1162 self.num_original_columns = len(columns)
1163 return line, orig_names, columns
1165 if implicit_first_cols > 0:
1166 # Case 1
1167 self._implicit_index = True
1168 if self.index_col is None:
1169 self.index_col = list(range(implicit_first_cols))
1171 index_name = None
1173 else:
1174 # Case 2
1175 (index_name, _, self.index_col) = self._clean_index_names(
1176 columns, self.index_col
1177 )
1179 return index_name, orig_names, columns
1181 def _rows_to_cols(self, content: list[list[Scalar]]) -> list[np.ndarray]:
1182 col_len = self.num_original_columns
1184 if self._implicit_index:
1185 col_len += len(self.index_col)
1187 max_len = max(len(row) for row in content)
1189 # Check that there are no rows with too many
1190 # elements in their row (rows with too few
1191 # elements are padded with NaN).
1192 if max_len > col_len and self.index_col is not False and self.usecols is None:
1193 footers = self.skipfooter if self.skipfooter else 0
1194 bad_lines = []
1196 iter_content = enumerate(content)
1197 content_len = len(content)
1198 content = []
1200 for i, _content in iter_content:
1201 actual_len = len(_content)
1202 if actual_len > col_len:
1203 if callable(self.on_bad_lines):
1204 new_l = self.on_bad_lines(_content)
1205 if new_l is not None:
1206 new_l = cast(list[Scalar], new_l)
1207 if len(new_l) > col_len:
1208 row_num = self.pos - (content_len - i + footers)
1209 bad_lines.append((row_num, len(new_l), "callable"))
1210 new_l = new_l[:col_len]
1211 content.append(new_l)
1213 elif self.on_bad_lines in (
1214 self.BadLineHandleMethod.ERROR,
1215 self.BadLineHandleMethod.WARN,
1216 ):
1217 row_num = self.pos - (content_len - i + footers)
1218 bad_lines.append((row_num, actual_len, "normal"))
1219 if self.on_bad_lines == self.BadLineHandleMethod.ERROR:
1220 break
1221 else:
1222 content.append(_content)
1224 for row_num, actual_len, source in bad_lines:
1225 msg = (
1226 f"Expected {col_len} fields in line {row_num + 1}, saw {actual_len}"
1227 )
1228 if source == "callable":
1229 msg += " from bad_lines callable"
1230 elif (
1231 self.delimiter
1232 and len(self.delimiter) > 1
1233 and self.quoting != csv.QUOTE_NONE
1234 ):
1235 # see gh-13374
1236 reason = (
1237 "Error could possibly be due to quotes being "
1238 "ignored when a multi-char delimiter is used."
1239 )
1240 msg += ". " + reason
1242 self._alert_malformed(msg, row_num + 1)
1244 # see gh-13320
1245 zipped_content = list(lib.to_object_array(content, min_width=col_len).T)
1247 if self.usecols:
1248 assert self._col_indices is not None
1249 col_indices = self._col_indices
1251 if self._implicit_index:
1252 zipped_content = [
1253 a
1254 for i, a in enumerate(zipped_content)
1255 if (
1256 i < len(self.index_col)
1257 or i - len(self.index_col) in col_indices
1258 )
1259 ]
1260 else:
1261 zipped_content = [
1262 a for i, a in enumerate(zipped_content) if i in col_indices
1263 ]
1264 return zipped_content
1266 def _get_lines(self, rows: int | None = None) -> list[list[Scalar]]:
1267 lines = self.buf
1268 new_rows = None
1270 # already fetched some number
1271 if rows is not None:
1272 # we already have the lines in the buffer
1273 if len(self.buf) >= rows:
1274 new_rows, self.buf = self.buf[:rows], self.buf[rows:]
1276 # need some lines
1277 else:
1278 rows -= len(self.buf)
1280 if new_rows is None:
1281 if isinstance(self.data, list):
1282 if self.pos > len(self.data):
1283 raise StopIteration
1284 if rows is None:
1285 new_rows = self.data[self.pos :]
1286 new_pos = len(self.data)
1287 else:
1288 new_rows = self.data[self.pos : self.pos + rows]
1289 new_pos = self.pos + rows
1291 new_rows = self._remove_skipped_rows(new_rows)
1292 lines.extend(new_rows)
1293 self.pos = new_pos
1295 else:
1296 new_rows = []
1297 try:
1298 if rows is not None:
1299 row_index = 0
1300 row_ct = 0
1301 offset = self.pos if self.pos is not None else 0
1302 while row_ct < rows:
1303 new_row = next(self.data)
1304 if not self.skipfunc(offset + row_index):
1305 row_ct += 1
1306 row_index += 1
1307 new_rows.append(new_row)
1309 len_new_rows = len(new_rows)
1310 new_rows = self._remove_skipped_rows(new_rows)
1311 lines.extend(new_rows)
1312 else:
1313 rows = 0
1315 while True:
1316 next_row = self._next_iter_line(row_num=self.pos + rows + 1)
1317 rows += 1
1319 if next_row is not None:
1320 new_rows.append(next_row)
1321 len_new_rows = len(new_rows)
1323 except StopIteration:
1324 len_new_rows = len(new_rows)
1325 new_rows = self._remove_skipped_rows(new_rows)
1326 lines.extend(new_rows)
1327 if len(lines) == 0:
1328 raise
1329 self.pos += len_new_rows
1331 self.buf = []
1332 else:
1333 lines = new_rows
1335 if self.skipfooter:
1336 lines = lines[: -self.skipfooter]
1338 lines = self._check_comments(lines)
1339 if self.skip_blank_lines:
1340 lines = self._remove_empty_lines(lines)
1341 lines = self._check_thousands(lines)
1342 return self._check_decimal(lines)
1344 def _remove_skipped_rows(self, new_rows: list[list[Scalar]]) -> list[list[Scalar]]:
1345 if self.skiprows:
1346 return [
1347 row for i, row in enumerate(new_rows) if not self.skipfunc(i + self.pos)
1348 ]
1349 return new_rows
1351 def _set_no_thousand_columns(self) -> set[int]:
1352 no_thousands_columns: set[int] = set()
1353 if self.columns and self.parse_dates:
1354 assert self._col_indices is not None
1355 no_thousands_columns = self._set_noconvert_dtype_columns(
1356 self._col_indices, self.columns
1357 )
1358 if self.columns and self.dtype:
1359 assert self._col_indices is not None
1360 for i, col in zip(self._col_indices, self.columns, strict=True):
1361 if not isinstance(self.dtype, dict) and not is_numeric_dtype(
1362 self.dtype
1363 ):
1364 no_thousands_columns.add(i)
1365 if (
1366 isinstance(self.dtype, dict)
1367 and col in self.dtype
1368 and (
1369 not is_numeric_dtype(self.dtype[col])
1370 or is_bool_dtype(self.dtype[col])
1371 )
1372 ):
1373 no_thousands_columns.add(i)
1374 return no_thousands_columns
1377class FixedWidthReader(abc.Iterator):
1378 """
1379 A reader of fixed-width lines.
1380 """
1382 def __init__(
1383 self,
1384 f: IO[str] | ReadCsvBuffer[str],
1385 colspecs: list[tuple[int, int]] | Literal["infer"],
1386 delimiter: str | None,
1387 comment: str | None,
1388 skiprows: set[int] | None = None,
1389 infer_nrows: int = 100,
1390 ) -> None:
1391 self.f = f
1392 self.buffer: Iterator | None = None
1393 self.delimiter = "\r\n" + delimiter if delimiter else "\n\r\t "
1394 self.comment = comment
1395 if colspecs == "infer":
1396 self.colspecs = self.detect_colspecs(
1397 infer_nrows=infer_nrows, skiprows=skiprows
1398 )
1399 else:
1400 self.colspecs = colspecs
1402 if not isinstance(self.colspecs, (tuple, list)):
1403 raise TypeError(
1404 "column specifications must be a list or tuple, "
1405 f"input was a {type(colspecs).__name__}"
1406 )
1408 for colspec in self.colspecs:
1409 if not (
1410 isinstance(colspec, (tuple, list))
1411 and len(colspec) == 2
1412 and isinstance(colspec[0], (int, np.integer, type(None)))
1413 and isinstance(colspec[1], (int, np.integer, type(None)))
1414 ):
1415 raise TypeError(
1416 "Each column specification must be "
1417 "2 element tuple or list of integers"
1418 )
1420 def get_rows(self, infer_nrows: int, skiprows: set[int] | None = None) -> list[str]:
1421 """
1422 Read rows from self.f, skipping as specified.
1424 We distinguish buffer_rows (the first <= infer_nrows
1425 lines) from the rows returned to detect_colspecs
1426 because it's simpler to leave the other locations
1427 with skiprows logic alone than to modify them to
1428 deal with the fact we skipped some rows here as
1429 well.
1431 Parameters
1432 ----------
1433 infer_nrows : int
1434 Number of rows to read from self.f, not counting
1435 rows that are skipped.
1436 skiprows: set, optional
1437 Indices of rows to skip.
1439 Returns
1440 -------
1441 detect_rows : list of str
1442 A list containing the rows to read.
1444 """
1445 if skiprows is None:
1446 skiprows = set()
1447 buffer_rows = []
1448 detect_rows = []
1449 for i, row in enumerate(self.f):
1450 if i not in skiprows:
1451 detect_rows.append(row)
1452 buffer_rows.append(row)
1453 if len(detect_rows) >= infer_nrows:
1454 break
1455 self.buffer = iter(buffer_rows)
1456 return detect_rows
1458 def detect_colspecs(
1459 self, infer_nrows: int = 100, skiprows: set[int] | None = None
1460 ) -> list[tuple[int, int]]:
1461 # Regex escape the delimiters
1462 delimiters = "".join([rf"\{x}" for x in self.delimiter])
1463 pattern = re.compile(f"([^{delimiters}]+)")
1464 rows = self.get_rows(infer_nrows, skiprows)
1465 if not rows:
1466 raise EmptyDataError("No rows from which to infer column width")
1467 max_len = max(map(len, rows))
1468 mask = np.zeros(max_len + 1, dtype=int)
1469 if self.comment is not None:
1470 rows = [row.partition(self.comment)[0] for row in rows]
1471 for row in rows:
1472 for m in pattern.finditer(row):
1473 mask[m.start() : m.end()] = 1
1474 shifted = np.roll(mask, 1)
1475 shifted[0] = 0
1476 edges = np.where((mask ^ shifted) == 1)[0]
1477 edge_pairs = list(zip(edges[::2], edges[1::2], strict=True))
1478 return edge_pairs
1480 def __next__(self) -> list[str]:
1481 if self.buffer is not None:
1482 try:
1483 line = next(self.buffer)
1484 except StopIteration:
1485 self.buffer = None
1486 line = next(self.f) # type: ignore[arg-type]
1487 else:
1488 line = next(self.f) # type: ignore[arg-type]
1489 # Note: 'colspecs' is a sequence of half-open intervals.
1490 return [line[from_:to].strip(self.delimiter) for (from_, to) in self.colspecs]
1493class FixedWidthFieldParser(PythonParser):
1494 """
1495 Specialization that Converts fixed-width fields into DataFrames.
1496 See PythonParser for details.
1497 """
1499 def __init__(self, f: ReadCsvBuffer[str], **kwds) -> None:
1500 # Support iterators, convert to a list.
1501 self.colspecs = kwds.pop("colspecs")
1502 self.infer_nrows = kwds.pop("infer_nrows")
1503 PythonParser.__init__(self, f, **kwds)
1505 def _make_reader(self, f: IO[str] | ReadCsvBuffer[str]) -> FixedWidthReader:
1506 return FixedWidthReader(
1507 f,
1508 self.colspecs,
1509 self.delimiter,
1510 self.comment,
1511 self.skiprows,
1512 self.infer_nrows,
1513 )
1515 def _remove_empty_lines(self, lines: list[list[T]]) -> list[list[T]]:
1516 """
1517 Returns the list of lines without the empty ones. With fixed-width
1518 fields, empty lines become arrays of empty strings.
1520 See PythonParser._remove_empty_lines.
1521 """
1522 return [
1523 line
1524 for line in lines
1525 if any(not isinstance(e, str) or e.strip() for e in line)
1526 ]
1529def _validate_skipfooter_arg(skipfooter: int) -> int:
1530 """
1531 Validate the 'skipfooter' parameter.
1533 Checks whether 'skipfooter' is a non-negative integer.
1534 Raises a ValueError if that is not the case.
1536 Parameters
1537 ----------
1538 skipfooter : non-negative integer
1539 The number of rows to skip at the end of the file.
1541 Returns
1542 -------
1543 validated_skipfooter : non-negative integer
1544 The original input if the validation succeeds.
1546 Raises
1547 ------
1548 ValueError : 'skipfooter' was not a non-negative integer.
1549 """
1550 if not is_integer(skipfooter):
1551 raise ValueError("skipfooter must be an integer")
1553 if skipfooter < 0:
1554 raise ValueError("skipfooter cannot be negative")
1556 # Incompatible return value type (got "Union[int, integer[Any]]", expected "int")
1557 return skipfooter # type: ignore[return-value]