1# pyright: reportMissingImports=false
2from __future__ import annotations
3
4from typing import (
5 TYPE_CHECKING,
6 Any,
7)
8import warnings
9
10from pandas.compat._optional import import_optional_dependency
11from pandas.errors import Pandas4Warning
12from pandas.util._exceptions import find_stack_level
13
14from pandas.io.excel._base import BaseExcelReader
15
16if TYPE_CHECKING:
17 from pyxlsb import (
18 Workbook,
19 Worksheet,
20 )
21
22 from pandas._typing import (
23 FilePath,
24 ReadBuffer,
25 Scalar,
26 StorageOptions,
27 )
28
29
30class PyxlsbReader(BaseExcelReader["Workbook"]):
31 def __init__(
32 self,
33 filepath_or_buffer: FilePath | ReadBuffer[bytes],
34 storage_options: StorageOptions | None = None,
35 engine_kwargs: dict[str, Any] | None = None,
36 ) -> None:
37 """
38 Reader using pyxlsb engine.
39
40 Parameters
41 ----------
42 filepath_or_buffer : str, path object, or Workbook
43 Object to be parsed.
44 storage_options : dict, optional
45 Extra options that make sense for a particular storage connection, e.g.
46 host, port, username, password, etc. For HTTP(S) URLs the key-value pairs
47 are forwarded to ``urllib.request.Request`` as header options. For other
48 URLs (e.g. starting with "s3://", and "gcs://") the key-value pairs are
49 forwarded to ``fsspec.open``. Please see ``fsspec`` and ``urllib`` for more
50 details, and for more examples on storage options refer `here
51 <https://pandas.pydata.org/docs/user_guide/io.html?
52 highlight=storage_options#reading-writing-remote-files>`_.
53 engine_kwargs : dict, optional
54 Arbitrary keyword arguments passed to excel engine.
55 """
56 import_optional_dependency("pyxlsb")
57 warnings.warn(
58 "The pyxlsb engine is deprecated and will be removed in a future version. "
59 "Use the calamine engine instead, by setting engine='calamine'.",
60 Pandas4Warning,
61 stacklevel=find_stack_level(),
62 )
63 # This will call load_workbook on the filepath or buffer
64 # And set the result to the book-attribute
65 super().__init__(
66 filepath_or_buffer,
67 storage_options=storage_options,
68 engine_kwargs=engine_kwargs,
69 )
70
71 @property
72 def _workbook_class(self) -> type[Workbook]:
73 from pyxlsb import Workbook
74
75 return Workbook
76
77 def load_workbook(
78 self,
79 filepath_or_buffer: FilePath | ReadBuffer[bytes],
80 engine_kwargs: dict[str, Any],
81 ) -> Workbook:
82 from pyxlsb import open_workbook
83
84 # TODO: hack in buffer capability
85 # This might need some modifications to the Pyxlsb library
86 # Actual work for opening it is in xlsbpackage.py, line 20-ish
87
88 return open_workbook(filepath_or_buffer, **engine_kwargs)
89
90 @property
91 def sheet_names(self) -> list[str]:
92 return self.book.sheets
93
94 def get_sheet_by_name(self, name: str) -> Worksheet:
95 self.raise_if_bad_sheet_by_name(name)
96 return self.book.get_sheet(name)
97
98 def get_sheet_by_index(self, index: int) -> Worksheet:
99 self.raise_if_bad_sheet_by_index(index)
100 # pyxlsb sheets are indexed from 1 onwards
101 # There's a fix for this in the source, but the pypi package doesn't have it
102 return self.book.get_sheet(index + 1)
103
104 def _convert_cell(self, cell: Any) -> Scalar:
105 # TODO: there is no way to distinguish between floats and datetimes in pyxlsb
106 # This means that there is no way to read datetime types from an xlsb file yet
107 if cell.v is None:
108 return "" # Prevents non-named columns from not showing up as Unnamed: i
109 if isinstance(cell.v, float):
110 val = int(cell.v)
111 if val == cell.v:
112 return val
113 else:
114 return float(cell.v)
115
116 return cell.v
117
118 def get_sheet_data(
119 self,
120 sheet: Worksheet,
121 file_rows_needed: int | None = None,
122 ) -> list[list[Scalar]]:
123 data: list[list[Scalar]] = []
124 previous_row_number = -1
125 # When sparse=True the rows can have different lengths and empty rows are
126 # not returned. The cells are namedtuples of row, col, value (r, c, v).
127 for row in sheet.rows(sparse=True):
128 row_number = row[0].r
129 converted_row = [self._convert_cell(cell) for cell in row]
130 while converted_row and converted_row[-1] == "":
131 # trim trailing empty elements
132 converted_row.pop()
133 if converted_row:
134 data.extend([[]] * (row_number - previous_row_number - 1))
135 data.append(converted_row)
136 previous_row_number = row_number
137 if file_rows_needed is not None and len(data) >= file_rows_needed:
138 break
139 if data:
140 # extend rows to max_width
141 max_width = max(len(data_row) for data_row in data)
142 if min(len(data_row) for data_row in data) < max_width:
143 empty_cell: list[Scalar] = [""]
144 data = [
145 data_row + (max_width - len(data_row)) * empty_cell
146 for data_row in data
147 ]
148 return data