Coverage for graphica/gui/workers.py: 95.4%

1"""データファイルの読み込み(TaskRunner で別スレッドで動かす関数)と、文字コード・区切り文字の判定。"""

2import csv

3import dataclasses

4import re

5import warnings

7import pandas as pd

9# 順に試す文字コード。utf-8-sig は BOM の有無どちらの UTF-8 も読める。latin-1 は必ず読める最後の手段。

10# UTF-16 は入れない(cp932 や latin-1 が例外を出さずに化けた結果を返す)。BOM で別に見分ける。

11CSV_ENCODING_FALLBACKS = ['utf-8-sig', 'cp932', 'latin-1']

13# 組み込みで読める拡張子。開くダイアログ、ドラッグ&ドロップ、フォルダ一括取り込み、プレビューがすべてここを見る。

14# 区切り文字付きテキストは CSV と同じく文字コードと区切りを判定する。.xlsx は openpyxl、.xls は xlrd。

15DELIMITED_TEXT_EXTENSIONS = ('.csv', '.txt')

16EXCEL_EXTENSIONS = ('.xlsx', '.xls')

17BUILTIN_DATA_FILE_EXTENSIONS = DELIMITED_TEXT_EXTENSIONS + EXCEL_EXTENSIONS

20def is_delimited_text_file(file_path):

21 return bool(file_path) and file_path.lower().endswith(DELIMITED_TEXT_EXTENSIONS)

24def is_excel_file(file_path):

25 return bool(file_path) and file_path.lower().endswith(EXCEL_EXTENSIONS)

28def excel_engine_for(file_path):

29 """openpyxl は .xls を読めないので、拡張子で xlrd と使い分ける。"""

30 return 'xlrd' if file_path.lower().endswith('.xls') else 'openpyxl'

33def pandas_separator(delimiter):

34 """(sep, engine)。空白は空白の連続とみなす(桁揃えの空白で空の列が大量にできないように)。正規表現は python エンジン。"""

35 if delimiter == ' ':

36 return r'\s+', 'python'

37 if len(delimiter) == 1: 37 ↛ 39line 37 didn't jump to line 39 because the condition on line 37 was always true

38 return delimiter, 'c'

39 return delimiter, 'python'

42def _detect_bom_encoding(file_path):

43 """BOM から文字コードを判定する。無ければ None。"""

44 with open(file_path, 'rb') as f:

45 head = f.read(4)

46 if head.startswith(b'\xff\xfe') or head.startswith(b'\xfe\xff'):

47 return 'utf-16'

48 if head.startswith(b'\xef\xbb\xbf'):

49 return 'utf-8-sig'

50 return None

53def _csv_encoding_candidates(file_path):

54 """BOM の判定を先頭に、CSV_ENCODING_FALLBACKS を並べた候補(読み込みと判定で順序を共有する)。"""

55 bom_encoding = _detect_bom_encoding(file_path)

56 candidates = [bom_encoding] if bom_encoding else []

57 candidates += [e for e in CSV_ENCODING_FALLBACKS if e not in candidates]

58 return candidates

61def detect_csv_encoding(file_path):

62 """テキストとしてデコードできる最初の文字コード。DataFrame にはしないので軽い。"""

63 last_error = None

64 for encoding in _csv_encoding_candidates(file_path): 64 ↛ 72line 64 didn't jump to line 72 because the loop on line 64 didn't complete

65 try:

66 with open(file_path, encoding=encoding) as f:

67 f.read()

68 return encoding

69 except UnicodeDecodeError as e:

70 last_error = e

71 continue

72 raise ValueError(

73 f"CSVファイルの文字コードを判定できませんでした "

74 f"(試行: {', '.join(CSV_ENCODING_FALLBACKS)})。詳細: {last_error}"

75 )

78def _sniff_delimiter_from_text(sample_text):

79 """カンマ・タブ・セミコロン・空白のどれかを推測する。判定できなければカンマ。"""

80 if not sample_text.strip():

81 return ','

82 try:

83 dialect = csv.Sniffer().sniff(sample_text, delimiters=',\t; ')

84 return dialect.delimiter

85 except csv.Error:

86 return ','

89def detect_csv_delimiter(file_path, encoding, sample_lines=50):

90 lines = []

91 try:

92 with open(file_path, encoding=encoding, errors='replace') as f:

93 for _ in range(sample_lines): 93 ↛ 100line 93 didn't jump to line 100

94 line = f.readline()

95 if not line:

96 break

97 lines.append(line)

98 except OSError:

99 return ','

100 return _sniff_delimiter_from_text(''.join(lines))

101

102

103def detect_clipboard_delimiter(text, sample_lines=50):

104 """クリップボードの区切り文字(Excel からのコピーはタブ、CSV の文字列ならカンマなど)。"""

105 sample = ''.join(text.splitlines(keepends=True)[:sample_lines])

106 return _sniff_delimiter_from_text(sample)

107

108

109# 測定装置の書き出し(JASCO の TXT など)は、数値の表の前後に測定条件の行が付く。そういうファイルでだけ数値の表を探す

110NUMERIC_TABLE_DELIMITERS = ('\t', ',', ';', ' ')

111MIN_NUMERIC_TABLE_ROWS = 3

112

113

114@dataclasses.dataclass

115class NumericTable:

116 first_line: int # 0 始まり、表の最初の数値の行

117 stop_line: int # 表の最後の行の次

118 delimiter: str

119 header: list | None # 表の直前の行が同じ列数の文字なら列名

120

121

122def _split_numeric_row(line, delimiter):

123 """2 列以上がすべて数値として読めれば、その列の文字列。読めなければ None。"""

124 text = line.strip()

125 fields = text.split() if delimiter == ' ' else [f.strip() for f in text.split(delimiter)]

126 if len(fields) < 2:

127 return None

128 try:

129 for field in fields:

130 float(field)

131 except ValueError:

132 return None

133 return fields

134

135

136def find_numeric_table(lines):

137 """列数のそろった数値の行が最も長く続くところ。MIN_NUMERIC_TABLE_ROWS 行に満たなければ None。"""

138 best = None

139 for delimiter in NUMERIC_TABLE_DELIMITERS:

140 start, columns = None, 0

141 for index, line in enumerate([*lines, '']):

142 fields = _split_numeric_row(line, delimiter)

143 if fields is not None and start is not None and len(fields) == columns:

144 continue

145 if start is not None and (best is None or index - start > best.stop_line - best.first_line):

146 best = NumericTable(start, index, delimiter, None)

147 start, columns = (index, len(fields)) if fields is not None else (None, 0)

148 if best is None or best.stop_line - best.first_line < MIN_NUMERIC_TABLE_ROWS:

149 return None

150 if best.first_line > 0: 150 ↛ 156line 150 didn't jump to line 156 because the condition on line 150 was always true

151 above = lines[best.first_line - 1].strip()

152 names = above.split() if best.delimiter == ' ' else [f.strip() for f in above.split(best.delimiter)]

153 column_count = len(_split_numeric_row(lines[best.first_line], best.delimiter))

154 if above and len(names) == column_count and _split_numeric_row(above, best.delimiter) is None:

155 best.header = names

156 return best

157

158

159def read_numeric_table(file_path, encoding):

160 """前後の説明の行を除いた数値の表を DataFrame にする。(DataFrame, NumericTable)、見つからなければ None。"""

161 import io

162

163 with open(file_path, encoding=encoding) as f:

164 lines = f.read().splitlines()

165 table = find_numeric_table(lines)

166 if table is None:

167 return None

168 sep, engine = pandas_separator(table.delimiter)

169 body = '\n'.join(line.strip() for line in lines[table.first_line:table.stop_line])

170 df = pd.read_csv(io.StringIO(body), sep=sep, engine=engine, header=None)

171 names = table.header or [f"列{i + 1}" for i in range(df.shape[1])]

172 # 同じ列名は read_csv と同じく .1, .2 を付けて分ける

173 seen = {}

174 unique = []

175 for name in names:

176 count = seen.get(name, 0)

177 unique.append(name if count == 0 else f"{name}.{count}")

178 seen[name] = count + 1

179 df.columns = unique

180 return df, table

181

182

183def has_numeric_column(df):

184 return any(pd.api.types.is_numeric_dtype(df[column]) for column in df.columns)

185

186

187# 年(4 桁の数字)を含む値だけを日付とみなす。"12:30" のような時刻だけの文字は、読んだ日の日付が補われてしまう

