# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.
"""
Sigima Common tools for signal and image io support
"""
# pylint: disable=invalid-name # Allows short reference names like x, y, ...
from __future__ import annotations
import dataclasses
import enum
import os
import os.path as osp
import re
from typing import Generic, Literal, Sequence
from sigima.config import _
from sigima.objects.base import BaseObj, TypeObj
from sigima.worker import CallbackWorkerProtocol
class UnsupportedFormatOperationError(NotImplementedError):
"""Operation not supported by a file format, rather than an abstract hook."""
class IOAction(enum.Enum):
"""I/O action type"""
LOAD = enum.auto()
SAVE = enum.auto()
class FormatBase(Generic[TypeObj]):
"""Object representing a data file io"""
FORMAT_INFO: FormatInfo = None
def __init__(self):
self.info = self.FORMAT_INFO
if self.info is None:
raise ValueError(f"Format info not set for {self.__class__.__name__}")
if self.info.name is None:
raise ValueError(f"Format name not set for {self.__class__.__name__}")
if self.info.extensions is None:
raise ValueError(f"Format extensions not set for {self.__class__.__name__}")
if not self.info.readable and not self.info.writeable:
raise ValueError(f"Format {self.info.name} is not readable nor writeable")
self.extlist = get_file_extensions(self.info.extensions)
if not self.extlist:
raise ValueError(f"Invalid format extensions for {self.__class__.__name__}")
if self.info.requires:
for package in self.info.requires:
try:
__import__(package)
except ImportError as exc:
raise ImportError(
f"Format {self.info.name} requires {package} package"
) from exc
def get_filter(self, action: IOAction) -> str | None:
"""Return file filter for Qt file dialog
Args:
action: I/O action type
Returns:
File filter string
"""
assert action in (IOAction.LOAD, IOAction.SAVE)
if action == IOAction.LOAD and not self.info.readable:
return None
if action == IOAction.SAVE and not self.info.writeable:
return None
return f"{self.info.name} ({self.info.extensions})"
def read(
self, filename: str, worker: CallbackWorkerProtocol | None = None
) -> Sequence[TypeObj]:
"""Read list of native objects (signal or image) from file.
For single object, return a list with one object.
Args:
filename: file name
worker: Callback worker object
Raises:
NotImplementedError: if format is not supported
Returns:
List of native objects (signal or image)
"""
raise UnsupportedFormatOperationError(
f"Reading from {self.info.name} is not supported"
)
def write(self, filename: str, obj: BaseObj) -> None:
"""Write data to file
Args:
filename: file name
obj: native object (signal or image)
Raises:
NotImplementedError: if format is not supported
"""
raise UnsupportedFormatOperationError(
f"Writing to {self.info.name} is not supported"
)
# pylint: disable=bad-mcs-classmethod-argument
class BaseIORegistry(Generic[TypeObj], type):
"""Metaclass for registering I/O handler classes"""
REGISTRY_INFO: str = "" # Registry info, override in subclasses
_io_format_instances: list[FormatBase] = []
def __init__(cls, name, bases, attrs):
super().__init__(name, bases, attrs)
if not name.endswith("FormatBase"):
try:
# pylint: disable=no-value-for-parameter
cls._io_format_instances.append(cls())
except ImportError:
# This format is not supported
pass
@classmethod
def get_formats(cls) -> list[FormatBase]:
"""Return I/O format handlers
Returns:
List of I/O format handlers
"""
return cls._io_format_instances
@classmethod
def register_format(cls, format_class: type[FormatBase]) -> bool:
"""Register the handler of a format class, if it is not registered
Format classes are registered when they are defined: this method registers
again a format class removed with :meth:`unregister_format`.
Args:
format_class: format class of this registry
Returns:
True if the format was registered, False if it already was
Raises:
TypeError: if format_class is not a format class of this registry
"""
if not isinstance(format_class, cls):
raise TypeError(f"{format_class!r} is not a {cls.__name__} format class")
if any(fmt.__class__ is format_class for fmt in cls._io_format_instances):
return False
cls._io_format_instances.append(format_class())
return True
@classmethod
def unregister_format(cls, format_class: type[FormatBase]) -> bool:
"""Unregister the handler of a format class, e.g. when the plugin defining
it is unloaded
Handlers of subclasses of format_class are kept: they are formats of their
own.
Args:
format_class: format class of this registry
Returns:
True if the format was unregistered, False if it was not registered
"""
instances = cls._io_format_instances
kept = [fmt for fmt in instances if fmt.__class__ is not format_class]
removed = len(kept) != len(instances)
instances[:] = kept
return removed
@classmethod
def get_format_info(cls, mode: Literal["rst", "text"] = "rst") -> str:
"""Return I/O format info
Args:
mode: Output format, either 'rst' (reStructuredText) or 'text'
Returns:
Text description for all I/O formats
"""
if mode == "rst":
txt = f"{cls.REGISTRY_INFO}:\n\n.. list-table::\n :header-rows: 1\n\n"
txt += " * - Name\n - Extensions\n "
txt += "- Readable\n - Writeable\n - Requires\n"
txt += "\n".join([fmt.info.to_rst_table_row() for fmt in cls.get_formats()])
else:
txt = f"{cls.REGISTRY_INFO}:{os.linesep}"
indent = " " * 4
finfo = "\n".join([str(fmt.info) for fmt in cls.get_formats()])
txt += "\n".join([indent + line for line in finfo.splitlines()])
return txt
@classmethod
def __get_all_supported_filter(cls, action: IOAction) -> str:
"""Return all supported file filter for Qt file dialog
Args:
action: I/O action type
Returns:
File filter
"""
extlist = [] # file extension list
for fmt in cls.get_formats():
fmt: FormatBase
if not fmt.info.readable and action == IOAction.LOAD:
continue
if not fmt.info.writeable and action == IOAction.SAVE:
continue
extlist.extend(fmt.extlist)
allsupported = _("All supported files")
return f"{allsupported} ({'*.' + ' *.'.join(extlist)})"
@classmethod
def get_filters(cls, action: IOAction) -> str:
"""Return file filters for Qt file dialog
Args:
action: I/O action type
Returns:
File filters
"""
flist = [] # file filter list
flist.append(cls.__get_all_supported_filter(action))
for fmt in cls.get_formats():
fmt: FormatBase
flt = fmt.get_filter(action)
if flt is not None:
flist.append(flt)
return "\n".join(flist)
@classmethod
def get_read_filters(cls) -> str:
"""Return file filters for Qt open file dialog
Returns:
File filters
"""
return cls.get_filters(IOAction.LOAD)
@classmethod
def get_write_filters(cls) -> str:
"""Return file filters for Qt save file dialog
Returns:
File filters
"""
return cls.get_filters(IOAction.SAVE)
@classmethod
def get_format(cls, filename: str, action: IOAction) -> FormatBase:
"""Return format handler for filename
Args:
filename: file name
action: I/O action type
Raises:
NotImplementedError: if file data type is not supported
Returns:
Format handler
"""
for fmt in cls.get_formats():
fmt: FormatBase
if osp.splitext(filename)[1][1:].lower() in fmt.extlist:
if not fmt.info.readable and action == IOAction.LOAD:
continue
if not fmt.info.writeable and action == IOAction.SAVE:
continue
return fmt
raise NotImplementedError(
f"{filename} is not supported for {action.name.lower()}"
)
@classmethod
def read(
cls, filename: str, worker: CallbackWorkerProtocol | None = None
) -> Sequence[TypeObj]:
"""Read data from file, return native object (signal or image) list.
For single object, return a list with one object.
Args:
filename: file name
worker: Callback worker object
Raises:
NotImplementedError: if file data type is not supported
Returns:
List of native objects (signal or image)
"""
fmt = cls.get_format(filename, IOAction.LOAD)
return fmt.read(filename, worker)
@classmethod
def write(cls, filename: str, obj: BaseObj) -> None:
"""Write data to file from native object (signal or image).
Args:
filename: file name
obj: native object (signal or image)
Raises:
NotImplementedError: if file data type is not supported
"""
fmt = cls.get_format(filename, IOAction.SAVE)
fmt.write(filename, obj)
def get_file_extensions(string: str) -> list[str]:
"""Return a sorted list of unique file extensions contained in `string`.
Args:
string: String containing file extensions.
Returns:
List of file extensions.
"""
pattern = r"\S+\.[\w-]+"
matches = re.findall(pattern, string)
return sorted({match.split(".")[-1].lower() for match in matches})
