RecycleViewを使ってchat viewerを実装

chat_viewer.py

import itertools
from string import Template
from collections import defaultdict
from kivy.config import Config
Config.set('graphics', 'width', 400)
Config.set('graphics', 'height', 800)
from kivy.uix.boxlayout import BoxLayout
from kivy.lang import Builder
from kivy.utils import escape_markup
from kivy.clock import Clock


Builder.load_string(
'''
<ChatMsg@Label>:
    # size_hint: 1, None
    height: self.texture_size[1]
    text_size: self.width, None
    canvas.before:
        Color:
            rgba: 1, 1, 1, .4
        Line:
            points: [self.x, self.y, self.right, self.y, ]
    
<ChatViewer>:
    orientation: 'vertical'
    BoxLayout:
        size_hint_y: None
        height: self.minimum_height
        padding: 5
        spacing: 5
        canvas.before:
            Color:
                rgb: .1, .1, .1
            Rectangle:
                pos: self.pos
                size: self.size
        Widget:
        Button:
            text: 'clear'
            size_hint_x: None
            width: self.texture_size[0] + 10
            size_hint_min_y: self.texture_size[1] + 10
            on_press: rv.data = []
    RecycleView:
        id: rv
        viewclass: 'ChatMsg'
        always_overscroll: False
        RecycleBoxLayout:
            size_hint: 1, None
            height: self.minimum_height
            default_size_hint: 1, None
            default_size: None, None
            spacing: 5
            padding: 5
            orientation: 'vertical'
''')


class ChatViewer(BoxLayout):
    COLORS = '#00FF00 #FF00FF #008888 #44AA44 #808080 #FF7722 #6699FF'.split()

    def on_kv_post(self, *args, **kwargs):
        super().on_kv_post(*args, **kwargs)
        self._rv = self.ids.rv.__self__
        self._user_color_map = defaultdict(lambda color_iter=itertools.cycle(self.COLORS): next(color_iter))
        self._queued_msgs = []
        self._trigger_update = Clock.create_trigger(self._update, 1)

    def add_msg(self, *, msg:str, who:str):
        self._queued_msgs.append((msg, who, ))
        self._trigger_update()

    def _update(self, dt, *, _template=Template(r"[b][color=${color}]${who}:[/color][/b] ${msg}")):
        self._rv.data.extend(
            {
                'markup': True,
                'text': _template.substitute(color=self._user_color_map[who], who=who, msg=escape_markup(msg)),
            } for msg, who in self._queued_msgs
        )
        self._queued_msgs.clear()


def _test():
    import random
    from kivy.clock import Clock
    from kivy.app import App

    user_names = 'Alice Bob Chris David Elena Frank Georg Helen Ivy Jeff '.split()
    msgs = (
        "Hello", "What's up?", "All your base are belong to us",
        "Python is a programming language that lets you work quickly "
        "and integrate systems more effectively. Learn More <<<",
        "PEP 492 introduced support for native coroutines and async/await "
        "syntax to Python 3.5. It is proposed here to extend Python's "
        "asynchronous capabilities by adding support for asynchronous "
        "generators.",
        "special characters : [ & ]",
    )
    
    class SampleApp(App):
        def build(self):
            return ChatViewer()
        def on_start(self):
            clock_event = Clock.schedule_interval(self.add_random_msg, .1)
            Clock.schedule_once(lambda __: clock_event.cancel(), 20)
        def add_random_msg(self, __):
            self.root.add_msg(msg=random.choice(msgs), who=random.choice(user_names))
    SampleApp().run()


if __name__ == '__main__':
    _test()