Aluode/PerceptionLabPortable

1import ast2import contextlib3import logging4import os5import re6from collections.abc import Sequence7from typing import ClassVar8 9import panel as pn10 11from .core import OpenFile, get_filesystem_class, split_protocol12from .registry import known_implementations13 14pn.extension()15logger = logging.getLogger("fsspec.gui")16 17 18class SigSlot:19    """Signal-slot mixin, for Panel event passing20 21    Include this class in a widget manager's superclasses to be able to22    register events and callbacks on Panel widgets managed by that class.23 24    The method ``_register`` should be called as widgets are added, and external25    code should call ``connect`` to associate callbacks.26 27    By default, all signals emit a DEBUG logging statement.28    """29 30    # names of signals that this class may emit each of which must be31    # set by _register for any new instance32    signals: ClassVar[Sequence[str]] = []33    # names of actions that this class may respond to34    slots: ClassVar[Sequence[str]] = []35 36    # each of which must be a method name37 38    def __init__(self):39        self._ignoring_events = False40        self._sigs = {}41        self._map = {}42        self._setup()43 44    def _setup(self):45        """Create GUI elements and register signals"""46        self.panel = pn.pane.PaneBase()47        # no signals to set up in the base class48 49    def _register(50        self, widget, name, thing="value", log_level=logging.DEBUG, auto=False51    ):52        """Watch the given attribute of a widget and assign it a named event53 54        This is normally called at the time a widget is instantiated, in the55        class which owns it.56 57        Parameters58        ----------59        widget : pn.layout.Panel or None60            Widget to watch. If None, an anonymous signal not associated with61            any widget.62        name : str63            Name of this event64        thing : str65            Attribute of the given widget to watch66        log_level : int67            When the signal is triggered, a logging event of the given level68            will be fired in the dfviz logger.69        auto : bool70            If True, automatically connects with a method in this class of the71            same name.72        """73        if name not in self.signals:74            raise ValueError(f"Attempt to assign an undeclared signal: {name}")75        self._sigs[name] = {76            "widget": widget,77            "callbacks": [],78            "thing": thing,79            "log": log_level,80        }81        wn = "-".join(82            [83                getattr(widget, "name", str(widget)) if widget is not None else "none",84                thing,85            ]86        )87        self._map[wn] = name88        if widget is not None:89            widget.param.watch(self._signal, thing, onlychanged=True)90        if auto and hasattr(self, name):91            self.connect(name, getattr(self, name))92 93    def _repr_mimebundle_(self, *args, **kwargs):94        """Display in a notebook or a server"""95        try:96            return self.panel._repr_mimebundle_(*args, **kwargs)97        except (ValueError, AttributeError) as exc:98            raise NotImplementedError(99                "Panel does not seem to be set up properly"100            ) from exc101 102    def connect(self, signal, slot):103        """Associate call back with given event104 105        The callback must be a function which takes the "new" value of the106        watched attribute as the only parameter. If the callback return False,107        this cancels any further processing of the given event.108 109        Alternatively, the callback can be a string, in which case it means110        emitting the correspondingly-named event (i.e., connect to self)111        """112        self._sigs[signal]["callbacks"].append(slot)113 114    def _signal(self, event):115        """This is called by a an action on a widget116 117        Within an self.ignore_events context, nothing happens.118 119        Tests can execute this method by directly changing the values of120        widget components.121        """122        if not self._ignoring_events:123            wn = "-".join([event.obj.name, event.name])124            if wn in self._map and self._map[wn] in self._sigs:125                self._emit(self._map[wn], event.new)126 127    @contextlib.contextmanager128    def ignore_events(self):129        """Temporarily turn off events processing in this instance130 131        (does not propagate to children)132        """133        self._ignoring_events = True134        try:135            yield136        finally:137            self._ignoring_events = False138 139    def _emit(self, sig, value=None):140        """An event happened, call its callbacks141 142        This method can be used in tests to simulate message passing without143        directly changing visual elements.144 145        Calling of callbacks will halt whenever one returns False.146        """147        logger.log(self._sigs[sig]["log"], f"{sig}: {value}")148        for callback in self._sigs[sig]["callbacks"]:149            if isinstance(callback, str):150                self._emit(callback)151            else:152                try:153                    # running callbacks should not break the interface154                    ret = callback(value)155                    if ret is False:156                        break157                except Exception as e:158                    logger.exception(159                        "Exception (%s) while executing callback for signal: %s",160                        e,161                        sig,162                    )163 164    def show(self, threads=False):165        """Open a new browser tab and display this instance's interface"""166        self.panel.show(threads=threads, verbose=False)167        return self168 169 170class SingleSelect(SigSlot):171    """A multiselect which only allows you to select one item for an event"""172 173    signals = ["_selected", "selected"]  # the first is internal174    slots = ["set_options", "set_selection", "add", "clear", "select"]175 176    def __init__(self, **kwargs):177        self.kwargs = kwargs178        super().__init__()179 180    def _setup(self):181        self.panel = pn.widgets.MultiSelect(**self.kwargs)182        self._register(self.panel, "_selected", "value")183        self._register(None, "selected")184        self.connect("_selected", self.select_one)185 186    def _signal(self, *args, **kwargs):187        super()._signal(*args, **kwargs)188 189    def select_one(self, *_):190        with self.ignore_events():191            val = [self.panel.value[-1]] if self.panel.value else []192            self.panel.value = val193        self._emit("selected", self.panel.value)194 195    def set_options(self, options):196        self.panel.options = options197 198    def clear(self):199        self.panel.options = []200 201    @property202    def value(self):203        return self.panel.value204 205    def set_selection(self, selection):206        self.panel.value = [selection]207 208 209class FileSelector(SigSlot):210    """Panel-based graphical file selector widget211 212    Instances of this widget are interactive and can be displayed in jupyter by having213    them as the output of a cell,  or in a separate browser tab using ``.show()``.214    """215 216    signals = [217        "protocol_changed",218        "selection_changed",219        "directory_entered",220        "home_clicked",221        "up_clicked",222        "go_clicked",223        "filters_changed",224    ]225    slots = ["set_filters", "go_home"]226 227    def __init__(self, url=None, filters=None, ignore=None, kwargs=None):228        """229 230        Parameters231        ----------232        url : str (optional)233            Initial value of the URL to populate the dialog; should include protocol234        filters : list(str) (optional)235            File endings to include in the listings. If not included, all files are236            allowed. Does not affect directories.237            If given, the endings will appear as checkboxes in the interface238        ignore : list(str) (optional)239            Regex(s) of file basename patterns to ignore, e.g., "\\." for typical240            hidden files on posix241        kwargs : dict (optional)242            To pass to file system instance243        """244        if url:245            self.init_protocol, url = split_protocol(url)246        else:247            self.init_protocol, url = "file", os.getcwd()248        self.init_url = url249        self.init_kwargs = (kwargs if isinstance(kwargs, str) else str(kwargs)) or "{}"250        self.filters = filters251        self.ignore = [re.compile(i) for i in ignore or []]252        self._fs = None253        super().__init__()254 255    def _setup(self):256        self.url = pn.widgets.TextInput(257            name="url",258            value=self.init_url,259            align="end",260            sizing_mode="stretch_width",261            width_policy="max",262        )263        self.protocol = pn.widgets.Select(264            options=sorted(known_implementations),265            value=self.init_protocol,266            name="protocol",267            align="center",268        )269        self.kwargs = pn.widgets.TextInput(270            name="kwargs", value=self.init_kwargs, align="center"271        )272        self.go = pn.widgets.Button(name="⇨", align="end", width=45)273        self.main = SingleSelect(size=10)274        self.home = pn.widgets.Button(name="🏠", width=40, height=30, align="end")275        self.up = pn.widgets.Button(name="‹", width=30, height=30, align="end")276 277        self._register(self.protocol, "protocol_changed", auto=True)278        self._register(self.go, "go_clicked", "clicks", auto=True)279        self._register(self.up, "up_clicked", "clicks", auto=True)280        self._register(self.home, "home_clicked", "clicks", auto=True)281        self._register(None, "selection_changed")282        self.main.connect("selected", self.selection_changed)283        self._register(None, "directory_entered")284        self.prev_protocol = self.protocol.value285        self.prev_kwargs = self.storage_options286 287        self.filter_sel = pn.widgets.CheckBoxGroup(288            value=[], options=[], inline=False, align="end", width_policy="min"289        )290        self._register(self.filter_sel, "filters_changed", auto=True)291 292        self.panel = pn.Column(293            pn.Row(self.protocol, self.kwargs),294            pn.Row(self.home, self.up, self.url, self.go, self.filter_sel),295            self.main.panel,296        )297        self.set_filters(self.filters)298        self.go_clicked()299 300    def set_filters(self, filters=None):301        self.filters = filters302        if filters:303            self.filter_sel.options = filters304            self.filter_sel.value = filters305        else:306            self.filter_sel.options = []307            self.filter_sel.value = []308 309    @property310    def storage_options(self):311        """Value of the kwargs box as a dictionary"""312        return ast.literal_eval(self.kwargs.value) or {}313 314    @property315    def fs(self):316        """Current filesystem instance"""317        if self._fs is None:318            cls = get_filesystem_class(self.protocol.value)319            self._fs = cls(**self.storage_options)320        return self._fs321 322    @property323    def urlpath(self):324        """URL of currently selected item"""325        return (326            (f"{self.protocol.value}://{self.main.value[0]}")327            if self.main.value328            else None329        )330 331    def open_file(self, mode="rb", compression=None, encoding=None):332        """Create OpenFile instance for the currently selected item333 334        For example, in a notebook you might do something like335 336        .. code-block::337 338            [ ]: sel = FileSelector(); sel339 340            # user selects their file341 342            [ ]: with sel.open_file('rb') as f:343            ...      out = f.read()344 345        Parameters346        ----------347        mode: str (optional)348            Open mode for the file.349        compression: str (optional)350            The interact with the file as compressed. Set to 'infer' to guess351            compression from the file ending352        encoding: str (optional)353            If using text mode, use this encoding; defaults to UTF8.354        """355        if self.urlpath is None:356            raise ValueError("No file selected")357        return OpenFile(self.fs, self.urlpath, mode, compression, encoding)358 359    def filters_changed(self, values):360        self.filters = values361        self.go_clicked()362 363    def selection_changed(self, *_):364        if self.urlpath is None:365            return366        if self.fs.isdir(self.urlpath):367            self.url.value = self.fs._strip_protocol(self.urlpath)368        self.go_clicked()369 370    def go_clicked(self, *_):371        if (372            self.prev_protocol != self.protocol.value373            or self.prev_kwargs != self.storage_options374        ):375            self._fs = None  # causes fs to be recreated376            self.prev_protocol = self.protocol.value377            self.prev_kwargs = self.storage_options378        listing = sorted(379            self.fs.ls(self.url.value, detail=True), key=lambda x: x["name"]380        )381        listing = [382            l383            for l in listing384            if not any(i.match(l["name"].rsplit("/", 1)[-1]) for i in self.ignore)385        ]386        folders = {387            "📁 " + o["name"].rsplit("/", 1)[-1]: o["name"]388            for o in listing389            if o["type"] == "directory"390        }391        files = {392            "📄 " + o["name"].rsplit("/", 1)[-1]: o["name"]393            for o in listing394            if o["type"] == "file"395        }396        if self.filters:397            files = {398                k: v399                for k, v in files.items()400                if any(v.endswith(ext) for ext in self.filters)401            }402        self.main.set_options(dict(**folders, **files))403 404    def protocol_changed(self, *_):405        self._fs = None406        self.main.options = []407        self.url.value = ""408 409    def home_clicked(self, *_):410        self.protocol.value = self.init_protocol411        self.kwargs.value = self.init_kwargs412        self.url.value = self.init_url413        self.go_clicked()414 415    def up_clicked(self, *_):416        self.url.value = self.fs._parent(self.url.value)417        self.go_clicked()418