Konubinix' opinionated web of thoughts

Pomodo Timer on Kivy

Fleeting

pomodo timer on kivy, using my a python runtime on android

import time

import plyer
from kivy.app import App
from jnius import autoclass
import requests
from kivy.clock import Clock
from kivy.lang import Builder
from kivy.uix.boxlayout import BoxLayout
from kivy.core.camera import Camera
import base64
from datetime import datetime, timedelta
from kivy.graphics import Color, Rectangle
from kivy.properties import StringProperty
import os
from datetime import datetime
from kivy.uix.screenmanager import Screen, ScreenManagerException, ScreenManager
from helpers.osc import oschandler
from helpers.wakelock import WakeLock
from logging import getLogger

logger = getLogger(__name__)

wl = WakeLock('pomodorotimer', wakeup=True)

pink = (1, 0.75, 0.8, 1)
cyan = (0, 1, 1, 1)
grey = (0.3, 0.3, 0.3, 1)
green = (0, 0.9, 0.3, 1)
blue = (0, 0.8, 0.8, 1)
yellow = (0.8, 0.8, 0, 1)

Builder.load_string("""

<Pomodoro>:
    id: pomodoro
    orientation: "vertical"

    BoxLayout:
        orientation: "horizontal"
        size_hint: (1, 0.5)
        Label:
            id: endtime
            text: ""
            font_size: 100
            color: (0, 0, 0, 1)

        Label:
            id: countdown
            text: ""
            font_size: 100
            color: (0, 0, 0, 1)

    ProgressBar:
        id: progress

    BoxLayout:
        orientation: "horizontal"
        size_hint: (1, 0.5)
        Button:
            id: toggle
            disabled: pomodoro.state not in ("idle", "break")
            text: "Start"
            on_press: root.start()
            color: (0, 0, 0, 1)
            background_color: {pink}

        Button:
            id: toggle
            text: "Break"
            disabled: pomodoro.state != "idle"
            on_press: root.short_break()
            color: (0, 0, 0, 1)
            background_color: {green}

        Button:
            id: toggle
            disabled: pomodoro.state != "idle"
            text: "Long Break"
            on_press: root.long_break()
            color: (0, 0, 0, 1)
            background_color: {green}

        Button:
            id: toggle
            disabled: pomodoro.state not in ("pomodoro", "interruption")
            text: "Interruption"
            on_press: root.interruption()
            color: (0, 0, 0, 1)
            background_color: {blue}

        Button:
            id: toggle
            text: "Exterruption"
            disabled: pomodoro.state not in ("pomodoro", "exterruption")
            on_press: root.exterruption()
            color: (0, 0, 0, 1)
            background_color: {blue}

        Button:
            id: toggle
            disabled: pomodoro.state not in ("pomodoro", "twominutes")
            text: "Two minutes"
            on_press: root.twominutes()
            color: (0, 0, 0, 1)
            background_color: {yellow}


# copy from the source code of kivy, but increased
<ProgressBar>:
    canvas:
        Color:
            rgb: 1, 1, 1
        BorderImage:
            border: (12, 12, 12, 12)
            pos: self.x, self.y
            size: self.width, self.height
            source: 'atlas://data/images/defaulttheme/progressbar_background'
        BorderImage:
            border: [int(min(self.width * (self.value / float(self.max)) if self.max else 0, 12))] * 4
            pos: self.x, self.y
            size: self.width * (self.value / float(self.max)) if self.max else 0, self.height
            source: 'atlas://data/images/defaulttheme/progressbar'

""".format(**{"blue": blue, "yellow": yellow, "pink": pink, "cyan": cyan, "grey": grey, "green": green, "yellow": yellow}))

class PomodoroScreen(Screen):
    def __init__(self, *args, **kwargs):
        super(PomodoroScreen, self).__init__(*args, **kwargs)
        self.pomodorotimer = Pomodoro()
        self.add_widget(self.pomodorotimer)
        self.bind(on_pre_enter=self.pre_enter_handler)
        self.bind(on_pre_leave=self.pre_leave_handler)
        @oschandler("pomodoro:start")
        def _(_):
            self.pomodorotimer.start()

        @oschandler("pomodoro:break")
        def _(_):
            self.pomodorotimer.short_break()

    def pre_leave_handler(self, _):
        self.pomodorotimer.leave()

    def pre_enter_handler(self, _):
        from plyer import orientation
        orientation.set_landscape(reverse=True)
        self.pomodorotimer.enter()

