# This file is part of Sympathy for Data.
# Copyright (c) 2013, Combine Control Systems AB
#
# SYMPATHY FOR DATA COMMERCIAL LICENSE
# You should have received a link to the License with Sympathy for Data.
import contextlib
import datetime
import os
import numpy as np
from sympathy.api import node as synode
from sympathy.api import importers
from sympathy.api import qt2 as qt_compat
from sympathy.api import table
from sympathy.api.exceptions import sywarn, SyDataError
from sylib.xl_utils import is_xls, is_xlsx, get_xlsx_sheetnames, load_workbook
from sylib.table_sources import TableSourceModel, PreviewWorker
from sylib.table_importer_gui import (
TableImportWidget, TableImportController)
QtCore = qt_compat.import_module('QtCore')
QtGui = qt_compat.import_module('QtGui')
QtWidgets = qt_compat.import_module('QtWidgets')
@contextlib.contextmanager
def open_workbook(filename):
"""Open workbook, on demand - if possible."""
try:
wb = load_workbook(filename, data_only=True)
except Exception as exc:
raise SyDataError(
f"Could not open (possibly non-standard) Excel file: {filename}.\n"
f"Open and Save using Excel may fix such problems."
) from exc
else:
try:
yield wb
finally:
wb.close()
_read_to_end = 'Read to the end of file'
_read_selected = 'Read specified number of rows'
_read_from_end = 'Read to specified number of rows from the end'
_read_options = [_read_to_end, _read_selected, _read_from_end]
class TableSourceModelXLS(TableSourceModel):
"""Model layer between GUI and xlsx importer."""
get_preview = qt_compat.Signal(
str, bool, bool, int, int, bool, int, int, int)
def __init__(self, parameters, fq_infilename, mode, valid, multi):
super().__init__(
parameters, fq_infilename, mode)
self.data_table = None
self._multi = multi
self._valid = valid
self._init_model_specific_parameters()
self._init_preview_worker()
def _init_model_specific_parameters(self):
"""Init special parameters for xlsx importer."""
self.headers = self._parameters['headers']
self.header_row = self._parameters['header_row']
self.units = self._parameters['units']
self.unit_row = self._parameters['unit_row']
self.descriptions = self._parameters['descriptions']
self.description_row = self._parameters['description_row']
self.transposed = self._parameters['transposed']
self.worksheet_name = self._parameters['worksheet_name']
self.import_all = self._parameters['import_all']
self.import_first = self._parameters['import_first']
def _init_preview_worker(self):
"""Collect preview data from xlsx file."""
if self._valid:
sheet_names = get_xlsx_sheetnames(self._fq_infilename)
self.worksheet_name.list = sheet_names
self._importer = ImporterXLS(self._fq_infilename, False, True)
self._preview_thread = QtCore.QThread()
self._preview_worker = PreviewWorker(self._importer.import_xls)
self._preview_worker.moveToThread(self._preview_thread)
self._preview_thread.finished.connect(self._preview_worker.deleteLater)
self.get_preview.connect(self._preview_worker.create_preview_table)
self._preview_worker.preview_ready.connect(self.set_preview_table)
self._preview_worker.preview_failed.connect(self.set_preview_failed)
self._preview_thread.start()
self.collect_preview_values()
@qt_compat.Slot()
def collect_preview_values(self):
"""Collect preview data from xlsx file."""
data_start_row = self.data_offset.value - 1
data_end_row = data_start_row + self.no_preview_rows.value
sheet_name = self.worksheet_name.selected
import_all = self.import_all.value
import_first = self.import_first.value
transposed = self.transposed.value
if self.headers.value:
headers_row = self.header_row.value - 1
else:
headers_row = -1
if self.units.value:
units_row = self.unit_row.value - 1
else:
units_row = -1
if self.descriptions.value:
descriptions_row = self.description_row.value - 1
else:
descriptions_row = -1
if self._valid:
self.get_preview.emit(
sheet_name, import_first, import_all, data_start_row,
data_end_row,
transposed, headers_row, units_row, descriptions_row)
else:
self.data_table = table.File()
self.update_table.emit()
@qt_compat.Slot(table.File)
def set_preview_table(self, data_table):
self.data_table = data_table
self.update_table.emit()
def cleanup(self):
self._preview_thread.quit()
self._preview_thread.wait()
class ImporterXLS:
"""Class for importat of data in the xlsx format."""
row_limit = 1048576 # 2 ** 20
column_limit = 16384 # 2 ** 14
def __init__(self, fq_infilename, multi, preview=False):
self._warnings = set()
self._multi = multi
self._fq_infilename = fq_infilename
self._preview = preview
def raise_mixed_types(x):
class MixedTypesError(Exception):
pass
raise MixedTypesError()
def type_converters(allowed_types):
"""
Takes a dict with convertable types and returns a new dict with all
types represented. Unallowed types will be filled in with
raise_mixed_types, except for NoneType which is always allowed and
results in a None value.
"""
defaults = {
str: raise_mixed_types,
bool: raise_mixed_types,
int: raise_mixed_types,
float: raise_mixed_types,
datetime.datetime: raise_mixed_types,
datetime.time: raise_mixed_types,
datetime.timedelta: raise_mixed_types,
type(None): lambda x: None,
}
return {**defaults, **allowed_types}
def float_to_float(x):
# Excel numbers never have more than 15 significant digits
# so anything beyond that is just an effect of conversion
# from excel format to python floats.
return round(x, 15)
def date_to_date(x):
if x < datetime.datetime(1900, 3, 1):
self._warnings.add(
"Ignoring any datetimes before 1900-03-01. All dates "
"before 1900-03-01 are ambiguous because of a bug in "
"Excel.")
return None
return x
def date_to_str(x):
date = date_to_date(x)
if date is None:
return None
else:
return str(date)
def time_to_timedelta(x):
return datetime.timedelta(
hours=x.hour,
minutes=x.minute,
seconds=x.second,
microseconds=x.microsecond)
def no_conversion(x):
return x
self._xl_cell_to_str = type_converters({
str: str,
bool: lambda x: str(bool(x)),
int: str,
float: lambda x: str(float_to_float(x)),
datetime.datetime: date_to_str,
datetime.time: lambda x: str(time_to_timedelta(x)),
type(None): lambda x: '',
})
self._xl_cell_to_bool = type_converters({
bool: bool,
})
self._xl_cell_to_int = type_converters({
bool: int,
int: int,
})
self._xl_cell_to_float = type_converters({
bool: float,
int: float,
float: float_to_float,
})
self._xl_cell_to_date = type_converters({
datetime.datetime: date_to_date,
})
self._xl_cell_to_time = type_converters({
datetime.time: time_to_timedelta,
datetime.timedelta: no_conversion,
})
self._xl_cell_to_timedelta = self._xl_cell_to_time
def get_row_to_array(
self, ws, row, data_start_column, data_end_column):
"""Read a row in the xls file and returns it as an array."""
return self._to_array(ws, 'row', row,
data_start=data_start_column,
data_end=data_end_column)
def get_column_to_array(
self, ws, column, data_start_row, data_end_row):
"""Read a column in the xls file and returns it as an array."""
return self._to_array(ws, 'column', column,
data_start=data_start_row,
data_end=data_end_row)
def _to_array(self, ws, alignment, i, data_start, data_end):
"""Read a row/column in the xls file and returns it as an array."""
i += 1
data_start += 1
if alignment == 'row':
iter_values = ws.iter_rows
min_row = i
max_row = i
min_col = data_start
max_col = data_end
elif alignment == 'column':
iter_values = ws.iter_cols
min_col = i
max_col = i
min_row = data_start
max_row = data_end
else:
raise ValueError("Invalid alignment: {}. Should be either "
"'row' or 'column'.".format(alignment))
min_row = max(min_row, 1)
min_col = max(min_col, 1)
values_gen = iter_values(
min_col=min_col,
max_col=max_col,
min_row=min_row,
max_row=max_row,
values_only=True,
)
if values_gen == tuple():
# openpyxl documentation states that iter_rows/iter_cols can return
# an empty tuple instead of a generator if there are no cells in
# the worksheet. This doesn't seem to actually be the case in
# openpyxl==3.0.2, but it does happen for Worksheet.columns so this
# is just to be on the safe side.
values = []
else:
values = list(next(values_gen))
types = {type(v) for v in values} - {type(None)}
if len(types) == 0:
# Only emtpy cells in this row/column => convert to str
target_type = str
elif len(types) == 1:
# Only a single type => convert to that type
target_type = list(types)[0]
else:
if not types - {int, bool}:
# Only integer types (including bool) => convert all to int
target_type = int
elif not types - {float, int, bool}:
# Only numeric types (including bool) => convert all to float
target_type = float
else:
# A combination of str/datetime/numeric => convert all to str
target_type = str
# Convert all values to target type
converter = {
str: self._xl_cell_to_str,
int: self._xl_cell_to_int,
float: self._xl_cell_to_float,
datetime.datetime: self._xl_cell_to_date,
datetime.time: self._xl_cell_to_time,
datetime.timedelta: self._xl_cell_to_timedelta,
bool: self._xl_cell_to_bool,
}[target_type]
array_values = [converter[type(v)](v) for v in values]
mask = [v is None for v in array_values]
if any(mask):
# Deal with empty cells in row/column
if target_type == datetime.datetime:
empty_value = datetime.datetime(1900, 1, 1)
else:
empty_value = target_type()
array_values = [empty_value if v is None else v
for v in array_values]
return np.ma.MaskedArray(array_values, mask=mask)
else:
return np.array(array_values)
def import_xls(self, out_data, sheet_name,
import_first,
import_all,
data_start, data_end,
transposed, headers_row_offset=-1,
units_row_offset=-1,
descriptions_row_offset=-1):
"""Adminstration method for import of data in xls format."""
def sheet_by_name(sheet_name):
ws = None
try:
ws = wb[sheet_name]
except KeyError:
if import_first:
ws = wb.worksheets[0]
else:
raise
return ws
def test_cell(ws, row, column):
try:
ws.cell(row, column)
return True
except Exception:
return False
def import_sheet(ws, out_table):
max_row = ws.max_row
max_column = ws.max_column
if max_row > self.row_limit or max_column > self.column_limit:
# First, check if getting the last cell would fail.
row_ok = test_cell(ws, max_row, 1)
column_ok = test_cell(ws, 1, max_column)
if not row_ok:
max_row = min(self.row_limit, max_row)
if not column_ok:
max_column = min(self.column_limit, max_column)
if not self._preview:
sywarn(
f'Data range exceeds limits defined by the '
f'format: {self.row_limit} rows x {self.column_limit} '
f'columns. Cells out of range may be ignored.')
return _import_sheet(ws, out_table, max_row, max_column)
def _import_sheet(ws, out_table, max_row, max_column):
def get_row_to_array(ws, row, data_start_column=0,
data_end_column=None):
if data_end_column is None:
data_end_column = max_column
data_end_column = min(data_end_column + 1, max_column)
return self.get_row_to_array(
ws, row, data_start_column=data_start_column,
data_end_column=data_end_column)
def get_column_to_array(ws, col, data_start_row=0,
data_end_row=None):
if data_end_row is None:
data_end_row = max_row
data_end_row = min(data_end_row + 1, max_row)
return self.get_column_to_array(
ws, col, data_start_row=data_start_row,
data_end_row=data_end_row)
out_table.set_name(ws.title)
if transposed:
nr_rows = max_column
nr_cols = max_row
data_collector = get_row_to_array
header_collector = get_column_to_array
else:
nr_rows = max_row
nr_cols = max_column
data_collector = get_column_to_array
header_collector = get_row_to_array
if data_end < 0:
data_end_ = nr_rows + data_end
elif data_end == 0:
data_end_ = nr_rows
else:
data_end_ = data_start + data_end
if headers_row_offset >= 0:
headers = header_collector(ws, headers_row_offset)
if isinstance(headers, np.ma.MaskedArray):
headers = headers.filled('')
else:
headers = ['f{0}'.format(index) for index in range(nr_cols)]
if units_row_offset >= 0:
units = [str(x) for x in
header_collector(ws, units_row_offset)]
else:
units = None
if descriptions_row_offset >= 0:
descs = [str(x) for x in
header_collector(ws, descriptions_row_offset)]
else:
descs = None
for col in range(nr_cols):
if not headers[col]:
continue
attr = {}
out_table.set_column_from_array(
headers[col], data_collector(
ws, col, data_start, data_end_))
if units:
attr['unit'] = units[col]
if descs:
attr['description'] = descs[col]
out_table.set_column_attributes(headers[col], attr)
if not os.path.exists(self._fq_infilename):
raise SyDataError(
"File does not exist: {}".format(self._fq_infilename))
elif is_xls(self._fq_infilename):
raise SyDataError(
"Support for legacy Excel file format (XLS) has been removed "
"from Sympathy 2.0.0. Please use Excel or another external "
"tool to convert the file to the newer Excel file format "
"(XLSX).")
with open_workbook(self._fq_infilename) as wb:
if self._multi:
if import_all:
for ws in wb.worksheets:
out_table = out_data.create()
import_sheet(ws, out_table)
out_data.append(out_table)
else:
out_table = out_data.create()
import_sheet(sheet_by_name(sheet_name), out_table)
out_data.append(out_table)
else:
out_table = out_data
import_sheet(sheet_by_name(sheet_name), out_table)
for warning in self._warnings:
sywarn(warning)
class TableImportControllerXLS(TableImportController):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._table_source.transpose_state_changed.connect(
self._model.collect_preview_values)
self._table_source.worksheet_changed[int].connect(
self._model.collect_preview_values)
class TableImportWidgetXLS(TableImportWidget):
MODE = 'XLS'
def __init__(self, parameters, fq_infilename, valid=True, multi=False):
self._multi = multi
self.model = TableSourceModelXLS(
parameters, fq_infilename, self.MODE, valid, multi=multi)
super().__init__(
parameters, fq_infilename, self.MODE, valid)
def _collect_table_source_widget(self, model):
return TableSourceWidgetXLS(model, self._multi)
def _collect_controller(self, **kwargs):
return TableImportControllerXLS(**kwargs)
class TableSourceWidgetXLS(QtWidgets.QWidget):
worksheet_changed = qt_compat.Signal(int)
transpose_state_changed = qt_compat.Signal(int)
def __init__(self, model, multi=False, parent=None):
super().__init__(parent)
self._multi = multi
self._init_gui(model)
def _init_gui(self, model):
layout = QtWidgets.QVBoxLayout()
# Worksheet Combobox
self._worksheet_combobox = model.worksheet_name.gui()
self._import_first_checkbox = model.import_first.gui()
self._import_all_checkbox = model.import_all.gui()
if self._multi:
self._worksheet_combobox.setDisabled(model.import_all.value)
self._import_first_checkbox.setDisabled(model.import_all.value)
layout.addWidget(self._import_all_checkbox)
self._worksheet_combobox.setFixedWidth(200)
layout.addWidget(self._worksheet_combobox)
layout.addWidget(self._import_first_checkbox)
self._worksheet_combobox.editor().currentIndexChanged[int].connect(
self.worksheet_changed)
self._import_all_checkbox.valueChanged.connect(
self._worksheet_combobox.setDisabled)
self._import_all_checkbox.valueChanged.connect(
self._import_first_checkbox.setDisabled)
# Transpose checkbox
self._transposed_checkbox = model.transposed.gui()
layout.addWidget(self._transposed_checkbox)
self._transposed_checkbox.stateChanged[int].connect(
self.transpose_state_changed)
self.setLayout(layout)
[docs]
class DataImportXLS(importers.TableDataImporterBase):
"""Importer for Excel files."""
IMPORTER_NAME = 'XLS'
DISPLAY_NAME = 'XLSX'
PARAMETER_VIEW = TableImportWidgetXLS
def __init__(self, fq_infilename, parameters):
super().__init__(fq_infilename, parameters)
if parameters is not None:
self._init_parameters()
def cardinalities(self):
return [self.one_to_many, self.one_to_one]
def _init_parameters(self):
parameter_root = self._parameters
nbr_of_rows = 99999
nbr_of_end_rows = 9999999
try:
parameter_root['import_all']
except KeyError:
parameter_root.set_boolean(
'import_all', label='Import all worksheets',
description=(
'Ignore worksheet selection and import all worksheets. '
'This requires an output capable of storing multiple '
'output elements.'),
value=False)
try:
parameter_root['worksheet_name']
except KeyError:
parameter_root.set_list(
'worksheet_name', label='Select worksheet',
description='The worksheet to import from.',
editor=synode.editors.combo_editor())
try:
parameter_root['import_first']
except KeyError:
parameter_root.set_boolean(
'import_first',
label='Use first worksheet if the selected is missing',
value=True,
description=(
'This option is used if the input data does not contain '
'the selected worksheet.'
'\n\n'
'Enabled: the first available worksheet will be '
'imported instead of the selected.'
'\n'
'Disabled: the import will fail and inform that '
'the selected worksheet is missing.'))
# Init header start row spinbox
try:
parameter_root['header_row']
except KeyError:
parameter_root.set_integer(
'header_row', value=1,
description='The row where the headers are located.',
editor=synode.editors.bounded_spinbox_editor(
1, nbr_of_rows, 1))
# Init unit row spinbox
try:
parameter_root['unit_row']
except KeyError:
parameter_root.set_integer(
'unit_row', value=1,
description='The row where the units are located.',
editor=synode.editors.bounded_spinbox_editor(
1, nbr_of_rows, 1))
# Init description row spinbox
try:
parameter_root['description_row']
except KeyError:
parameter_root.set_integer(
'description_row', value=1,
description='The row where the descriptions are located.',
editor=synode.editors.bounded_spinbox_editor(
1, nbr_of_rows, 1))
# Init data start row spinbox
try:
parameter_root['data_start_row']
except KeyError:
parameter_root.set_integer(
'data_start_row', value=2,
description='The first row where data is stored.',
editor=synode.editors.bounded_spinbox_editor(
1, nbr_of_rows, 1))
# Init data end row spinbox
try:
parameter_root['data_end_row']
except KeyError:
parameter_root.set_integer(
'data_end_row', value=0, description='The last data row.',
editor=synode.editors.bounded_spinbox_editor(
0, nbr_of_end_rows, 1))
# Init headers checkbox
try:
parameter_root['headers']
except KeyError:
parameter_root.set_boolean(
'headers', value=True,
description='File has headers.')
# Init units checkbox
try:
parameter_root['units']
except KeyError:
parameter_root.set_boolean(
'units', value=False,
description='File has headers.')
# Init descriptions checkbox
try:
parameter_root['descriptions']
except KeyError:
parameter_root.set_boolean(
'descriptions', value=False,
description='File has headers.')
# Init transposed checkbox
try:
parameter_root['transposed']
except KeyError:
parameter_root.set_boolean(
'transposed', value=False, label='Transpose input',
description='Transpose the data.')
try:
parameter_root['end_of_file']
except KeyError:
parameter_root.set_boolean(
'end_of_file', value=True,
description='Select all rows to the end of the file.')
try:
parameter_root['read_selection']
except KeyError:
parameter_root.set_list(
'read_selection', value=[0],
list=_read_options,
editor=synode.editors.combo_editor())
# Move value of old parameter to new the format.
if not parameter_root['end_of_file'].value:
parameter_root['read_selection'].value_names = [_read_from_end]
try:
parameter_root['preview_start_row']
except KeyError:
parameter_root.set_integer(
'preview_start_row', value=1, label='Preview start row',
description='The first row where data will review from.',
editor=synode.editors.bounded_spinbox_editor(1, 200, 1))
try:
parameter_root['no_preview_rows']
except KeyError:
parameter_root.set_integer(
'no_preview_rows', value=20, label='Number of preview rows',
description='The number of preview rows to show.',
editor=synode.editors.bounded_spinbox_editor(1, 200, 1))
def name(self):
return self.IMPORTER_NAME
def valid_for_file(self):
if self._fq_infilename is None or not os.path.isfile(
self._fq_infilename):
return False
return is_xlsx(self._fq_infilename)
def parameter_view(self, parameters):
valid_for_file = self.valid_for_file()
return TableImportWidgetXLS(
parameters, self._fq_infilename, valid_for_file,
self.cardinality() == self.one_to_many)
def import_data(self, out_datafile, parameters=None, progress=None):
parameter_root = parameters
headers_bool = parameter_root['headers'].value
headers_row_offset = parameter_root['header_row'].value - 1
units_bool = parameter_root['units'].value
units_row_offset = parameter_root['unit_row'].value - 1
descs_bool = parameter_root['descriptions'].value
descs_row_offset = parameter_root['description_row'].value - 1
data_start_row = parameter_root['data_start_row'].value - 1
read_selection = parameter_root['read_selection'].selected
transposed = parameter_root['transposed'].value
sheet_name = parameter_root['worksheet_name'].selected
import_all = parameter_root['import_all'].value
import_first = parameter_root['import_first'].value
if not headers_bool:
headers_row_offset = -1
if not units_bool:
units_row_offset = -1
if not descs_bool:
descs_row_offset = -1
if read_selection == _read_to_end:
data_end_row = 0
elif read_selection == _read_selected:
data_end_row = parameter_root['data_end_row'].value
elif read_selection == _read_from_end:
data_end_row = - parameter_root['data_end_row'].value
else:
raise ValueError('Unknown Read Selection.')
importer = ImporterXLS(
self._fq_infilename, self.cardinality() == self.one_to_many)
importer.import_xls(out_datafile, sheet_name,
import_first,
import_all,
data_start_row,
data_end_row, transposed,
headers_row_offset=headers_row_offset,
units_row_offset=units_row_offset,
descriptions_row_offset=descs_row_offset)
