from __future__ import annotations

from typing import (
    TYPE_CHECKING, ClassVar, Mapping, Optional,
)

import param

from ..models.feed import Feed as PnFeed, ScrollButtonClick, ScrollLatestEvent
from ..util import edit_readonly
from .base import Column

if TYPE_CHECKING:
    from bokeh.document import Document
    from bokeh.model import Model
    from pyviz_comms import Comm

    from ..viewable import Viewable


class Feed(Column):
    """
    The `Feed` class inherits from the `Column` layout, thereby enabling the arrangement of
    multiple panel objects within a vertical container. However, it restrictively manages the number
    of objects displayed at any moment. This layout is particularly useful for efficiently
    rendering a substantial number of objects.

    Similar to `Column`, the `Feed` provides a list-like API, including methods such as `append`,
    `extend`, `clear`, `insert`, `pop`, `remove`, and `__setitem__`. These methods facilitate
    interactive updates and modifications to the layout.

    Reference: https://panel.holoviz.org/reference/layouts/Feed.html

    :Example:

    >>> pn.Feed(some_widget, some_pane, some_python_object, ..., python_object_1002)
    """

    load_buffer = param.Integer(default=50, bounds=(0, None), doc="""
        The number of objects loaded on each side of the visible objects.
        When scrolled halfway into the buffer, the feed will automatically
        load additional objects while unloading objects on the opposite side.""")

    scroll = param.Selector(
        default="y",
        objects=[False, True, "both-auto", "y-auto", "x-auto", "both", "x", "y"],
        doc="""Whether to add scrollbars if the content overflows the size
        of the container. If "both-auto", will only add scrollbars if
        the content overflows in either directions. If "x-auto" or "y-auto",
        will only add scrollbars if the content overflows in the
        respective direction. If "both", will always add scrollbars.
        If "x" or "y", will always add scrollbars in the respective
        direction. If False, overflowing content will be clipped.
        If True, will only add scrollbars in the direction of the container,
        (e.g. Column: vertical, Row: horizontal).""")

    visible_range = param.Range(readonly=True, doc="""
        Read-only upper and lower bounds of the currently visible feed objects.
        This list is automatically updated based on scrolling.""")

    _bokeh_model: ClassVar[type[Model]] = PnFeed

    _direction = 'vertical'

    _rename: ClassVar[Mapping[str, str | None]] = {
        'objects': 'children', 'visible_range': 'visible_children',
        'load_buffer': None,
    }

    def __init__(self, *objects, **params):
        for height_param in ["height", "min_height", "max_height"]:
            if height_param in params:
                break
        else:
            # sets a default height to prevent infinite load
            params["height"] = 300

        super().__init__(*objects, **params)
        self._last_synced = None
        self.param.watch(self._trigger_view_latest, 'objects')

    @param.depends("visible_range", "load_buffer", watch=True)
    def _trigger_get_objects(self):
        if self.visible_range is None:
            return

        # visible start, end / synced start, end
        vs, ve = self.visible_range
        ss, se = self._last_synced
        half_buffer = self.load_buffer // 2

        top_trigger = (vs - ss) < half_buffer
        bottom_trigger = (se - ve) < half_buffer
        invalid_trigger = (
            # to prevent being trapped and unable to scroll
            ve - vs < self.load_buffer and
            ve - vs < len(self.objects)
        )
        if top_trigger or bottom_trigger or invalid_trigger:
            self.param.trigger("objects")

    def _trigger_view_latest(self, event):
        if (event.type == 'triggered' or not self.view_latest or
            not event.new or event.new[-1] in event.old):
            return
        self.scroll_to_latest()

    @property
    def _synced_range(self):
        n = len(self.objects)
        if self.visible_range:
            return (
                max(self.visible_range[0] - self.load_buffer, 0),
                min(self.visible_range[-1] + self.load_buffer, n)
            )
        elif self.view_latest:
            return (max(n - self.load_buffer * 2, 0), n)
        else:
            return (0, min(self.load_buffer, n))

    def _get_model(
        self, doc: Document, root: Optional[Model] = None,
        parent: Optional[Model] = None, comm: Optional[Comm] = None
    ) -> Model:
        model = super()._get_model(doc, root, parent, comm)
        self._register_events('scroll_button_click', model=model, doc=doc, comm=comm)
        return model

    def _process_property_change(self, msg):
        if 'visible_children' in msg:
            visible = msg.pop('visible_children')
            for model, _ in self._models.values():
                refs = [c.ref['id'] for c in model.children]
                if visible and visible[0] in refs:
                    indexes = sorted(refs.index(v) for v in visible if v in refs)
                    break
            else:
                return super()._process_property_change(msg)
            offset = self._synced_range[0]
            n = len(self.objects)
            visible_range = [
                max(offset + indexes[0], 0),
                min(offset + indexes[-1], n)
            ]
            if visible_range[0] >= visible_range[1]:
                visible_range[0] = visible_range[1] - self.load_buffer
            msg['visible_range'] = tuple(visible_range)
        return super()._process_property_change(msg)

    def _process_param_change(self, msg):
        msg.pop('visible_range', None)
        return super()._process_param_change(msg)

    def _get_objects(
        self, model: Model, old_objects: list[Viewable], doc: Document,
        root: Model, comm: Optional[Comm] = None
    ):
        from ..pane.base import RerenderError
        new_models, old_models = [], []
        self._last_synced = self._synced_range

        for obj in old_objects:
            if obj not in self.objects:
                obj._cleanup(root)

        current_objects = list(self.objects)
        ref = root.ref['id']
        for i in range(*self._last_synced):
            pane = current_objects[i]
            if ref in pane._models:
                child, _ = pane._models[root.ref['id']]
                old_models.append(child)
            else:
                try:
                    child = pane._get_model(doc, root, model, comm)
                except RerenderError as e:
                    if e.layout is not None and e.layout is not self:
                        raise e
                    e.layout = None
                    return self._get_objects(model, current_objects[:i], doc, root, comm)
            new_models.append(child)
        return new_models, old_models

    def _process_event(self, event: ScrollButtonClick | None = None) -> None:
        """
        Process a scroll button click event.
        """
        if not self.visible_range:
            return

        # need to get it all the way to the bottom rather
        # than the center of the buffer zone
        load_buffer = self.load_buffer
        with param.discard_events(self):
            self.load_buffer = 1

        n = len(self.objects)
        n_visible = self.visible_range[-1] - self.visible_range[0]
        with edit_readonly(self):
            # plus one to center on the last object
            self.visible_range = (min(max(n - n_visible + 1, 0), n), n)

        with param.discard_events(self):
            # reset the buffers and loaded objects
            self.load_buffer = load_buffer

    def scroll_to_latest(self):
        """
        Scrolls the Feed to the latest entry.
        """
        rerender = self._last_synced and self._last_synced[-1] < len(self.objects)
        if rerender:
            self._process_event()
        self._send_event(ScrollLatestEvent, rerender=rerender)