"""Register an adapter to display a type

assert '.' in target_type

top_module_name, type_name = target_type.split('.', maxsplit=1)

module_adapters = REGISTERED_ADAPTERS_USING_STRINGS.setdefault(top_module_name, {})

if type_name in module_adapters:

logger.warning(f"Replacing adapter for {target_type}")

module_adapters[type_name] = adapter_creator

# container = REGISTERED_ADAPTERS_USING_STRINGS

# parts = target_type.split('.')

# for i, p in enumerate(parts):

# if i == len(parts) - 1:

# if p in container:

# print(f"Warning: replacing adapter for {target_type}")

# container[p] = adapter_creator

# else:

"""Register an adapter to display a type

if isinstance(target_type, str):

register_adapter_using_string(target_type, adapter_creator)

return

if target_type in REGISTERED_ADAPTERS:

logger.warning(f"Warning: replacing adapter for {target_type}")

REGISTERED_ADAPTERS[target_type] = adapter_creator

# normally, the list is created only once when a first adapter is

# asked for, but if an adapter is registered after that point we need

# to update the list

if REGISTERED_ADAPTER_TYPES is not None:

"""Class decorator to register new adapters

def decorate_callable(adapter_creator):

register_adapter(target_type, adapter_creator)

return adapter_creator

"""Register an adapter to display a file type (extension)

if suffix in PATH_SUFFIX_ADAPTERS:

logger.warning(f"Replacing path adapter for {suffix}")

"""Class/function decorator to register new file-type adapters

def decorate_callable(adapter_creator):

register_path_adapter(suffix, adapter_creator, required_module)

return adapter_creator

# first check precise type

if data_type in REGISTERED_ADAPTERS:

return REGISTERED_ADAPTERS[data_type]

data_type_full_module_name = data_type.__module__

if '.' in data_type_full_module_name:

data_type_top_module_name, _ = data_type_full_module_name.split('.', maxsplit=1)

else:

data_type_top_module_name = data_type_full_module_name

# handle string types

if data_type_top_module_name in REGISTERED_ADAPTERS_USING_STRINGS:

assert data_type_top_module_name in sys.modules

module = sys.modules[data_type_top_module_name]

module_adapters = REGISTERED_ADAPTERS_USING_STRINGS[data_type_top_module_name]

# register all adapters for that module using concrete types (instead

# of string types)

for str_adapter_type, adapter in list(module_adapters.items()):

# submodule

type_name = str_adapter_type

while '.' in type_name and module is not None:

submodule_name, type_name = type_name.split('.', maxsplit=1)

module = getattr(module, submodule_name, None)

# submodule not found (probably not loaded yet)

if module is None:

continue

adapter_type = getattr(module, type_name, None)

if adapter_type is None:

continue

# cache real adapter type if we have (more) objects of that kind to

# display later

REGISTERED_ADAPTERS[adapter_type] = adapter

update_registered_adapter_types()

# remove string form from adapters mapping

del module_adapters[str_adapter_type]

if not module_adapters:

del REGISTERED_ADAPTERS_USING_STRINGS[data_type_top_module_name]

# then check subclasses

if REGISTERED_ADAPTER_TYPES is None:

update_registered_adapter_types()

for adapter_type in REGISTERED_ADAPTER_TYPES:

if issubclass(data_type, adapter_type):

return REGISTERED_ADAPTERS[adapter_type]

obj_type = type(data)

creator = get_adapter_creator_for_type(obj_type)

# 3 options:

# - the type is not handled

if creator is None:

return None

# - all instances of the type are handled by the same adapter

elif isinstance(creator, type) and issubclass(creator, AbstractAdapter):

return creator

# - different adapters handle that type and/or not all instance are handled

else:

global REGISTERED_ADAPTER_TYPES

REGISTERED_ADAPTER_TYPES = list(REGISTERED_ADAPTERS.keys())

# sort classes with longer MRO first, so that subclasses come before

# their parent class

def class_mro_length(cls):

return len(cls.mro())

if data is None:

return None

adapter_creator = get_adapter_creator(data)

if adapter_creator is None:

raise TypeError(f"No Adapter implemented for data with type {type(data)}")

resource_handle = adapter_creator.open(data)

"""

shape_v = shape[:-num_h_axes] if num_h_axes else shape

