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 abc
4import csv
5from io import StringIO
6import re
7from typing import (
8 IO,
9 TYPE_CHECKING,
10 Any,
11 Literal,
12 cast,
13 final,
14)
15import warnings
17import numpy as np
19from pandas._libs import lib
20from pandas.errors import (
21 EmptyDataError,
22 ParserError,
23 ParserWarning,
24)
25from pandas.util._decorators import cache_readonly
26from pandas.util._exceptions import find_stack_level
28from pandas.core.dtypes.astype import astype_array
29from pandas.core.dtypes.common import (
30 is_bool_dtype,
31 is_extension_array_dtype,
32 is_integer,
33 is_numeric_dtype,
34 is_string_dtype,
35 pandas_dtype,
36)
37from pandas.core.dtypes.dtypes import (
38 CategoricalDtype,
39 ExtensionDtype,
40)
42from pandas.core import algorithms
43from pandas.core.arrays import (
44 Categorical,
45 ExtensionArray,
46)
47from pandas.core.arrays.boolean import BooleanDtype
48from pandas.core.indexes.api import Index
50from pandas.io.common import (
51 dedup_names,
52 is_potential_multi_index,
53 mangle_dupe_names,
54)
55from pandas.io.parsers.base_parser import (
56 ParserBase,
57 evaluate_callable_usecols,
58 get_na_values,
59 parser_defaults,
60 validate_parse_dates_presence,
61)
63if TYPE_CHECKING:
64 from collections.abc import (
65 Hashable,
66 Iterator,
67 Mapping,
68 Sequence,
69 )
71 from pandas._typing import (
72 ArrayLike,
73 DtypeObj,
74 ReadCsvBuffer,
75 Scalar,
76 T,
77 )
79 from pandas import (
80 MultiIndex,
81 Series,
82 )
84# BOM character (byte order mark)
85# This exists at the beginning of a file to indicate endianness
86# of a file (stream). Unfortunately, this marker screws up parsing,
87# so we need to remove it if we see it.
88_BOM = "\ufeff"
91class PythonParser(ParserBase):
92 _no_thousands_columns: set[int]
93 # field counts the file itself implies; set only when index_col is False
94 _first_row_len: int
95 _header_row_len: int
97 def __init__(self, f: ReadCsvBuffer[str] | list, **kwds) -> None:
98 """
99 Workhorse function for processing nested list into DataFrame
100 """
101 super().__init__(kwds)
103 self.data: Iterator[list[str]] | list[list[Scalar]] = []
104 self.buf: list = []
105 # Line number of each line pushed onto self.buf. Every site that shrinks
106 # self.buf drops entries from the front, so it stays a suffix of this list.
107 self.buf_pos: list[int] = []
108 self.pos = 0
109 self.line_pos = 0
110 self._first_row_len = 0
111 self._header_row_len = 0
113 self.skiprows = kwds["skiprows"]
115 if callable(self.skiprows):
116 self.skipfunc = self.skiprows
117 else:
118 self.skipfunc = lambda x: x in self.skiprows
120 self.skipfooter = _validate_skipfooter_arg(kwds["skipfooter"])
121 self.delimiter = kwds["delimiter"]
123 self.quotechar = kwds["quotechar"]
124 if isinstance(self.quotechar, str):
125 self.quotechar = str(self.quotechar)
127 self.escapechar = kwds["escapechar"]
128 self.doublequote = kwds["doublequote"]
129 self.skipinitialspace = kwds["skipinitialspace"]
130 self.lineterminator = kwds["lineterminator"]
131 self.quoting = kwds["quoting"]
132 self.skip_blank_lines = kwds["skip_blank_lines"]
134 # Passed from read_excel
135 self.has_index_names = kwds.get("has_index_names", False)
137 self.thousands = kwds["thousands"]
138 self.decimal = kwds["decimal"]
140 self.comment = kwds["comment"]
142 # Set self.data to something that can read lines.
143 if isinstance(f, list):
144 # read_excel: f is a nested list, can contain non-str
145 self.data = f
146 else:
147 assert hasattr(f, "readline")
148 # yields list of str
149 self.data = self._make_reader(f)
151 # Get columns in two steps: infer from data, then
152 # infer column indices from self.usecols if it is specified.
153 self._col_indices: list[int] | None = None
154 columns: list[list[Scalar | None]]
155 (
156 columns,
157 self.num_original_columns,
158 self.unnamed_cols,
159 ) = self._infer_columns()
161 # Now self.columns has the set of columns that we will process.
162 # The original set is stored in self.original_columns.
163 # error: Cannot determine type of 'index_names'
164 (
165 self.columns,
166 self.index_names,
167 self.col_names,
168 _,
169 ) = self._extract_multi_indexer_columns(
170 columns,
171 self.index_names,
172 )
174 # get popped off for index
175 self.orig_names: list[Hashable] = list(self.columns)
177 index_names, self.orig_names, self.columns = self._get_index_name()
178 if self.index_names is None:
179 self.index_names = index_names
181 if self._col_indices is None:
182 self._col_indices = list(range(len(self.columns)))
184 self._no_thousands_columns = self._set_no_thousand_columns()
186 if len(self.decimal) != 1:
187 raise ValueError("Only length-1 decimal markers supported")
189 @cache_readonly
190 def num(self) -> re.Pattern:
191 decimal = re.escape(self.decimal)
192 if self.thousands is None:
193 regex = rf"^[\-\+]?[0-9]*({decimal}[0-9]*)?([0-9]?(E|e)\-?[0-9]+)?$"
194 else:
195 thousands = re.escape(self.thousands)
196 # GH#52619 - use non-backtracking structure to avoid catastrophic
197 # backtracking on cells with many comma-separated digit groups
198 # followed by non-numeric text.
199 regex = (
200 rf"^[\-\+]?(?:[0-9]+(?:{thousands}[0-9]+)*{thousands}?)?"
201 rf"({decimal}[0-9]*)?"
202 rf"((E|e)\-?[0-9]+)?$"
203 )
204 return re.compile(regex)
206 def _make_reader(self, f: IO[str] | ReadCsvBuffer[str]) -> Iterator[list[str]]:
207 sep = self.delimiter
209 if sep is None or len(sep) == 1:
210 if self.lineterminator:
211 raise ValueError(
212 "Custom line terminators not supported in python parser (yet)"
213 )
215 class MyDialect(csv.Dialect):
216 delimiter = self.delimiter
217 quotechar = self.quotechar
218 escapechar = self.escapechar
219 doublequote = self.doublequote
220 skipinitialspace = self.skipinitialspace
221 quoting = self.quoting
222 lineterminator = "\n"
224 dia = MyDialect
226 if sep is not None:
227 dia.delimiter = sep
228 # Skip rows at file level before csv.reader sees them
229 # prevents CSV parsing errors on lines that will be discarded
230 if self.skiprows is not None:
231 while self.skipfunc(self.pos):
232 line = f.readline()
233 if not line:
234 break
235 self.pos += 1
236 else:
237 # attempt to sniff the delimiter from the first valid line,
238 # i.e. no comment line and not in skiprows
239 line = f.readline()
240 lines = self._check_comments([[line]])[0]
241 while self.skipfunc(self.pos) or not lines:
242 self.pos += 1
243 line = f.readline()
244 lines = self._check_comments([[line]])[0]
245 lines_str = cast("list[str]", lines)
247 # since `line` was a string, lines will be a list containing
248 # only a single string
249 line = lines_str[0]
251 self.pos += 1
252 self.line_pos += 1
253 sniffed = csv.Sniffer().sniff(line)
254 dia.delimiter = sniffed.delimiter
256 # Note: encoding is irrelevant here
257 line_rdr = csv.reader(StringIO(line), dialect=dia)
258 self.buf.extend(list(line_rdr))
259 self.buf_pos.append(self.pos - 1)
261 # Note: encoding is irrelevant here
262 reader = csv.reader(f, dialect=dia, strict=True)
264 else:
266 def _read():
267 line = f.readline()
268 pat = re.compile(sep)
270 yield pat.split(line.strip())
272 for line in f:
273 yield pat.split(line.strip())
275 reader = _read()
277 return reader
279 def read(
280 self, rows: int | None = None
281 ) -> tuple[
282 Index | None,
283 Sequence[Hashable] | MultiIndex,
284 Mapping[Hashable, ArrayLike | Series],
285 ]:
286 try:
287 content = self._get_lines(rows)
288 except StopIteration:
289 if self._first_chunk:
290 content = []
291 else:
292 self.close()
293 raise
295 # done with first read, next time raise StopIteration
296 self._first_chunk = False
298 index: Index | None
299 columns: Sequence[Hashable] = list(self.orig_names)
300 if not content: # pragma: no cover
301 # DataFrame with the right metadata, even though it's length 0
302 # error: Cannot determine type of 'index_col'
303 names = dedup_names(
304 self.orig_names,
305 is_potential_multi_index(
306 self.orig_names,
307 self.index_col,
308 ),
309 )
310 index, columns, col_dict = self._get_empty_meta(
311 names,
312 self.dtype,
313 )
314 conv_columns = self._maybe_make_multi_index_columns(columns, self.col_names)
315 return index, conv_columns, col_dict
317 # handle new style for names in index
318 indexnamerow = None
319 if self.has_index_names and sum(
320 int(v == "" or v is None) for v in content[0]
321 ) == len(columns):
322 indexnamerow = content[0]
323 content = content[1:]
325 alldata = self._rows_to_cols(content)
326 data, columns = self._exclude_implicit_index(alldata)
328 conv_data = self._convert_data(data)
329 conv_data = self._do_date_conversions(columns, conv_data)
331 index, result_columns = self._make_index(alldata, columns, indexnamerow)
333 return index, result_columns, conv_data
335 def _exclude_implicit_index(
336 self,
337 alldata: list[np.ndarray],
338 ) -> tuple[Mapping[Hashable, np.ndarray], Sequence[Hashable]]:
339 # error: Cannot determine type of 'index_col'
340 names = dedup_names(
341 self.orig_names,
342 is_potential_multi_index(
343 self.orig_names,
344 self.index_col,
345 ),
346 )
348 offset = 0
349 if self._implicit_index:
350 offset = len(self.index_col)
352 len_alldata = len(alldata)
353 self._check_data_length(names, alldata)
355 return {
356 name: alldata[i + offset] for i, name in enumerate(names) if i < len_alldata
357 }, names
359 # legacy
360 def get_chunk(
361 self, size: int | None = None
362 ) -> tuple[
363 Index | None,
364 Sequence[Hashable] | MultiIndex,
365 Mapping[Hashable, ArrayLike | Series],
366 ]:
367 if size is None:
368 # error: "PythonParser" has no attribute "chunksize"
369 size = self.chunksize # type: ignore[attr-defined]
370 return self.read(rows=size)
372 def _convert_data(
373 self,
374 data: Mapping[Hashable, np.ndarray],
375 ) -> Mapping[Hashable, ArrayLike]:
376 # apply converters
377 clean_conv = self._clean_mapping(self.converters)
378 clean_dtypes = self._clean_mapping(self.dtype)
380 # Apply NA values.
381 clean_na_values = {}
382 clean_na_fvalues = {}
384 if isinstance(self.na_values, dict):
385 for col in self.na_values:
386 if col is not None:
387 na_value = self.na_values[col]
388 na_fvalue = self.na_fvalues[col]
390 if isinstance(col, int) and col not in self.orig_names:
391 col = self.orig_names[col]
393 clean_na_values[col] = na_value
394 clean_na_fvalues[col] = na_fvalue
395 else:
396 clean_na_values = self.na_values
397 clean_na_fvalues = self.na_fvalues
399 return self._convert_to_ndarrays(
400 data,
401 clean_na_values,
402 clean_na_fvalues,
403 clean_conv,
404 clean_dtypes,
405 )
407 @final
408 def _convert_to_ndarrays(
409 self,
410 dct: Mapping,
411 na_values,
412 na_fvalues,
413 converters=None,
414 dtypes=None,
415 ) -> dict[Any, np.ndarray]:
416 result = {}
417 parse_date_cols = validate_parse_dates_presence(self.parse_dates, self.columns)
418 for c, values in dct.items():
419 conv_f = None if converters is None else converters.get(c, None)
420 if isinstance(dtypes, dict):
421 cast_type = dtypes.get(c, None)
422 else:
423 # single dtype or None
424 cast_type = dtypes
426 if self.na_filter:
427 col_na_values, col_na_fvalues = get_na_values(
428 c, na_values, na_fvalues, self.keep_default_na
429 )
430 else:
431 col_na_values, col_na_fvalues = set(), set()
433 if c in parse_date_cols:
434 # GH#26203 Do not convert columns which get converted to dates
435 # but replace nans to ensure to_datetime works
436 mask = algorithms.isin(
437 values,
438 set(col_na_values) | col_na_fvalues, # pyright: ignore[reportArgumentType]
439 )
440 np.putmask(values, mask, np.nan)
441 result[c] = values
442 continue
444 if conv_f is not None:
445 # conv_f applied to data before inference
446 if cast_type is not None:
447 warnings.warn(
448 (
449 "Both a converter and dtype were specified "
450 f"for column {c} - only the converter will be used."
451 ),
452 ParserWarning,
453 stacklevel=find_stack_level(),
454 )
456 try:
457 values = lib.map_infer(values, conv_f)
458 except ValueError:
459 mask = algorithms.isin(values, list(na_values)).view(np.uint8)
460 values = lib.map_infer_mask(values, conv_f, mask)
462 cvals, na_count = self._infer_types(
463 values,
464 set(col_na_values) | col_na_fvalues,
465 cast_type is None,
466 try_num_bool=False,
467 )
468 else:
469 is_ea = is_extension_array_dtype(cast_type)
470 is_str_or_ea_dtype = is_ea or is_string_dtype(cast_type)
471 # skip inference if specified dtype is object
472 # or casting to an EA
473 try_num_bool = not (cast_type and is_str_or_ea_dtype)
475 # general type inference and conversion
476 cvals, na_count = self._infer_types(
477 values,
478 set(col_na_values) | col_na_fvalues,
479 cast_type is None,
480 try_num_bool,
481 )
483 # type specified in dtype param or cast_type is an EA
484 if cast_type is not None:
485 cast_type = pandas_dtype(cast_type)
486 if cast_type and (cvals.dtype != cast_type or is_ea):
487 if not is_ea and na_count > 0:
488 if is_bool_dtype(cast_type):
489 raise ValueError(f"Bool column has NA values in column {c}")
490 cvals = self._cast_types(cvals, cast_type, c)
492 result[c] = cvals
493 return result
495 @final
496 def _cast_types(self, values: ArrayLike, cast_type: DtypeObj, column) -> ArrayLike:
497 """
498 Cast values to specified type
500 Parameters
501 ----------
502 values : ndarray or ExtensionArray
503 cast_type : np.dtype or ExtensionDtype
504 dtype to cast values to
505 column : string
506 column name - used only for error reporting
508 Returns
509 -------
510 converted : ndarray or ExtensionArray
511 """
512 if isinstance(cast_type, CategoricalDtype):
513 cats = Index(values, copy=False).unique().dropna()
514 codes = cats.get_indexer(values)
515 if cast_type.categories is None:
516 # GH#56044 mirror the type inference performed on ordinary
517 # (non-categorical) columns so that all engines agree.
518 # to_numeric is unaware of the thousands/decimal options, so
519 # keep string categories when those are set.
520 convert_numeric = (
521 self.thousands is None
522 and self.decimal == parser_defaults["decimal"]
523 )
524 converted = Categorical._maybe_convert_categories(
525 cats,
526 true_values=self.true_values,
527 false_values=self.false_values,
528 convert_numeric=convert_numeric,
529 convert_bool=True,
530 )
531 values = Categorical._from_converted_categories(
532 cats if converted is None else converted,
533 codes,
534 ordered=cast_type.ordered,
535 )
536 else:
537 values = Categorical._from_inferred_categories(
538 cats, codes, cast_type, true_values=self.true_values
539 )
541 # use the EA's implementation of casting
542 elif isinstance(cast_type, ExtensionDtype):
543 array_type = cast_type.construct_array_type()
544 try:
545 if isinstance(cast_type, BooleanDtype):
546 # error: Unexpected keyword argument "true_values" for
547 # "_from_sequence_of_strings" of "ExtensionArray"
548 values_str = [str(val) for val in values]
549 return array_type._from_sequence_of_strings( # type: ignore[call-arg]
550 values_str,
551 dtype=cast_type,
552 true_values=self.true_values, # pyright: ignore[reportCallIssue]
553 false_values=self.false_values, # pyright: ignore[reportCallIssue]
554 none_values=self.na_values, # pyright: ignore[reportCallIssue]
555 )
556 else:
557 return array_type._from_sequence_of_strings(values, dtype=cast_type)
558 except NotImplementedError as err:
559 raise NotImplementedError(
560 f"Extension Array: {array_type} must implement "
561 "_from_sequence_of_strings in order to be used in parser methods"
562 ) from err
564 elif isinstance(values, ExtensionArray):
565 casted: ArrayLike = values.astype(cast_type, copy=False)
566 # with dtype_backend="pyarrow", _infer_types has already boxed the
567 # column, so an explicit numpy integer dtype reaches this branch,
568 # not the one below
569 _validate_integer_cast(values, casted, cast_type, column)
570 values = casted
571 elif issubclass(cast_type.type, str):
572 # TODO: why skipna=True here and False above? some tests depend
573 # on it here, but nothing fails if we change it above
574 # (as no tests get there as of 2022-12-06)
575 values = lib.ensure_string_array(
576 values, skipna=True, convert_na_value=False
577 )
578 else:
579 try:
580 casted = astype_array(values, cast_type, copy=True)
581 except ValueError as err:
582 raise ValueError(
583 f"Unable to convert column {column} to type {cast_type}"
584 ) from err
585 _validate_integer_cast(values, casted, cast_type, column)
586 values = casted
587 return values
589 @cache_readonly
590 def _have_mi_columns(self) -> bool:
591 if self.header is None:
592 return False
594 header = self.header
595 if isinstance(header, (list, tuple, np.ndarray)):
596 return len(header) > 1
597 else:
598 return False
600 def _infer_columns(
601 self,
602 ) -> tuple[list[list[Scalar | None]], int, set[Scalar | None]]:
603 names = self.names
604 num_original_columns = 0
605 clear_buffer = True
606 unnamed_cols: set[Scalar | None] = set()
608 if self.header is not None:
609 header = self.header
610 have_mi_columns = self._have_mi_columns
612 if isinstance(header, (list, tuple, np.ndarray)):
613 # we have a mi columns, so read an extra line
614 if have_mi_columns:
615 header = [*list(header), header[-1] + 1]
616 else:
617 header = [header]
619 columns: list[list[Scalar | None]] = []
620 for level, hr in enumerate(header):
621 try:
622 line = self._buffered_line()
624 while self.line_pos <= hr:
625 line = self._next_line()
627 except StopIteration as err:
628 if 0 < self.line_pos <= hr and (
629 not have_mi_columns or hr != header[-1]
630 ):
631 # If no rows we want to raise a different message and if
632 # we have mi columns, the last line is not part of the header
633 joi = list(map(str, header[:-1] if have_mi_columns else header))
634 msg = f"[{','.join(joi)}], len of {len(joi)}, "
635 raise ValueError(
636 f"Passed header={msg}but only {self.line_pos} lines in file"
637 ) from err
639 # We have an empty file, so check
640 # if columns are provided. That will
641 # serve as the 'line' for parsing
642 if have_mi_columns and hr > 0:
643 if clear_buffer:
644 self.buf.clear()
645 columns.append([None] * len(columns[-1]))
646 return columns, num_original_columns, unnamed_cols
648 if not self.names:
649 raise EmptyDataError("No columns to parse from file") from err
651 line = self.names[:]
653 this_columns: list[Scalar | None] = []
654 this_unnamed_cols = []
656 for i, c in enumerate(line):
657 if c == "":
658 if have_mi_columns:
659 col_name = f"Unnamed: {i}_level_{level}"
660 else:
661 col_name = f"Unnamed: {i}"
663 this_unnamed_cols.append(i)
664 this_columns.append(col_name)
665 else:
666 this_columns.append(c)
668 if not have_mi_columns:
669 this_columns = cast(
670 "list[Scalar | None]",
671 mangle_dupe_names(this_columns, this_unnamed_cols, self.dtype),
672 )
673 elif have_mi_columns:
674 # if we have grabbed an extra line, but it's not in our
675 # format so save in the buffer, and create a blank extra
676 # line for the rest of the parsing code
677 if hr == header[-1]:
678 lc = len(this_columns)
679 sic = self.index_col
680 ic = len(sic) if sic is not None else 0
681 unnamed_count = len(this_unnamed_cols)
683 # if wrong number of blanks or no index, not our format
684 if (lc != unnamed_count and lc - ic > unnamed_count) or ic == 0:
685 clear_buffer = False
686 this_columns = [None] * lc
687 self.buf = [self.buf[-1]]
689 columns.append(this_columns)
690 unnamed_cols.update({this_columns[i] for i in this_unnamed_cols})
692 if len(columns) == 1:
693 num_original_columns = len(this_columns)
695 if clear_buffer:
696 self.buf.clear()
698 first_line: list[Scalar] | None
699 if names is not None:
700 # Read first row after header to check if data are longer
701 try:
702 first_line = self._next_line()
703 except StopIteration:
704 first_line = None
706 len_first_data_row = 0 if first_line is None else len(first_line)
708 if len(names) > len(columns[0]) and len(names) > len_first_data_row:
709 raise ValueError(
710 "Number of passed names did not match "
711 "number of header fields in the file"
712 )
713 if len(columns) > 1:
714 raise TypeError("Cannot pass names with multi-index columns")
716 if self.index_col is False:
717 # num_original_columns becomes len(names) below, but with
718 # index_col=False the expected row width still comes from the
719 # file. see GH#49279
720 self._header_row_len = len(columns[0])
722 if self.usecols is not None:
723 # Set _use_cols. We don't store columns because they are
724 # overwritten.
725 self._handle_usecols(columns, names, num_original_columns)
726 else:
727 num_original_columns = len(names)
728 if self._col_indices is not None and len(names) != len(
729 self._col_indices
730 ):
731 columns = [[names[i] for i in sorted(self._col_indices)]]
732 else:
733 columns = [names]
734 else:
735 columns = self._handle_usecols(
736 columns, columns[0], num_original_columns
737 )
738 else:
739 ncols = len(self._header_line)
740 num_original_columns = ncols
742 if not names:
743 columns = [list(range(ncols))]
744 columns = self._handle_usecols(columns, columns[0], ncols)
745 elif self.usecols is None or len(names) >= ncols:
746 columns = self._handle_usecols([names], names, ncols)
747 num_original_columns = len(names)
748 elif not callable(self.usecols) and len(names) != len(self.usecols):
749 raise ValueError(
750 "Number of passed names did not match number of "
751 "header fields in the file"
752 )
753 else:
754 # Ignore output but set used columns.
755 columns = [names]
756 self._handle_usecols(columns, columns[0], ncols)
758 return columns, num_original_columns, unnamed_cols
760 @cache_readonly
761 def _header_line(self):
762 # Store line for reuse in _get_index_name
763 if self.header is not None:
764 return None
766 try:
767 line = self._buffered_line()
768 except StopIteration as err:
769 if not self.names:
770 raise EmptyDataError("No columns to parse from file") from err
772 line = self.names[:]
773 return line
775 def _handle_usecols(
776 self,
777 columns: list[list[Scalar | None]],
778 usecols_key: list[Scalar | None],
779 num_original_columns: int,
780 ) -> list[list[Scalar | None]]:
781 """
782 Sets self._col_indices
784 usecols_key is used if there are string usecols.
785 """
786 col_indices: set[int] | list[int]
787 if self.usecols is not None:
788 if callable(self.usecols):
789 col_indices = evaluate_callable_usecols(self.usecols, usecols_key)
790 elif any(isinstance(u, str) for u in self.usecols):
791 if len(columns) > 1:
792 raise ValueError(
793 "If using multiple headers, usecols must be integers."
794 )
795 col_indices = []
797 for col in self.usecols:
798 if isinstance(col, str):
799 try:
800 col_indices.append(usecols_key.index(col))
801 except ValueError:
802 self._validate_usecols_names(self.usecols, usecols_key)
803 else:
804 col_indices.append(col)
805 else:
806 missing_usecols = [
807 col for col in self.usecols if col >= num_original_columns
808 ]
809 if missing_usecols:
810 raise ParserError(
811 "Defining usecols with out-of-bounds indices is not allowed. "
812 f"{missing_usecols} are out-of-bounds.",
813 )
814 col_indices = self.usecols
816 columns = [
817 [n for i, n in enumerate(column) if i in col_indices]
818 for column in columns
819 ]
820 self._col_indices = sorted(col_indices)
821 return columns
823 def _buffered_line(self) -> list[Scalar]:
824 """
825 Return a line from buffer, filling buffer if required.
826 """
827 if len(self.buf) > 0:
828 return self.buf[0]
829 else:
830 return self._next_line()
832 def _check_for_bom(self, first_row: list[Scalar]) -> list[Scalar]:
833 """
834 Checks whether the file begins with the BOM character.
835 If it does, remove it. In addition, if there is quoting
836 in the field subsequent to the BOM, remove it as well
837 because it technically takes place at the beginning of
838 the name, not the middle of it.
839 """
840 # first_row will be a list, so we need to check
841 # that the list is not empty before proceeding.
842 if not first_row:
843 return first_row
845 # The first element of this row is the one that could have the
846 # BOM that we want to remove. Check that the first element is a
847 # string before proceeding.
848 if not isinstance(first_row[0], str):
849 return first_row
851 # Check that the string is not empty, as that would
852 # obviously not have a BOM at the start of it.
853 if not first_row[0]:
854 return first_row
856 # Since the string is non-empty, check that it does
857 # in fact begin with a BOM.
858 first_elt = first_row[0][0]
859 if first_elt != _BOM:
860 return first_row
862 first_row_bom = first_row[0]
863 new_row: str
865 if len(first_row_bom) > 1 and first_row_bom[1] == self.quotechar:
866 start = 2
867 quote = first_row_bom[1]
868 end = first_row_bom[2:].index(quote) + 2
870 # Extract the data between the quotation marks
871 new_row = first_row_bom[start:end]
873 # Extract any remaining data after the second
874 # quotation mark.
875 if len(first_row_bom) > end + 1:
876 new_row += first_row_bom[end + 1 :]
878 else:
879 # No quotation so just remove BOM from first element
880 new_row = first_row_bom[1:]
882 new_row_list: list[Scalar] = [new_row]
883 return new_row_list + first_row[1:]
885 def _is_line_empty(self, line: Sequence[Scalar]) -> bool:
886 """
887 Check if a line is empty or not.
889 Parameters
890 ----------
891 line : str, array-like
892 The line of data to check.
894 Returns
895 -------
896 boolean : Whether or not the line is empty.
897 """
898 return not line or all(not x for x in line)
900 def _next_line(self) -> list[Scalar]:
901 if isinstance(self.data, list):
902 while self.skipfunc(self.pos):
903 if self.pos >= len(self.data):
904 break
905 self.pos += 1
907 while True:
908 try:
909 line = self._check_comments([self.data[self.pos]])[0]
910 self.pos += 1
911 # either uncommented or blank to begin with
912 if not self.skip_blank_lines and (
913 self._is_line_empty(self.data[self.pos - 1]) or line
914 ):
915 break
916 if self.skip_blank_lines:
917 ret = self._remove_empty_lines([line])
918 if ret:
919 line = ret[0]
920 break
921 except IndexError as err:
922 raise StopIteration from err
923 else:
924 while self.skipfunc(self.pos):
925 # consume first: on an exhausted file this raises and self.pos must
926 # not advance past the last line, or skipfooter trims from the wrong
927 # end (GH#36827)
928 next(self.data)
929 self.pos += 1
931 while True:
932 orig_line = self._next_iter_line(row_num=self.pos + 1)
933 self.pos += 1
935 if orig_line is not None:
936 line = self._check_comments([orig_line])[0]
938 if self.skip_blank_lines:
939 ret = self._remove_empty_lines([line])
941 if ret:
942 line = ret[0]
943 break
944 elif self._is_line_empty(orig_line) or line:
945 break
947 # This was the first line of the file,
948 # which could contain the BOM at the
949 # beginning of it.
950 if self.pos == 1:
951 line = self._check_for_bom(line)
953 self.line_pos += 1
954 self.buf.append(line)
955 self.buf_pos.append(self.pos - 1)
956 return line
958 def _alert_malformed(self, msg: str, row_num: int) -> None:
959 """
960 Alert a user about a malformed row, depending on value of
961 `self.on_bad_lines` enum.
963 If `self.on_bad_lines` is BLHM_ERROR, the alert will be `ParserError`.
964 If `self.on_bad_lines` is BLHM_WARN, the alert will be printed out.
966 Parameters
967 ----------
968 msg: str
969 The error message to display.
970 row_num: int
971 The row number where the parsing error occurred.
972 Because this row number is displayed, we 1-index,
973 even though we 0-index internally.
974 """
975 if self.on_bad_lines == self.BadLineHandleMethod.BLHM_ERROR:
976 raise ParserError(msg)
977 if self.on_bad_lines == self.BadLineHandleMethod.BLHM_WARN or callable(
978 self.on_bad_lines
979 ):
980 warnings.warn(
981 f"Skipping line {row_num}: {msg}\n",
982 ParserWarning,
983 stacklevel=find_stack_level(),
984 )
986 def _next_iter_line(self, row_num: int) -> list[Scalar] | None:
987 """
988 Wrapper around iterating through `self.data` (CSV source).
990 When a CSV error is raised, we check for specific
991 error messages that allow us to customize the
992 error message displayed to the user.
994 Parameters
995 ----------
996 row_num: int
997 The row number of the line being parsed.
998 """
999 try:
1000 assert not isinstance(self.data, list)
1001 line = next(self.data)
1002 # lie about list[str] vs list[Scalar] to minimize ignores
1003 return line # type: ignore[return-value]
1004 except csv.Error as e:
1005 if self.on_bad_lines in (
1006 self.BadLineHandleMethod.BLHM_ERROR,
1007 self.BadLineHandleMethod.BLHM_WARN,
1008 ):
1009 msg = str(e)
1011 if "NULL byte" in msg or "line contains NUL" in msg:
1012 msg = (
1013 "NULL byte detected. This byte "
1014 "cannot be processed in Python's "
1015 "native csv library at the moment, "
1016 "so please pass in engine='c' instead"
1017 )
1019 if self.skipfooter > 0:
1020 reason = (
1021 "Error could possibly be due to "
1022 "parsing errors in the skipped footer rows "
1023 "(the skipfooter keyword is only applied "
1024 "after Python's csv library has parsed "
1025 "all rows)."
1026 )
1027 msg += ". " + reason
1029 self._alert_malformed(msg, row_num)
1030 return None
1032 def _check_comments(self, lines: list[list[Scalar]]) -> list[list[Scalar]]:
1033 if self.comment is None:
1034 return lines
1035 ret = []
1036 for line in lines:
1037 rl = []
1038 for x in line:
1039 if (
1040 not isinstance(x, str)
1041 or self.comment not in x
1042 or x in self.na_values
1043 ):
1044 rl.append(x)
1045 else:
1046 x = x[: x.find(self.comment)]
1047 if len(x) > 0:
1048 rl.append(x)
1049 break
1050 ret.append(rl)
1051 return ret
1053 def _remove_empty_lines(self, lines: list[list[T]]) -> list[list[T]]:
1054 """
1055 Iterate through the lines and remove any that are
1056 either empty or contain only one whitespace value
1058 Parameters
1059 ----------
1060 lines : list of list of Scalars
1061 The array of lines that we are to filter.
1063 Returns
1064 -------
1065 filtered_lines : list of list of Scalars
1066 The same array of lines with the "empty" ones removed.
1067 """
1068 # Remove empty lines and lines with only one whitespace value
1069 ret = [
1070 line
1071 for line in lines
1072 if (
1073 len(line) > 1
1074 or (
1075 len(line) == 1 and (not isinstance(line[0], str) or line[0].strip())
1076 )
1077 )
1078 ]
1079 return ret
1081 def _check_thousands(self, lines: list[list[Scalar]]) -> list[list[Scalar]]:
1082 if self.thousands is None:
1083 return lines
1085 return self._search_replace_num_columns(
1086 lines=lines, search=self.thousands, replace=""
1087 )
1089 def _search_replace_num_columns(
1090 self, lines: list[list[Scalar]], search: str, replace: str
1091 ) -> list[list[Scalar]]:
1092 ret = []
1093 for line in lines:
1094 rl = []
1095 for i, x in enumerate(line):
1096 if (
1097 not isinstance(x, str)
1098 or search not in x
1099 or i in self._no_thousands_columns
1100 or not self.num.search(x.strip())
1101 ):
1102 rl.append(x)
1103 else:
1104 rl.append(x.replace(search, replace))
1105 ret.append(rl)
1106 return ret
1108 def _check_decimal(self, lines: list[list[Scalar]]) -> list[list[Scalar]]:
1109 if self.decimal == parser_defaults["decimal"]:
1110 return lines
1112 return self._search_replace_num_columns(
1113 lines=lines, search=self.decimal, replace="."
1114 )
1116 def _get_index_name(
1117 self,
1118 ) -> tuple[Sequence[Hashable] | None, list[Hashable], list[Hashable]]:
1119 """
1120 Try several cases to get lines:
1122 0) There are headers on row 0 and row 1 and their
1123 total summed lengths equals the length of the next line.
1124 Treat row 0 as columns and row 1 as indices
1125 1) Look for implicit index: there are more columns
1126 on row 1 than row 0. If this is true, assume that row
1127 1 lists index columns and row 0 lists normal columns.
1128 2) Get index from the columns if it was listed.
1129 """
1130 columns: Sequence[Hashable] = self.orig_names
1131 orig_names = list(columns)
1132 columns = list(columns)
1134 line: list[Scalar] | None
1135 if self._header_line is not None:
1136 line = self._header_line
1137 else:
1138 try:
1139 line = self._next_line()
1140 except StopIteration:
1141 line = None
1143 next_line: list[Scalar] | None
1144 try:
1145 next_line = self._next_line()
1146 except StopIteration:
1147 next_line = None
1149 if self.index_col is False:
1150 # index_col=False suppresses implicit-index inference, so the first data
1151 # row can widen the expected width past the header, as in the c engine.
1152 # Take it from buf, not `line` -- `line` is a later row, or None, once
1153 # _infer_columns has buffered the first. see GH#49279
1154 first_row = self.buf[0] if self.buf else line
1155 if first_row is not None:
1156 self._first_row_len = len(first_row)
1158 # implicitly index_col=0 b/c 1 fewer column names
1159 implicit_first_cols = 0
1160 if line is not None:
1161 # leave it 0, #2442
1162 # Case 1
1163 index_col = self.index_col
1164 if index_col is not False:
1165 implicit_first_cols = len(line) - self.num_original_columns
1167 # Case 0
1168 if (
1169 next_line is not None
1170 and self.header is not None
1171 and index_col is not False
1172 ):
1173 if len(next_line) == len(line) + self.num_original_columns:
1174 # column and index names on diff rows
1175 self.index_col = list(range(len(line)))
1176 self.buf = self.buf[1:]
1178 for c in reversed(line):
1179 columns.insert(0, c)
1181 # Update list of original names to include all indices.
1182 orig_names = list(columns)
1183 self.num_original_columns = len(columns)
1184 return line, orig_names, columns
1186 if implicit_first_cols > 0:
1187 # Case 1
1188 self._implicit_index = True
1189 if self.index_col is None:
1190 self.index_col = list(range(implicit_first_cols))
1192 index_name = None
1194 else:
1195 # Case 2
1196 (index_name, _, self.index_col) = self._clean_index_names(
1197 columns, self.index_col
1198 )
1200 return index_name, orig_names, columns
1202 def _rows_to_cols(self, content: list[list[Scalar]]) -> list[np.ndarray]:
1203 col_len = self.num_original_columns
1205 if self._implicit_index:
1206 col_len += len(self.index_col)
1208 max_len = max(len(row) for row in content)
1210 # Check that there are no rows with too many
1211 # elements in their row (rows with too few
1212 # elements are padded with NaN).
1213 expected_len = max(col_len, self._first_row_len, self._header_row_len)
1214 if max_len > expected_len and self.usecols is None:
1215 footers = self.skipfooter if self.skipfooter else 0
1216 bad_lines = []
1218 iter_content = enumerate(content)
1219 content_len = len(content)
1220 content = []
1222 for i, _content in iter_content:
1223 actual_len = len(_content)
1224 if actual_len > expected_len:
1225 if callable(self.on_bad_lines):
1226 new_l = self.on_bad_lines(_content)
1227 if new_l is not None:
1228 new_l = cast("list[Scalar]", new_l)
1229 if len(new_l) > expected_len:
1230 row_num = self.pos - (content_len - i + footers)
1231 bad_lines.append((row_num, len(new_l), "callable"))
1232 new_l = new_l[:expected_len]
1233 content.append(new_l)
1235 elif self.on_bad_lines in (
1236 self.BadLineHandleMethod.BLHM_ERROR,
1237 self.BadLineHandleMethod.BLHM_WARN,
1238 ):
1239 row_num = self.pos - (content_len - i + footers)
1240 bad_lines.append((row_num, actual_len, "normal"))
1241 if self.on_bad_lines == self.BadLineHandleMethod.BLHM_ERROR:
1242 break
1243 else:
1244 content.append(_content)
1246 for row_num, actual_len, source in bad_lines:
1247 msg = (
1248 f"Expected {expected_len} fields in line {row_num + 1}, "
1249 f"saw {actual_len}"
1250 )
1251 if source == "callable":
1252 msg += " from bad_lines callable"
1253 elif (
1254 self.delimiter
1255 and len(self.delimiter) > 1
1256 and self.quoting != csv.QUOTE_NONE
1257 ):
1258 # see gh-13374
1259 reason = (
1260 "Error could possibly be due to quotes being "
1261 "ignored when a multi-char delimiter is used."
1262 )
1263 msg += ". " + reason
1265 self._alert_malformed(msg, row_num + 1)
1267 # see gh-13320
1268 zipped_content = list(lib.to_object_array(content, min_width=col_len).T)
1270 if self.usecols:
1271 assert self._col_indices is not None
1272 col_indices = self._col_indices
1274 if self._implicit_index:
1275 zipped_content = [
1276 a
1277 for i, a in enumerate(zipped_content)
1278 if (
1279 i < len(self.index_col)
1280 or i - len(self.index_col) in col_indices
1281 )
1282 ]
1283 else:
1284 zipped_content = [
1285 a for i, a in enumerate(zipped_content) if i in col_indices
1286 ]
1287 return zipped_content
1289 def _get_lines(self, rows: int | None = None) -> list[list[Scalar]]:
1290 lines = self.buf
1291 new_rows = None
1292 num_buffered = len(self.buf)
1293 first_new_pos = self.pos
1294 # `rows` gets reused as a counter below, so latch whether we read to EOF
1295 read_to_eof = rows is None
1297 # already fetched some number
1298 if rows is not None:
1299 # we already have the lines in the buffer
1300 if len(self.buf) >= rows:
1301 new_rows, self.buf = self.buf[:rows], self.buf[rows:]
1303 # need some lines
1304 else:
1305 rows -= len(self.buf)
1307 if new_rows is None:
1308 if isinstance(self.data, list):
1309 if self.pos > len(self.data):
1310 raise StopIteration
1311 if rows is None:
1312 new_rows = self.data[self.pos :]
1313 new_pos = len(self.data)
1314 else:
1315 new_rows = self.data[self.pos : self.pos + rows]
1316 new_pos = self.pos + rows
1318 new_rows = self._remove_skipped_rows(new_rows)
1319 lines.extend(new_rows)
1320 self.pos = new_pos
1322 else:
1323 new_rows = []
1324 try:
1325 if rows is not None:
1326 row_index = 0
1327 row_ct = 0
1328 offset = self.pos if self.pos is not None else 0
1329 while row_ct < rows:
1330 new_row = next(self.data)
1331 if not self.skipfunc(offset + row_index):
1332 row_ct += 1
1333 row_index += 1
1334 new_rows.append(new_row)
1336 len_new_rows = len(new_rows)
1337 new_rows = self._remove_skipped_rows(new_rows)
1338 lines.extend(new_rows)
1339 else:
1340 rows = 0
1342 while True:
1343 next_row = self._next_iter_line(row_num=self.pos + rows + 1)
1344 rows += 1
1346 if next_row is not None:
1347 new_rows.append(next_row)
1349 except StopIteration:
1350 len_new_rows = len(new_rows)
1351 new_rows = self._remove_skipped_rows(new_rows)
1352 lines.extend(new_rows)
1353 if len(lines) == 0:
1354 raise
1355 self.pos += len_new_rows
1357 self.buf = []
1358 else:
1359 lines = new_rows
1361 if self.skipfooter:
1362 if read_to_eof:
1363 lines = self._remove_footer_lines(lines, num_buffered, first_new_pos)
1364 else:
1365 # With an explicit row count self.pos need not be the file's line
1366 # count, so there is no footer to measure from; keep the
1367 # pre-GH#36827 behavior.
1368 lines = lines[: -self.skipfooter]
1370 lines = self._check_comments(lines)
1371 if self.skip_blank_lines:
1372 lines = self._remove_empty_lines(lines)
1373 lines = self._check_thousands(lines)
1374 return self._check_decimal(lines)
1376 def _remove_footer_lines(
1377 self, lines: list[list[Scalar]], num_buffered: int, first_new_pos: int
1378 ) -> list[list[Scalar]]:
1379 """
1380 Drop the lines that fall within the last ``skipfooter`` lines of the file.
1382 Lines consumed while inferring the header, and ``skiprows`` lines, are
1383 missing from ``lines`` but still count towards ``skipfooter``, so the
1384 cutoff is applied by line number. Only called once the whole file has been
1385 read, so ``self.pos`` is its line count. See GH#36827.
1386 """
1387 positions = self.buf_pos[len(self.buf_pos) - num_buffered :]
1388 positions += [
1389 pos
1390 for pos in range(first_new_pos, self.pos)
1391 if not (self.skiprows and self.skipfunc(pos))
1392 ]
1394 cutoff = self.pos - self.skipfooter
1395 return lines[: sum(pos < cutoff for pos in positions)]
1397 def _remove_skipped_rows(self, new_rows: list[list[Scalar]]) -> list[list[Scalar]]:
1398 if self.skiprows:
1399 return [
1400 row for i, row in enumerate(new_rows) if not self.skipfunc(i + self.pos)
1401 ]
1402 return new_rows
1404 def _set_no_thousand_columns(self) -> set[int]:
1405 no_thousands_columns: set[int] = set()
1406 if self.columns and self.parse_dates:
1407 assert self._col_indices is not None
1408 no_thousands_columns = self._set_noconvert_dtype_columns(
1409 self._col_indices, self.columns
1410 )
1411 if self.columns and self.dtype:
1412 assert self._col_indices is not None
1413 for i, col in zip(self._col_indices, self.columns, strict=True):
1414 if not isinstance(self.dtype, dict) and not is_numeric_dtype(
1415 self.dtype
1416 ):
1417 no_thousands_columns.add(i)
1418 if (
1419 isinstance(self.dtype, dict)
1420 and col in self.dtype
1421 and (
1422 not is_numeric_dtype(self.dtype[col])
1423 or is_bool_dtype(self.dtype[col])
1424 )
1425 ):
1426 no_thousands_columns.add(i)
1427 return no_thousands_columns
1430class FixedWidthReader(abc.Iterator):
1431 """
1432 A reader of fixed-width lines.
1433 """
1435 def __init__(
1436 self,
1437 f: IO[str] | ReadCsvBuffer[str],
1438 colspecs: list[tuple[int, int]] | Literal["infer"],
1439 delimiter: str | None,
1440 comment: str | None,
1441 skiprows: set[int] | None = None,
1442 infer_nrows: int = 100,
1443 ) -> None:
1444 self.f = f
1445 self.buffer: Iterator | None = None
1446 self.delimiter = "\r\n" + delimiter if delimiter else "\n\r\t "
1447 self.comment = comment
1448 if colspecs == "infer":
1449 self.colspecs = self.detect_colspecs(
1450 infer_nrows=infer_nrows, skiprows=skiprows
1451 )
1452 else:
1453 self.colspecs = colspecs
1455 if not isinstance(self.colspecs, (tuple, list)):
1456 raise TypeError(
1457 "column specifications must be a list or tuple, "
1458 f"input was a {type(colspecs).__name__}"
1459 )
1461 for colspec in self.colspecs:
1462 if not (
1463 isinstance(colspec, (tuple, list))
1464 and len(colspec) == 2
1465 and isinstance(colspec[0], (int, np.integer, type(None)))
1466 and isinstance(colspec[1], (int, np.integer, type(None)))
1467 ):
1468 raise TypeError(
1469 "Each column specification must be "
1470 "2 element tuple or list of integers"
1471 )
1473 def get_rows(self, infer_nrows: int, skiprows: set[int] | None = None) -> list[str]:
1474 """
1475 Read rows from self.f, skipping as specified.
1477 We distinguish buffer_rows (the first <= infer_nrows
1478 lines) from the rows returned to detect_colspecs
1479 because it's simpler to leave the other locations
1480 with skiprows logic alone than to modify them to
1481 deal with the fact we skipped some rows here as
1482 well.
1484 Parameters
1485 ----------
1486 infer_nrows : int
1487 Number of rows to read from self.f, not counting
1488 rows that are skipped.
1489 skiprows: set, optional
1490 Indices of rows to skip.
1492 Returns
1493 -------
1494 detect_rows : list of str
1495 A list containing the rows to read.
1497 """
1498 if skiprows is None:
1499 skiprows = set()
1500 buffer_rows = []
1501 detect_rows = []
1502 for i, row in enumerate(self.f):
1503 if i not in skiprows:
1504 detect_rows.append(row)
1505 buffer_rows.append(row)
1506 if len(detect_rows) >= infer_nrows:
1507 break
1508 self.buffer = iter(buffer_rows)
1509 return detect_rows
1511 def detect_colspecs(
1512 self, infer_nrows: int = 100, skiprows: set[int] | None = None
1513 ) -> list[tuple[int, int]]:
1514 # Regex escape the delimiters
1515 delimiters = "".join([rf"\{x}" for x in self.delimiter])
1516 pattern = re.compile(f"([^{delimiters}]+)")
1517 rows = self.get_rows(infer_nrows, skiprows)
1518 if not rows:
1519 raise EmptyDataError("No rows from which to infer column width")
1520 max_len = max(map(len, rows))
1521 mask = np.zeros(max_len + 1, dtype=int)
1522 if self.comment is not None:
1523 rows = [row.partition(self.comment)[0] for row in rows]
1524 for row in rows:
1525 for m in pattern.finditer(row):
1526 mask[m.start() : m.end()] = 1
1527 shifted = np.roll(mask, 1)
1528 shifted[0] = 0
1529 edges = np.where((mask ^ shifted) == 1)[0]
1530 edge_pairs = list(zip(edges[::2], edges[1::2], strict=True))
1531 return edge_pairs
1533 def __next__(self) -> list[str]:
1534 if self.buffer is not None:
1535 try:
1536 line = next(self.buffer)
1537 except StopIteration:
1538 self.buffer = None
1539 line = next(self.f) # type: ignore[arg-type]
1540 else:
1541 line = next(self.f) # type: ignore[arg-type]
1542 # Note: 'colspecs' is a sequence of half-open intervals.
1543 return [line[from_:to].strip(self.delimiter) for (from_, to) in self.colspecs]
1546class FixedWidthFieldParser(PythonParser):
1547 """
1548 Specialization that Converts fixed-width fields into DataFrames.
1549 See PythonParser for details.
1550 """
1552 def __init__(self, f: ReadCsvBuffer[str], **kwds) -> None:
1553 # Support iterators, convert to a list.
1554 self.colspecs = kwds.pop("colspecs")
1555 self.infer_nrows = kwds.pop("infer_nrows")
1556 PythonParser.__init__(self, f, **kwds)
1558 def _make_reader(self, f: IO[str] | ReadCsvBuffer[str]) -> FixedWidthReader:
1559 return FixedWidthReader(
1560 f,
1561 self.colspecs,
1562 self.delimiter,
1563 self.comment,
1564 self.skiprows,
1565 self.infer_nrows,
1566 )
1568 def _remove_empty_lines(self, lines: list[list[T]]) -> list[list[T]]:
1569 """
1570 Returns the list of lines without the empty ones. With fixed-width
1571 fields, empty lines become arrays of empty strings.
1573 See PythonParser._remove_empty_lines.
1574 """
1575 return [
1576 line
1577 for line in lines
1578 if any(not isinstance(e, str) or e.strip() for e in line)
1579 ]
1582def _validate_skipfooter_arg(skipfooter: int) -> int:
1583 """
1584 Validate the 'skipfooter' parameter.
1586 Checks whether 'skipfooter' is a non-negative integer.
1587 Raises a ValueError if that is not the case.
1589 Parameters
1590 ----------
1591 skipfooter : non-negative integer
1592 The number of rows to skip at the end of the file.
1594 Returns
1595 -------
1596 validated_skipfooter : non-negative integer
1597 The original input if the validation succeeds.
1599 Raises
1600 ------
1601 ValueError : 'skipfooter' was not a non-negative integer.
1602 """
1603 if not is_integer(skipfooter):
1604 raise ValueError("skipfooter must be an integer")
1606 if skipfooter < 0:
1607 raise ValueError("skipfooter cannot be negative")
1609 return skipfooter # pyright: ignore[reportReturnType]
1612def _validate_integer_cast(
1613 original: ArrayLike, casted: ArrayLike, cast_type: np.dtype, column
1614) -> None:
1615 """
1616 Raise if casting to an integer dtype wrapped around (GH#55232).
1617 """
1618 if cast_type.kind not in "iu" or original.dtype.kind not in "iuf":
1619 return
1620 values = np.asarray(original)
1621 if np.can_cast(values.dtype, cast_type):
1622 return
1623 if values.dtype.kind == "f":
1624 # discarding a float's fractional part is not wraparound and stays
1625 # allowed (see test_read_fwf.py::test_dtype)
1626 values = np.trunc(values)
1627 if (np.asarray(casted) != values).any():
1628 raise ValueError(
1629 f"cannot safely convert passed user dtype of "
1630 f"{cast_type} for {values.dtype.name} dtyped data in "
1631 f"column {column}"
1632 )