class Pomodoro(BoxLayout):
    state = StringProperty("idle")

    def _update_rect(self, instance, value):
        self.rect.pos = instance.pos
        self.rect.size = instance.size

    def __init__(self, **kwargs):
        super(Pomodoro, self).__init__(**kwargs)
        with self.canvas.before:
            self.color = Color(1, 0, 0, 1)
            self.rect = Rectangle(size=self.size, pos=self.pos)

        self.interrupted_time = None
        self.bind(size=self._update_rect, pos=self._update_rect)
        self.pomodoro_duration = 25

    def update(self, _):
        if self.running:
            self.ids.progress.value += 1
            remaining = int(self.ids.progress.max - self.ids.progress.value)
            self.ids.countdown.text = "{:02d}:{:02d}".format(remaining // 60, remaining % 60)
            if self.ids.progress.value == self.ids.progress.max:
                self.interrupted_time = None
                self.ids.progress.value = 0
                if self.state == "pomodoro":
                    requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "done"}''')
                elif self.state == "break":
                    requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "done_break"}''')
                elif self.state in ("interruption", "exterruption"):
                    requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "interruption_too_big"}''')
                elif self.state == "twominutes":
                    requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "two_minutes_over"}''')
                requests.delete("http://192.168.1.245:8500/v1/kv/pomodoro")
                self.reset()
        else:
            end = datetime.now() + timedelta(minutes=25)
            self.ids.endtime.text = end.strftime("%H:%M")
            self.ids.countdown.text = ""

    def enter(self):
        self.reset()
        resp = requests.get("http://192.168.1.245:8500/v1/kv/pomodoro")
        if resp.status_code == 200:
            self.start_state()
            # endtime = datetime.fromisoformat(base64.b64decode(resp.json()[0]["Value"]).decode())
            # compatibility python2
            endtime = datetime.strptime(base64.b64decode(resp.json()[0]["Value"]).decode(), "%Y-%m-%dT%H:%M:%S.%f")
            now = datetime.now()
            self.ids.progress.value = self.ids.progress.max - (endtime - now).total_seconds()
            self.ids.endtime.text = endtime.strftime("%H:%M")

        self.clock = Clock.schedule_interval(self.update, 1)
        wl.acquire()

    def leave(self):
        Clock.unschedule(self.clock)
        self.clock = None
        wl.release()

    def reset(self):
        self.state = "idle"
        self.running = False
        self.color.rgb = grey[:3]

    def interruption(self):
        if self.state == "interruption":
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "handled_interruption"}''')
            if self.interrupted_time:
                self.start()
            else:
                self.reset()
            return
        else:
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "interruption"}''')

        if self.state == "pomodoro" and not self.interrupted_time:
            self.interrupted_time = (self.ids.progress.value, self.ids.progress.max)

        self.reset()
        self.running = True
        self.state = "interruption"

        duration = 1
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = blue[:3]

    def exterruption(self):
        if self.state == "exterruption":
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "handled_ext_interruption"}''')
            if self.interrupted_time:
                self.start()
            else:
                self.reset()
            return
        else:
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "external_interruption"}''')

        if self.state == "pomodoro" and not self.interrupted_time:
            self.interrupted_time = (self.ids.progress.value, self.ids.progress.max)
        self.reset()
        self.running = True
        self.state = "exterruption"
        duration = 1
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = blue[:3]

    def twominutes(self):
        if self.state == "twominutes":
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "two_minutes_done"}''')
            if self.interrupted_time:
                self.start()
            else:
                self.reset()
            return
        else:
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "two_minutes"}''')

        if self.state == "pomodoro" and not self.interrupted_time:
            self.interrupted_time = (self.ids.progress.value, self.ids.progress.max)
        self.reset()

        self.running = True
        self.state = "twominutes"
        duration = 2
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = yellow[:3]

    def short_break(self):
        requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "start_break"}''')
        self.reset()

        self.running = True
        self.state = "break"
        duration = 4
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = green[:3]

    def long_break(self):
        requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "start_long_break"}''')
        self.reset()

        self.running = True
        self.state = "break"
        duration = 4
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = green[:3]

    def start_state(self):
        self.running = True
        self.state = "pomodoro"
        self.color.rgb = pink[:3]
        self.ids.progress.value = 0
        self.ids.progress.max = self.pomodoro_duration * 60

    def start(self):
        if self.running and self.state == "pomodoro":
            logger.debug("Already in a pomodoro")
            return
        self.start_state()
        if self.interrupted_time:
            (self.ids.progress.value, self.ids.progress.max) = self.interrupted_time
            remaining_seconds = self.ids.progress.max - self.ids.progress.value
            end = datetime.now() + timedelta(seconds=remaining_seconds)
            self.interrupted_time = None
            duration = remaining_seconds // 60
        else:
            duration = self.pomodoro_duration
            end = datetime.now() + timedelta(minutes=duration)
        self.ids.endtime.text = end.strftime("%H:%M")

        session_id = requests.put("http://192.168.1.245:8500/v1/session/create",
                     json={
                         "TTL": "{}m".format(duration),
                         "Behavior": "delete",
                         "Name": "pomodoro",
                         "Checks": [],
                     }).json()["ID"]
        logger.debug("Registering pomodoro for session {}".format(session_id))
        requests.delete("http://192.168.1.245:8500/v1/kv/pomodoro")
        requests.put("http://192.168.1.245:8500/v1/kv/pomodoro?acquire={}".format(session_id),
                     data=end.isoformat())
        requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='{{"message": "start_pomodoro", "end": "{}"}}'.format(end.isoformat()))