shape_h = shape[-num_h_axes:] if num_h_axes else ()

# TODO: we should have a way to provide other attributes: format, readonly, font (problematic for colwidth),

# align, tooltips, flags?, min_value, max_value (for the delegate), ...

# I guess data itself will need to be a dict: {'values': ...}

def __init__(self, data, attributes=None):

self.data = data

self.attributes = attributes

# CHECK: filters will probably not make it as-is after quickbar is implemented: they will need to move

# to the axes area

# AND possibly h/vlabels and

# must update the current command

self.current_filter = {}

self._current_sort = []

# FIXME: this is an ugly/quick&dirty workaround

# AFAICT, this is only used in ArrayDelegate

self.dtype = np.dtype(object)

# self.dtype = None

self.vmin = None

self.vmax = None

self._number_format = "%s"

self.sort_key = None # (kind='axis'|'column'|'row', idx_of_kind, direction (1, -1))

# ================================ #

# methods which MUST be overridden #

# ================================ #

# def get_values(self, h_start, v_start, h_stop, v_stop):

# raise NotImplementedError()

# TODO: split this into:

# - extract_chunk_from_data (result is in native/cheapest

# format to produce)

# and

# - native_chunk_to_2D_sequence

# the goal is to cache chunks

def get_chunk_from_data(self, data, h_start, v_start, h_stop, v_stop):

"""

raise NotImplementedError()

def shape2d(self):

raise NotImplementedError()

# =============================== #

# methods which CAN be overridden #

# =============================== #

def cell_activated(self, row_idx, column_idx):

"""

return None

def get_values(self, h_start, v_start, h_stop, v_stop):

return self.get_chunk_from_data(self.data, h_start, v_start, h_stop, v_stop)

@classmethod

def open(cls, data):

"""Open the ressources used by the adapter

return data

def close(self):

"""Close the ressources used by the adapter"""

pass

# TODO: factorize with LArrayArrayAdapter (so that we get the attributes

# handling of LArrayArrayAdapter for all types and the larray adapter

# can benefit from the generic code here

@timed(logger)

def get_data_values_and_attributes(self, h_start, v_start, h_stop, v_stop):

"""h_stop and v_stop should *not* be included"""

# TODO: implement region caching

logger.debug(

f"{self.__class__.__name__}.get_data_values_and_attributes("

f"{h_start=}, {v_start=}, {h_stop=}, {v_stop=})"

)

height, width = self.shape2d()

assert v_start >= 0, f"v_start ({v_start}) is out of bounds (should be >= 0)"

assert h_start >= 0, f"h_start ({h_start}) is out of bounds (should be >= 0)"

assert v_stop >= 0, f"v_stop ({v_stop}) is out of bounds (should be >= 0)"

assert h_stop >= 0, f"h_stop ({h_stop}) is out of bounds (should be >= 0)"

if height > 0:

assert v_start < height, f"v_start ({v_start}) is out of bounds (should be < {height})"

if width > 0:

assert h_start < width, f"h_start ({h_start}) is out of bounds (should be < {width})"

assert v_stop <= height, f"v_stop ({v_stop}) is out of bounds (should be <= {height})"

assert h_stop <= width, f"h_stop ({h_stop}) is out of bounds (should be <= {width})"

chunk_values = self.get_values(h_start, v_start, h_stop, v_stop)

if isinstance(chunk_values, np.ndarray):

assert chunk_values.ndim == 2

logger.debug(f" {chunk_values.shape=}")

elif isinstance(chunk_values, list) and len(chunk_values) == 0:

chunk_values = [[]]

# Without specifying dtype=object, asarray converts sequences

# containing both strings and numbers to all strings which then

# fail in get_color_value, but we do not want to convert

# existing numpy arrays to object dtype. This is a bit silly and

# inefficient for numeric-only sequences, but I do not see

# a better way.

if not isinstance(chunk_values, np.ndarray):

chunk_values = np.asarray(chunk_values, dtype=object)

finite_values = get_finite_numeric_values(chunk_values)

vmin, vmax = self.update_finite_min_max_values(finite_values,

h_start, v_start,

h_stop, v_stop)

color_value = scale_to_01range(finite_values, vmin, vmax)

chunk_format = self.get_format(chunk_values, h_start, v_start, h_stop, v_stop)

return {'data_format': chunk_format,

'values': chunk_values,

'bg_value': color_value}

def get_format(self, chunk_values, h_start, v_start, h_stop, v_stop):

return [[self._number_format]]

def set_format(self, fmt):

"""Change display format"""

# print(f"setting adapter format: {fmt}")

self._number_format = fmt

def from_clipboard_data_to_model_data(self, list_data):

return list_data

def get_axes_labels_and_data_values(self, row_min, row_max, col_min, col_max):

axes_names = self.get_axes_area()

axes_names = axes_names['values'] if isinstance(axes_names, dict) else axes_names

hlabels = self.get_hlabels_values(col_min, col_max)

vlabels = self.get_vlabels_values(row_min, row_max)

raw_data = self.get_values(col_min, row_min, col_max, row_max)

if isinstance(raw_data, list) and len(raw_data) == 0:

raw_data = [[]]

return axes_names, vlabels, hlabels, raw_data

def move_axis(self, data, attributes, old_index, new_index):

"""Move an axis of the data array and associated attribute arrays.