188_YEAR_PATTERN = re.compile(r"\d{4}")

189

190

191def parse_date_columns(df):

192 """文字の列のうち、空でない値がすべて日付として読める列を日付型にする(Excel の日付の列と同じ扱い)。"""

193 for column in df.columns:

194 series = df[column]

195 if series.dtype != object:

196 continue

197 text = series.dropna().astype(str).str.strip()

198 text = text[text != ""]

199 if text.empty or not text.map(lambda value: _YEAR_PATTERN.search(value) is not None).all():

200 continue

201 with warnings.catch_warnings():

202 # 書式を推測できない警告と、タイムゾーンが混ざった列の将来の仕様変更の警告(その列は下で日付にしない)

203 warnings.simplefilter("ignore", UserWarning)

204 warnings.simplefilter("ignore", FutureWarning)

205 try:

206 # 書式は値ごとに読む(先頭の値の書式に合わせると、時刻のある値と無い値が混ざった列を読めない)

207 parsed = pd.to_datetime(text, errors="coerce", format="mixed")

208 except (ValueError, TypeError):

209 continue

210 # タイムゾーンが混ざると日付型にならない(object のまま)

211 if parsed.isna().any() or not pd.api.types.is_datetime64_any_dtype(parsed):

212 continue

213 converted = pd.Series(pd.NaT, index=series.index, dtype=parsed.dtype)

214 converted.loc[parsed.index] = parsed

215 df[column] = converted

216 return df

217

218

219def read_data_file(file_path):

220 """データファイルを DataFrame にする。プラグインがその拡張子を登録していればそちらを使う。"""

221 ext = file_path.lower().split('.')[-1]

222

223 from graphica.core.plugin_api import get_plugin_api

224 from graphica.core.plugin_types import PluginExecutionError

225 api = get_plugin_api()

226 importer = api.get_importer_for_extension(ext) if api is not None else None

227 if importer is not None:

228 try:

229 result = importer.loader(file_path)

230 except Exception as e:

231 raise PluginExecutionError(importer.name, f"「{file_path}」の読み込みに失敗しました: {e}") from e

232 if not isinstance(result, pd.DataFrame):

233 raise PluginExecutionError(

234 importer.name,

235 "現在サポートされているのは単一のDataFrameを返すインポーターのみです"

236 "(複数シートを返す形式は未対応です)。"

237 )

238 return result

239

240 if is_delimited_text_file(file_path):

241 # 区切り文字もここで推測する(カンマで読むとタブ区切りが1列に潰れ、「2列以上必要」で弾かれる)

242 last_error = None

243 for encoding in _csv_encoding_candidates(file_path):

244 try:

245 sep, engine = pandas_separator(detect_csv_delimiter(file_path, encoding))

246 df = pd.read_csv(file_path, header=0, encoding=encoding, sep=sep, engine=engine)

247 except UnicodeDecodeError as e:

248 last_error = e

249 continue

250 except pd.errors.ParserError as e:

251 try:

252 found = read_numeric_table(file_path, encoding)

253 except UnicodeDecodeError:

254 found = None

255 if found is not None:

256 return parse_date_columns(found[0])

257 last_error = e

258 continue

259 # 普通に読めて数値の列があるファイルは今までどおり。どの列も数値にならないときだけ、前後の説明の行を疑う

260 if not has_numeric_column(df):

261 found = read_numeric_table(file_path, encoding)

262 if found is not None:

263 return parse_date_columns(found[0])

264 return parse_date_columns(df)

265 raise ValueError(

266 f"テキストファイルの文字コードを判定できませんでした "

267 f"(試行: {', '.join(CSV_ENCODING_FALLBACKS)})。詳細: {last_error}"

268 )

269 elif is_excel_file(file_path):

270 return pd.read_excel(file_path, engine=excel_engine_for(file_path))

271 else:

272 raise ValueError(f"未対応のファイル形式です: {ext}")

273

274

275def load_data_file_task(file_path, report_progress=None, is_cancelled=None):

276 """TaskRunner 用。読み込みは中断できないので、report_progress と is_cancelled は受け取るだけ。"""

277 df = read_data_file(file_path)

278 if len(df.columns) < 2:

279 raise ValueError("データには少なくとも2列必要です。")

280 return df