"""Wrapper for one data-reading filter module.""" from __future__ import annotations from types import ModuleType from typing import Any from .tkfilterutils import call_compatible, get_module_attr, is_error_result class tkFilter: """Thin wrapper around an existing module-style filter. A filter module can keep the current API: - check_file_type(infile, inf=None, app=None, cparams=None) - get_input_type(inf=None, app=None, cparams=None) - get_output_type(inf=None, app=None, cparams=None) - read_data(infile, app=None, cparams=None, is_print=False) - convert(inf, app=None, cparams=None) - print_data(inf, app=None, cparams=None) - plot_data(inf, app=None, cparams=None) - save_data(outfiles, inf, app=None, cparams=None, is_print=False) """ def __init__( self, name: str, module: ModuleType, path: str | None = None, target: str = "read_data", desc: str = "", ) -> None: self.name = name self.module = module self.path = path self.target = target self.desc = desc # Match the old tkApplication.load_module() behavior. setattr(self.module, "name", name) setattr(self.module, "target", target) setattr(self.module, "desc", desc) def __repr__(self) -> str: return f"tkFilter(name={self.name!r}, path={self.path!r})" def has(self, func_name: str) -> bool: return callable(getattr(self.module, func_name, None)) def get(self, name: str, default: Any = None) -> Any: return get_module_attr(self.module, name, default) def call(self, func_name: str, *args: Any, required: bool = False, **kwargs: Any) -> Any: func = getattr(self.module, func_name, None) if not callable(func): if required: raise AttributeError(f"Filter [{self.name}] does not define {func_name}()") return None return call_compatible(func, *args, **kwargs) def get_input_type(self, inf: Any = None, app: Any = None, cparams: Any = None) -> Any: if self.has("get_input_type"): return self.call("get_input_type", inf=inf, app=app, cparams=cparams) return {"file_type": self.get("input_type", self.name)} def get_output_type(self, inf: Any = None, app: Any = None, cparams: Any = None) -> Any: if self.has("get_output_type"): return self.call("get_output_type", inf=inf, app=app, cparams=cparams) return {"file_type": self.get("output_type", None)} def check_file_type(self, path: str, inf: Any = None, app: Any = None, cparams: Any = None) -> Any: if self.has("check_file_type"): return self.call("check_file_type", path, inf=inf, app=app, cparams=cparams) # Minimal fallback for old/simple filters that only define default_ext. default_ext = self.get("default_ext", None) input_type = self.get("input_type", self.name) if default_ext is None: return None import os _, ext = os.path.splitext(path) if ext.lower() == str(default_ext).lower(): return {"file_type": input_type} return None def is_match(self, path: str, inf: Any = None, app: Any = None, cparams: Any = None) -> tuple[bool, Any]: result = self.check_file_type(path, inf=inf, app=app, cparams=cparams) if result is None or is_error_result(result): return False, result return True, result def read_data( self, path: str, app: Any = None, cparams: Any = None, is_print: bool = False, convert: bool = True, ) -> Any: inf = self.call( "read_data", path, app=app, cparams=cparams, is_print=is_print, required=True, ) if convert and self.has("convert"): inf2 = self.call("convert", inf, app=app, cparams=cparams) if inf2 is not None: inf = inf2 return inf def print_data(self, inf: Any, app: Any = None, cparams: Any = None) -> Any: return self.call("print_data", inf, app=app, cparams=cparams) def plot_data(self, inf: Any, app: Any = None, cparams: Any = None) -> Any: return self.call("plot_data", inf, app=app, cparams=cparams) def save_data( self, outfiles: list[str], inf: Any, app: Any = None, cparams: Any = None, is_print: bool = False, ) -> Any: return self.call("save_data", outfiles, inf, app=app, cparams=cparams, is_print=is_print)