raise NotImplementedError()

def can_filter_axis(self, axis_idx) -> bool:

return False

def get_filter_names(self):

"""return [combo_label, ...]"""

return []

# TODO: change to get_filter_options(filter_idx, start, stop)

# ... in the end, filters will move to axes names

# AND possibly h/vlabels and

# must update the current command

def get_filter_options(self, filter_idx) -> Optional[list]:

"""return [combo_values]"""

return None

def update_filter(self, filter_idx, indices):

"""Update current filter for a given axis if labels selection from the array widget has changed

raise NotImplementedError()

def get_current_filter_indices(self, filter_idx):

"""Returns indices currently selected for a given filter.

return self.current_filter.get(filter_idx)

def map_filtered_to_global(self, filtered_shape, filter, local2dkey):

"""

raise NotImplementedError()

def translate_changes(self, data_model_changes):

to_global = self.map_filtered_to_global

# FIXME: filtered_data is specific to LArray. Either make it part of the API, or do not pass it as argument

# and get it in the implementation of map_filtered_to_global

global_changes = [

CellValueChange(to_global(self.filtered_data.shape, self.current_filter, key),

old_value, new_value)

for key, (old_value, new_value) in data_model_changes.items()

]

return global_changes

def get_sample(self):

"""Return a sample of the internal data"""

# TODO: use default_buffer sizes instead, or, better yet, a new get_preferred_buffer_size() method

height, width = self.shape2d()

# TODO: use this instead (but it currently does not work because get_values does not always

# return a numpy array while the current code does

# return self.get_values(0, 0, min(width, 20), min(height, 20))

return self.get_data_values_and_attributes(0, 0, min(width, 20), min(height, 20))['values']

@timed(logger)

def update_finite_min_max_values(self, finite_values: np.ndarray,

h_start: int, v_start: int,

h_stop: int, v_stop: int):

"""can return either two floats or two arrays"""

# we need initial to support empty arrays

vmin = np.nanmin(finite_values, initial=np.nan)

vmax = np.nanmax(finite_values, initial=np.nan)

self.vmin = (

np.nanmin([self.vmin, vmin]) if self.vmin is not None else vmin)

self.vmax = (

np.nanmax([self.vmax, vmax]) if self.vmax is not None else vmax)

return self.vmin, self.vmax

def get_axes_area(self):

# axes = self.filtered_data.axes

# test axes.size == 0 is required in case an instance built as Array([]) is passed

# test len(axes) == 0 is required when a user filters until getting a scalar (because in that case size is 1)

# TODO: store this in the adapter

# if axes.size == 0 or len(axes) == 0:

# return [[]]

# else:

shape = self.shape2d()

row_idx_names = [name if name is not None else ''

for name in self.get_vnames()]

num_v_axes = len(row_idx_names)

col_idx_names = [name if name is not None else ''

for name in self.get_hnames()]

num_h_axes = len(col_idx_names)

if (not len(row_idx_names) and not len(col_idx_names)) or any(d == 0 for d in shape):

return [[]]

names = np.full((max(num_h_axes, 1), max(num_v_axes, 1)), '', dtype=object)

if len(row_idx_names) > 1:

names[-1, :-1] = row_idx_names[:-1]

if len(col_idx_names) > 1:

names[:-1, -1] = col_idx_names[:-1]

part1 = row_idx_names[-1] if row_idx_names else ''

part2 = col_idx_names[-1] if col_idx_names else ''

sep = '\\' if part1 and part2 else ''

names[-1, -1] = f'{part1}{sep}{part2}'

current_sort = self.get_current_sort()

sorted_axes = {axis_idx: ascending for axis_idx, label_idx, ascending in current_sort

if label_idx == -1}

decoration = np.full_like(names, '', dtype=object)

ascending_to_decoration = {

True: 'arrow_up',

False: 'arrow_down',

}

decoration[-1, :-1] = [ascending_to_decoration[sorted_axes[i]] if i in sorted_axes else ''

for i in range(len(row_idx_names) - 1)]

return {'values': names.tolist(), 'decoration': decoration.tolist()}

def get_current_sort(self) -> list[tuple]:

"""Return current sort