class PomodoroApp(App):

    @staticmethod
    def populate(sm):
        try:
            sm.get_screen('pomodorotimer')
            return
        except ScreenManagerException:
            pass # not populated yet

        sm.add_widget(PomodoroScreen(name='pomodorotimer'))

    def build(self):
        sm = ScreenManager()
        self.populate(sm)
        return sm


def run():
    PomodoroApp().run()

import time

import plyer
from kivy.app import App
from jnius import autoclass
import requests
from kivy.clock import Clock
from kivy.lang import Builder
from kivy.uix.boxlayout import BoxLayout
from kivy.core.camera import Camera
import base64
from datetime import datetime, timedelta
from kivy.graphics import Color, Rectangle
from kivy.properties import StringProperty
import os
from datetime import datetime
from kivy.uix.screenmanager import Screen, ScreenManagerException, ScreenManager
from helpers.osc import oschandler
from helpers.wakelock import WakeLock
from logging import getLogger

logger = getLogger(__name__)

wl = WakeLock('pomodorotimer', wakeup=True)

pink = (1, 0.75, 0.8, 1)
cyan = (0, 1, 1, 1)
grey = (0.3, 0.3, 0.3, 1)
green = (0, 0.9, 0.3, 1)
blue = (0, 0.8, 0.8, 1)
yellow = (0.8, 0.8, 0, 1)

