Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pandas/io/excel/_odfreader.py: 19%

1from __future__ import annotations 

2 

3from typing import ( 

4 TYPE_CHECKING, 

5 cast, 

6) 

7 

8import numpy as np 

9 

10from pandas._typing import ( 

11 FilePath, 

12 ReadBuffer, 

13 Scalar, 

14 StorageOptions, 

15) 

16from pandas.compat._optional import import_optional_dependency 

17from pandas.util._decorators import doc 

18 

19import pandas as pd 

20from pandas.core.shared_docs import _shared_docs 

21 

22from pandas.io.excel._base import BaseExcelReader 

23 

24if TYPE_CHECKING: 

25 from odf.opendocument import OpenDocument 

26 

27 from pandas._libs.tslibs.nattype import NaTType 

28 

29 

30@doc(storage_options=_shared_docs["storage_options"]) 

31class ODFReader(BaseExcelReader["OpenDocument"]): 

32 def __init__( 

33 self, 

34 filepath_or_buffer: FilePath | ReadBuffer[bytes], 

35 storage_options: StorageOptions | None = None, 

36 engine_kwargs: dict | None = None, 

37 ) -> None: 

38 """ 

39 Read tables out of OpenDocument formatted files. 

40 

41 Parameters 

42 ---------- 

43 filepath_or_buffer : str, path to be parsed or 

44 an open readable stream. 

45 {storage_options} 

46 engine_kwargs : dict, optional 

47 Arbitrary keyword arguments passed to excel engine. 

48 """ 

49 import_optional_dependency("odf") 

50 super().__init__( 

51 filepath_or_buffer, 

52 storage_options=storage_options, 

53 engine_kwargs=engine_kwargs, 

54 ) 

55 

56 @property 

57 def _workbook_class(self) -> type[OpenDocument]: 

58 from odf.opendocument import OpenDocument 

59 

60 return OpenDocument 

61 

62 def load_workbook( 

63 self, filepath_or_buffer: FilePath | ReadBuffer[bytes], engine_kwargs 

64 ) -> OpenDocument: 

65 from odf.opendocument import load 

66 

67 return load(filepath_or_buffer, **engine_kwargs) 

68 

69 @property 

70 def empty_value(self) -> str: 

71 """Property for compat with other readers.""" 

72 return "" 

73 

74 @property 

75 def sheet_names(self) -> list[str]: 

76 """Return a list of sheet names present in the document""" 

77 from odf.table import Table 

78 

79 tables = self.book.getElementsByType(Table) 

80 return [t.getAttribute("name") for t in tables] 

81 

82 def get_sheet_by_index(self, index: int): 

83 from odf.table import Table 

84 

85 self.raise_if_bad_sheet_by_index(index) 

86 tables = self.book.getElementsByType(Table) 

87 return tables[index] 

88 

89 def get_sheet_by_name(self, name: str): 

90 from odf.table import Table 

91 

92 self.raise_if_bad_sheet_by_name(name) 

93 tables = self.book.getElementsByType(Table) 

94 

95 for table in tables: 

96 if table.getAttribute("name") == name: 

97 return table 

98 

99 self.close() 

100 raise ValueError(f"sheet {name} not found") 

101 

102 def get_sheet_data( 

103 self, sheet, file_rows_needed: int | None = None 

104 ) -> list[list[Scalar | NaTType]]: 

105 """ 

106 Parse an ODF Table into a list of lists 

107 """ 

108 from odf.table import ( 

109 CoveredTableCell, 

110 TableCell, 

111 TableRow, 

112 ) 

113 

114 covered_cell_name = CoveredTableCell().qname 

115 table_cell_name = TableCell().qname 

116 cell_names = {covered_cell_name, table_cell_name} 

117 

118 sheet_rows = sheet.getElementsByType(TableRow) 

119 empty_rows = 0 

120 max_row_len = 0 

121 

122 table: list[list[Scalar | NaTType]] = [] 

123 

124 for sheet_row in sheet_rows: 

125 empty_cells = 0 

126 table_row: list[Scalar | NaTType] = [] 

127 

128 for sheet_cell in sheet_row.childNodes: 

129 if hasattr(sheet_cell, "qname") and sheet_cell.qname in cell_names: 

130 if sheet_cell.qname == table_cell_name: 

131 value = self._get_cell_value(sheet_cell) 

132 else: 