return self._current_sort

# Adapter classes *may* implement this if can_sort_axis returns True for any axis

# (otherwise they will rely on this default implementation)

def axis_sort_direction(self, axis_idx):

"""must return 'ascending', 'descending' or 'unsorted'"""

for cur_sort_axis_idx, label_idx, ascending in self._current_sort:

if cur_sort_axis_idx == axis_idx:

return 'ascending' if ascending else 'descending'

return 'unsorted'

def hlabel_sort_direction(self, row_idx, col_idx):

"""must return 'ascending', 'descending' or 'unsorted'"""

cell_axis_idx = self.hlabel_row_to_axis_num(row_idx)

for axis_idx, label_idx, ascending in self._current_sort:

if axis_idx == cell_axis_idx and label_idx == col_idx:

return 'ascending' if ascending else 'descending'

return 'unsorted'

def can_filter_hlabel(self, row_idx, col_idx) -> bool:

return False

def can_sort_axis_labels(self, axis_idx) -> bool:

return False

def sort_axis_labels(self, axis_idx, ascending):

pass

# TODO: unsure a different result per label is useful. Per axis would probably be enough

def can_sort_hlabel(self, row_idx, col_idx) -> bool:

return False

def sort_hlabel(self, row_idx, col_idx, ascending):

pass

@timed(logger)

def get_vlabels(self, start, stop) -> dict:

chunk_values = self.get_vlabels_values(start, stop)

if isinstance(chunk_values, list) and len(chunk_values) == 0:

chunk_values = [[]]

return {'values': chunk_values}

def get_vlabels_values(self, start, stop):

# Note that using some kind of lazy object here is pointless given that

# we will use most of it (the buffer should not be much larger than the

# viewport). It would make sense to define one big lazy object as

# self._vlabels = Product([range(len(data))]) and use return

# self._vlabels[start:stop] here but I am unsure it is worth it because

# that would be slower than what we have now.

return [[i] for i in range(start, stop)]

@timed(logger)

def get_hlabels(self, start, stop):

values = self.get_hlabels_values(start, stop)

return {'values': values, 'decoration': self.get_hlabels_decorations(start, stop, values)}

def get_hlabels_values(self, start, stop):

return [list(range(start, stop))]

def hlabel_row_to_axis_num(self, row_idx):

return row_idx + self.num_v_axes()

def num_v_axes(self):

return 1

def get_hlabels_decorations(self, start, stop, labels):

current_sort = self.get_current_sort()

sorted_labels_by_axis = {}

for axis_idx, label_idx, ascending in current_sort:

sorted_labels_by_axis.setdefault(axis_idx, {})[label_idx] = ascending

ascending_to_decoration = {

True: 'arrow_up',

False: 'arrow_down',

}

decorations = []

for row_idx in range(len(labels)):

row_axis_idx = self.hlabel_row_to_axis_num(row_idx)

axis_sorted_labels = sorted_labels_by_axis.get(row_axis_idx, {})

decoration_row = [

ascending_to_decoration[axis_sorted_labels[col_idx]] if col_idx in axis_sorted_labels else ''

for col_idx in range(start, stop)

]

decorations.append(decoration_row)

return decorations

def get_vnames(self):

return ['']

def get_hnames(self):

return ['']

def get_vname(self):

return ' '.join(str(name) for name in self.get_vnames())

def get_hname(self):