Builder.load_string("""

<Pomodoro>:
    id: pomodoro
    orientation: "vertical"

    BoxLayout:
        orientation: "horizontal"
        size_hint: (1, 0.5)
        Label:
            id: endtime
            text: ""
            font_size: 100
            color: (0, 0, 0, 1)

        Label:
            id: countdown
            text: ""
            font_size: 100
            color: (0, 0, 0, 1)

    ProgressBar:
        id: progress

    BoxLayout:
        orientation: "horizontal"
        size_hint: (1, 0.5)
        Button:
            id: toggle
            disabled: pomodoro.state not in ("idle", "break")
            text: "Start"
            on_press: root.start()
            color: (0, 0, 0, 1)
            background_color: {pink}

        Button:
            id: toggle
            text: "Break"
            disabled: pomodoro.state != "idle"
            on_press: root.short_break()
            color: (0, 0, 0, 1)
            background_color: {green}

        Button:
            id: toggle
            disabled: pomodoro.state != "idle"
            text: "Long Break"
            on_press: root.long_break()
            color: (0, 0, 0, 1)
            background_color: {green}

        Button:
            id: toggle
            disabled: pomodoro.state not in ("pomodoro", "interruption")
            text: "Interruption"
            on_press: root.interruption()
            color: (0, 0, 0, 1)
            background_color: {blue}

        Button:
            id: toggle
            text: "Exterruption"
            disabled: pomodoro.state not in ("pomodoro", "exterruption")
            on_press: root.exterruption()
            color: (0, 0, 0, 1)
            background_color: {blue}

        Button:
            id: toggle
            disabled: pomodoro.state not in ("pomodoro", "twominutes")
            text: "Two minutes"
            on_press: root.twominutes()
            color: (0, 0, 0, 1)
            background_color: {yellow}


# copy from the source code of kivy, but increased
<ProgressBar>:
    canvas:
        Color:
            rgb: 1, 1, 1
        BorderImage:
            border: (12, 12, 12, 12)
            pos: self.x, self.y
            size: self.width, self.height
            source: 'atlas://data/images/defaulttheme/progressbar_background'
        BorderImage:
            border: [int(min(self.width * (self.value / float(self.max)) if self.max else 0, 12))] * 4
            pos: self.x, self.y
            size: self.width * (self.value / float(self.max)) if self.max else 0, self.height
            source: 'atlas://data/images/defaulttheme/progressbar'

""".format(**{"blue": blue, "yellow": yellow, "pink": pink, "cyan": cyan, "grey": grey, "green": green, "yellow": yellow}))

class PomodoroScreen(Screen):
    def __init__(self, *args, **kwargs):
        super(PomodoroScreen, self).__init__(*args, **kwargs)
        self.pomodorotimer = Pomodoro()
        self.add_widget(self.pomodorotimer)
        self.bind(on_pre_enter=self.pre_enter_handler)
        self.bind(on_pre_leave=self.pre_leave_handler)
        @oschandler("pomodoro:start")
        def _(_):
            self.pomodorotimer.start()

        @oschandler("pomodoro:break")
        def _(_):
            self.pomodorotimer.short_break()

    def pre_leave_handler(self, _):
        self.pomodorotimer.leave()

    def pre_enter_handler(self, _):
        from plyer import orientation
        orientation.set_landscape(reverse=True)
        self.pomodorotimer.enter()