133 value = self.empty_value 

134 

135 column_repeat = self._get_column_repeat(sheet_cell) 

136 

137 # Queue up empty values, writing only if content succeeds them 

138 if value == self.empty_value: 

139 empty_cells += column_repeat 

140 else: 

141 table_row.extend([self.empty_value] * empty_cells) 

142 empty_cells = 0 

143 table_row.extend([value] * column_repeat) 

144 

145 if max_row_len < len(table_row): 

146 max_row_len = len(table_row) 

147 

148 row_repeat = self._get_row_repeat(sheet_row) 

149 if len(table_row) == 0: 

150 empty_rows += row_repeat 

151 else: 

152 # add blank rows to our table 

153 table.extend([[self.empty_value]] * empty_rows) 

154 empty_rows = 0 

155 table.extend(table_row for _ in range(row_repeat)) 

156 if file_rows_needed is not None and len(table) >= file_rows_needed: 

157 break 

158 

159 # Make our table square 

160 for row in table: 

161 if len(row) < max_row_len: 

162 row.extend([self.empty_value] * (max_row_len - len(row))) 

163 

164 return table 

165 

166 def _get_row_repeat(self, row) -> int: 

167 """ 

168 Return number of times this row was repeated 

169 Repeating an empty row appeared to be a common way 

170 of representing sparse rows in the table. 

171 """ 

172 from odf.namespaces import TABLENS 

173 

174 return int(row.attributes.get((TABLENS, "number-rows-repeated"), 1)) 

175 

176 def _get_column_repeat(self, cell) -> int: 

177 from odf.namespaces import TABLENS 

178 

179 return int(cell.attributes.get((TABLENS, "number-columns-repeated"), 1)) 

180 

181 def _get_cell_value(self, cell) -> Scalar | NaTType: 

182 from odf.namespaces import OFFICENS 

183 

184 if str(cell) == "#N/A": 

185 return np.nan 

186 

187 cell_type = cell.attributes.get((OFFICENS, "value-type")) 

188 if cell_type == "boolean": 

189 if str(cell) == "TRUE": 

190 return True 

191 return False 

192 if cell_type is None: 

193 return self.empty_value 

194 elif cell_type == "float": 

195 # GH5394 

196 cell_value = float(cell.attributes.get((OFFICENS, "value"))) 

197 val = int(cell_value) 

198 if val == cell_value: 

199 return val 

200 return cell_value 

201 elif cell_type == "percentage": 

202 cell_value = cell.attributes.get((OFFICENS, "value")) 

203 return float(cell_value) 

204 elif cell_type == "string": 

205 return self._get_cell_string_value(cell) 

206 elif cell_type == "currency": 

207 cell_value = cell.attributes.get((OFFICENS, "value")) 

208 return float(cell_value) 

209 elif cell_type == "date": 

210 cell_value = cell.attributes.get((OFFICENS, "date-value")) 

211 return pd.Timestamp(cell_value) 

212 elif cell_type == "time": 

213 stamp = pd.Timestamp(str(cell)) 

214 # cast needed here because Scalar doesn't include datetime.time 

215 return cast(Scalar, stamp.time()) 

216 else: 

217 self.close() 

218 raise ValueError(f"Unrecognized type {cell_type}") 

219 

220 def _get_cell_string_value(self, cell) -> str: 

221 """ 

222 Find and decode OpenDocument text:s tags that represent 

223 a run length encoded sequence of space characters. 

224 """ 

225 from odf.element import Element 

226 from odf.namespaces import TEXTNS 

227 from odf.office import Annotation 

228 from odf.text import S 

229 

230 office_annotation = Annotation().qname 

231 text_s = S().qname 

232 

233 value = [] 

234 

235 for fragment in cell.childNodes: 

236 if isinstance(fragment, Element): 

237 if fragment.qname == text_s: 

238 spaces = int(fragment.attributes.get((TEXTNS, "c"), 1)) 

239 value.append(" " * spaces) 

240 elif fragment.qname == office_annotation: 

241 continue 

242 else: 

243 # recursive impl needed in case of nested fragments 

244 # with multiple spaces 

245 # https://github.com/pandas-dev/pandas/pull/36175#discussion_r484639704 

246 value.append(self._get_cell_string_value(fragment)) 

247 else: 

248 value.append(str(fragment).strip("\n")) 

249 return "".join(value)