return ' '.join(str(name) for name in self.get_hnames())

def combine_labels_and_data(self, raw_data, axes_names, vlabels, hlabels):

"""Return list

# we use itertools.chain so that we can combine any iterables, not just lists

chain = itertools.chain

topheaders = [list(chain(axis_row, hlabels_row))

for axis_row, hlabels_row in zip(axes_names, hlabels)]

datarows = [list(chain(row_labels, row_data))

for row_labels, row_data in zip(vlabels, raw_data)]

return topheaders + datarows

# FIXME (unsure this is still the case): this function does not support None axes_names, vlabels and hlabels

# which _selection_data() produces in some cases (notably when working

# on a scalar array). Unsure if we should fix _selection_data or this

# method though.

def get_combined_values(self, row_min, row_max, col_min, col_max):

"""Return ...

axes_names, vlabels, hlabels, raw_data = (

self.get_axes_labels_and_data_values(row_min, row_max,

col_min, col_max)

)

return self.combine_labels_and_data(raw_data, axes_names,

vlabels, hlabels)

def to_string(self, row_min, row_max, col_min, col_max, sep='\t'):

"""Copy selection as tab-separated (clipboard) text

data = self.get_combined_values(row_min, row_max, col_min, col_max)

# np.savetxt make things more complicated, especially on py3

# We do not use repr for everything to avoid having extra quotes for strings.

# XXX: but is it really a problem? Wouldn't it allow us to copy-paste values with sep (tabs) in them?

# I need to test what Excel does for strings

def vrepr(v):

if isinstance(v, float):

return repr(v)

else:

return str(v)

return '\n'.join(sep.join(vrepr(v) for v in line) for line in data)

def to_excel(self, row_min, row_max, col_min, col_max):

"""Export data to an Excel Sheet

import xlwings as xw

data = self.get_combined_values(row_min, row_max, col_min, col_max)

# convert (row) generators to lists then array

# TODO: the conversion to array is currently necessary even though xlwings will translate it back to a list

# anyway. The problem is that our lists contains numpy types and especially np.str_ crashes xlwings.

# unsure how we should fix this properly: in xlwings, or change get_combined_data() to return only

# standard Python types.

array = np.array([list(r) for r in data])

# Create a new Excel instance. We cannot simply use xw.view(array)

# because it reuses the active Excel instance if any, and if that one

# is hidden, the user will not see anything

app = xw.App(visible=True)

# Activate XLA(M) addins. By default, they are not activated when an

# Excel Workbook is opened via COM

xl_app = app.api

for i in range(1, xl_app.AddIns.Count + 1):

addin = xl_app.AddIns(i)

addin_path = addin.FullName

if addin.Installed and '.xll' not in addin_path.lower():

xl_app.Workbooks.Open(addin_path)

# Dump array to first sheet

book = app.books[0]

sheet = book.sheets[0]

with app.properties(screen_updating=False):

sheet["A1"].value = array

# Unsure whether we should do this or not

# sheet.tables.add(sheet["A1"].expand())

sheet.autofit()

# Move Excel Window at the front. Without steal_focus it does not seem

# to do anything

app.activate(steal_focus=True)

def plot(self, row_min, row_max, col_min, col_max):