class Pomodoro(BoxLayout):
    state = StringProperty("idle")

    def _update_rect(self, instance, value):
        self.rect.pos = instance.pos
        self.rect.size = instance.size

    def __init__(self, **kwargs):
        super(Pomodoro, self).__init__(**kwargs)
        with self.canvas.before:
            self.color = Color(1, 0, 0, 1)
            self.rect = Rectangle(size=self.size, pos=self.pos)

        self.interrupted_time = None
        self.bind(size=self._update_rect, pos=self._update_rect)
        self.pomodoro_duration = 25

    def update(self, _):
        if self.running:
            self.ids.progress.value += 1
            remaining = int(self.ids.progress.max - self.ids.progress.value)
            self.ids.countdown.text = "{:02d}:{:02d}".format(remaining // 60, remaining % 60)
            if self.ids.progress.value == self.ids.progress.max:
                self.interrupted_time = None
                self.ids.progress.value = 0
                if self.state == "pomodoro":
                    requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "done"}''')
                elif self.state == "break":
                    requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "done_break"}''')
                elif self.state in ("interruption", "exterruption"):
                    requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "interruption_too_big"}''')
                elif self.state == "twominutes":
                    requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "two_minutes_over"}''')
                requests.delete("http://192.168.1.245:8500/v1/kv/pomodoro")
                self.reset()
        else:
            end = datetime.now() + timedelta(minutes=25)
            self.ids.endtime.text = end.strftime("%H:%M")
            self.ids.countdown.text = ""

    def enter(self):
        self.reset()
        resp = requests.get("http://192.168.1.245:8500/v1/kv/pomodoro")
        if resp.status_code == 200:
            self.start_state()
            # endtime = datetime.fromisoformat(base64.b64decode(resp.json()[0]["Value"]).decode())
            # compatibility python2
            endtime = datetime.strptime(base64.b64decode(resp.json()[0]["Value"]).decode(), "%Y-%m-%dT%H:%M:%S.%f")
            now = datetime.now()
            self.ids.progress.value = self.ids.progress.max - (endtime - now).total_seconds()
            self.ids.endtime.text = endtime.strftime("%H:%M")

        self.clock = Clock.schedule_interval(self.update, 1)
        wl.acquire()

    def leave(self):
        Clock.unschedule(self.clock)
        self.clock = None
        wl.release()

    def reset(self):
        self.state = "idle"
        self.running = False
        self.color.rgb = grey[:3]

    def interruption(self):
        if self.state == "interruption":
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "handled_interruption"}''')
            if self.interrupted_time:
                self.start()
            else:
                self.reset()
            return
        else:
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "interruption"}''')

        if self.state == "pomodoro" and not self.interrupted_time:
            self.interrupted_time = (self.ids.progress.value, self.ids.progress.max)

        self.reset()
        self.running = True
        self.state = "interruption"

        duration = 1
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = blue[:3]

    def exterruption(self):
        if self.state == "exterruption":
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "handled_ext_interruption"}''')
            if self.interrupted_time:
                self.start()
            else:
                self.reset()
            return
        else:
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "external_interruption"}''')

        if self.state == "pomodoro" and not self.interrupted_time:
            self.interrupted_time = (self.ids.progress.value, self.ids.progress.max)
        self.reset()
        self.running = True
        self.state = "exterruption"
        duration = 1
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = blue[:3]

    def twominutes(self):
        if self.state == "twominutes":
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "two_minutes_done"}''')
            if self.interrupted_time:
                self.start()
            else:
                self.reset()
            return
        else:
            requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "two_minutes"}''')

        if self.state == "pomodoro" and not self.interrupted_time:
            self.interrupted_time = (self.ids.progress.value, self.ids.progress.max)
        self.reset()

        self.running = True
        self.state = "twominutes"
        duration = 2
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = yellow[:3]

    def short_break(self):
        requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "start_break"}''')
        self.reset()

        self.running = True
        self.state = "break"
        duration = 4
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = green[:3]

    def long_break(self):
        requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='''{"message": "start_long_break"}''')
        self.reset()

        self.running = True
        self.state = "break"
        duration = 4
        self.ids.progress.max = duration * 60
        self.ids.progress.value = 0
        self.color.rgb = green[:3]

    def start_state(self):
        self.running = True
        self.state = "pomodoro"
        self.color.rgb = pink[:3]
        self.ids.progress.value = 0
        self.ids.progress.max = self.pomodoro_duration * 60

    def start(self):
        if self.running and self.state == "pomodoro":
            logger.debug("Already in a pomodoro")
            return
        self.start_state()
        if self.interrupted_time:
            (self.ids.progress.value, self.ids.progress.max) = self.interrupted_time
            remaining_seconds = self.ids.progress.max - self.ids.progress.value
            end = datetime.now() + timedelta(seconds=remaining_seconds)
            self.interrupted_time = None
            duration = remaining_seconds // 60
        else:
            duration = self.pomodoro_duration
            end = datetime.now() + timedelta(minutes=duration)
        self.ids.endtime.text = end.strftime("%H:%M")

        session_id = requests.put("http://192.168.1.245:8500/v1/session/create",
                     json={
                         "TTL": "{}m".format(duration),
                         "Behavior": "delete",
                         "Name": "pomodoro",
                         "Checks": [],
                     }).json()["ID"]
        logger.debug("Registering pomodoro for session {}".format(session_id))
        requests.delete("http://192.168.1.245:8500/v1/kv/pomodoro")
        requests.put("http://192.168.1.245:8500/v1/kv/pomodoro?acquire={}".format(session_id),
                     data=end.isoformat())
        requests.put("http://192.168.1.245:8500/v1/event/fire/pomodoro", data='{{"message": "start_pomodoro", "end": "{}"}}'.format(end.isoformat()))

class PomodoroApp(App):

    @staticmethod
    def populate(sm):
        try:
            sm.get_screen('pomodorotimer')
            return
        except ScreenManagerException:
            pass # not populated yet

        sm.add_widget(PomodoroScreen(name='pomodorotimer'))

    def build(self):
        sm = ScreenManager()
        self.populate(sm)
        return sm


def run():
    PomodoroApp().run()

Notes linking here