"""Return a matplotlib.Figure object for selected subset.

from matplotlib.figure import Figure

# we do not use the axes_names part because the position of axes names is up to the adapter

_, vlabels, hlabels, raw_data = self.get_axes_labels_and_data_values(row_min, row_max, col_min, col_max)

if logger.isEnabledFor(logging.DEBUG):

logger.debug(f"AbstractAdapter.plot {vlabels=} {hlabels=}")

logger.debug(f"{raw_data=}")

if not isinstance(raw_data, np.ndarray):

# Without dtype=object, in the presence of a string, raw_data will

# be entirely converted to strings which is not what we want.

raw_data = np.asarray(raw_data, dtype=object)

raw_data = raw_data.reshape((raw_data.shape[0], -1))

assert isinstance(raw_data, np.ndarray), f"got data of type {type(raw_data)}"

assert raw_data.ndim == 2, f"ndim is {raw_data.ndim}"

finite_values = get_finite_numeric_values(raw_data)

figure = Figure()

# create an axis

ax = figure.add_subplot()

# we have a list of rows but we want a list of columns

xlabels = list(zip(*hlabels))

ylabels = vlabels

xlabel = self.get_hname()

ylabel = self.get_vname()

height, width = finite_values.shape

if width == 1:

# plot one column

xlabels, ylabels = ylabels, xlabels

xlabel, ylabel = ylabel, xlabel

height, width = width, height

finite_values = finite_values.T

# plot each row as a line

xticklabels = ['\n'.join(str(label) for label in label_col)

for label_col in xlabels]

xdata = np.arange(width)

for data_row, ylabels_row in zip(finite_values, ylabels):

row_label = ' '.join(str(label) for label in ylabels_row)

ax.plot(xdata, data_row, label=row_label)

# set x axis

if xlabel:

ax.set_xlabel(xlabel)

ax.set_xlim((0, width - 1))

# we need to do that because matplotlib is smart enough to

# not show all ticks but a selection. However, that selection

# may include ticks outside the range of x axis

xticks = [t for t in ax.get_xticks().astype(int) if t < len(xticklabels)]

xticklabels = [xticklabels[t] for t in xticks]

ax.set_xticks(xticks)

ax.set_xticklabels(xticklabels)

# add legend

all_empty_labels = all(not label for yrow in ylabels for label in yrow)

if width != 1 and not all_empty_labels:

kwargs = {'title': ylabel} if ylabel else {}

ax.legend(**kwargs)

"""For adapters where color is per column"""

def __init__(self, data, attributes=None):

super().__init__(data, attributes)

self.vmin = {}

self.vmax = {}

@timed(logger)

def update_finite_min_max_values(self, finite_values: np.ndarray,

h_start: int, v_start: int,

h_stop: int, v_stop: int):

assert isinstance(self.vmin, dict) and isinstance(self.vmax, dict)

assert h_stop >= h_start

# per column => axis=0

local_vmin = np.nanmin(finite_values, axis=0, initial=np.nan)

local_vmax = np.nanmax(finite_values, axis=0, initial=np.nan)

num_cols = h_stop - h_start

assert local_vmin.shape == (num_cols,), \

(f"unexpected shape: {local_vmin.shape} ({finite_values.shape=}) vs "

f"{(num_cols,)} ({h_start=} {h_stop=})")

# vmin or self.vmin can both be nan (if the whole section data

# is/was nan)

global_vmin = self.vmin

global_vmax = self.vmax

vmin_slice = np.empty(num_cols, dtype=np.float64)

vmax_slice = np.empty(num_cols, dtype=np.float64)

for global_col_idx in range(h_start, h_stop):

local_col_idx = global_col_idx - h_start

col_min = np.nanmin([global_vmin.get(global_col_idx, np.nan),

local_vmin[local_col_idx]])

# update the global vmin dict inplace

global_vmin[global_col_idx] = col_min

vmin_slice[local_col_idx] = col_min

col_max = np.nanmax([global_vmax.get(global_col_idx, np.nan),

local_vmax[local_col_idx]])

# update the global vmax dict inplace

global_vmax[global_col_idx] = col_max

vmax_slice[local_col_idx] = col_max

# Path adapters MAY override this method, but should probably override

# open() instead

@classmethod

def get_file_handler_and_adapter_creator(cls, fpath):

file_handler_object = cls.open(fpath)

return file_handler_object, get_adapter_creator(file_handler_object)

# Path adapters MAY override these methods

@classmethod

def open(cls, fpath):

"""The result of this method will be stored in the .data

_COL_NAMES = ['Name', 'Type', 'Date Modified', 'Size']

_SORT_FUNCS = {

# Technically, we should use p.stem, but p.name has a better behavior

# when there are several files with same stem but different suffixes

# (e.g. test.txt and test.csv)

0: lambda p: (not p.is_dir(), p.name),

1: lambda p: (not p.is_dir(), p.suffix.lower()),

2: lambda p: (not p.is_dir(), p.stat().st_mtime),

3: lambda p: (not p.is_dir(), p.stat().st_size),

}

def __init__(self, data, attributes):

# taking absolute() allows going outside of the initial directory

# via double click. This is both good and bad.

data = data.absolute()

super().__init__(data=data, attributes=attributes)

# sort by ascending name by default

self._current_sort = [(1, 0, True)]

self._update_sorted_path_objs()