13898387797

Committed 17 Mar 2025 11:32AM UTC coverage: 89.7% (+5.4%) from 84.265%

Build # 13898387797

Build Type

Pull #101

github

Committed by

web-flow

Commit Message

Merge ea150c1e1 into 09a2093b5

Pull Request Pull Request #101: Rc 0.10.0

Run Details

1681 of 1855 new or added lines in 36 files covered. (90.62%)

9 existing lines in 4 files now uncovered.

4276 of 4767 relevant lines covered (89.7%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.34

/src/fk/core/timer_data.py

#  Flowkeeper - Pomodoro timer for power users and teams
#  Copyright (c) 2023 Constantine Kulak
#
#  This program is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 3 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
import datetime
import logging

from fk.core.abstract_data_item import AbstractDataItem, generate_uid
from fk.core.pomodoro import Pomodoro, POMODORO_TYPE_NORMAL
from fk.core.workitem import Workitem

logger = logging.getLogger(__name__)


class TimerData(AbstractDataItem['User']):
    # State is one of the following: work, rest, idle
    _state: str | None
    _pomodoro: Pomodoro | None
    _planned_duration: float
    _remaining_duration: float
    _last_state_change: datetime.datetime | None
    _next_state_change: datetime.datetime | None
    _last_date: datetime.date
    _pomodoro_in_series: int

    def __init__(self,
                 user: 'User',
                 create_date: datetime.datetime):
        super().__init__(uid=generate_uid(), parent=user, create_date=create_date)
        self._state = 'idle'
        self._pomodoro = None
        self._planned_duration = 0
        self._remaining_duration = 0
        self._last_state_change = None
        self._next_state_change = None
        self._last_date = datetime.date.today()
        self._pomodoro_in_series = 0

    def get_running_pomodoro(self) -> Pomodoro | None:
        return self._pomodoro

    def get_running_workitem(self) -> Workitem | None:
        return self._pomodoro.get_parent() if self._pomodoro is not None else None

    def get_state(self) -> str:
        return self._state

    def idle(self, when: datetime.datetime | None = None) -> None:
        self._state = 'idle'
        self._pomodoro = None
        self._planned_duration = 0
        self._remaining_duration = 0
        self._last_state_change = datetime.datetime.now(datetime.timezone.utc) if when is None else when
        self._next_state_change = None
        self.item_updated(when)
        logger.debug(f'Timer: Transitioned to idle at {self._last_state_change}')

    def work(self, pomodoro: Pomodoro, work_duration: float, when: datetime.datetime | None = None) -> None:
        self._state = 'work'
        self._pomodoro = pomodoro
        self._planned_duration = work_duration
        self._remaining_duration = work_duration
        self._last_state_change = datetime.datetime.now(datetime.timezone.utc) if when is None else when
        if work_duration and pomodoro.get_type() == POMODORO_TYPE_NORMAL:   # It might be 0 for tracker workitems
            self._next_state_change = self._last_state_change + datetime.timedelta(seconds=work_duration)
        else:
            self._next_state_change = None
        self.item_updated(when)
        logger.debug(f'Timer: Transitioned to work at {self._last_state_change}. '
                     f'Next state change: {self._next_state_change}')

    def _refresh_today(self, when: datetime.datetime | None = None):
        if when is None:
            when = datetime.datetime.now()
        today = when.date()
        if self._last_date != today:
            self._last_date = today
            self._pomodoro_in_series = 0
            logger.debug('Reset pomodoro series because we started a new day')

    def rest(self, rest_duration: float, when: datetime.datetime | None = None) -> None:
        self._state = 'rest'
        self._planned_duration = rest_duration
        self._remaining_duration = rest_duration
        self._last_state_change = datetime.datetime.now(datetime.timezone.utc) if when is None else when

        self._refresh_today(when)  # Reset the series, if needed

        if rest_duration > 0 and self._pomodoro.get_type() == POMODORO_TYPE_NORMAL:   # It might be 0 for long / unlimited breaks
            self._next_state_change = self._last_state_change + datetime.timedelta(seconds=rest_duration)
            self._pomodoro_in_series += 1  # Increment the number of completed pomodoros in series
        else:
            self._next_state_change = None
            self._pomodoro_in_series = 0  # We started a long break, can now reset the series

        self.item_updated(when)

        logger.debug(f'Timer: Transitioned to rest at {self._last_state_change}. '
                     f'Next state change: {self._next_state_change}. '
                     f'Pomodoros in series: {self._pomodoro_in_series}.')

    def is_working(self) -> bool:
        return self._state == 'work'

    def is_resting(self) -> bool:
        return self._state == 'rest'

    def is_idling(self) -> bool:
        return self._state == 'idle'

    def is_ticking(self) -> bool:
        return self._state != 'idle'

    def get_planned_duration(self) -> int:
        return self._planned_duration

    def get_remaining_duration(self) -> float:
        return self._remaining_duration

    def get_next_state_change(self) -> datetime.datetime | None:
        return self._next_state_change

    # There's no "when" parameter, because it assumes we call update_remaining_duration first
    def format_remaining_duration(self) -> str:
        remaining_duration = self.get_remaining_duration()     # This is always >= 0
        remaining_minutes = str(int(remaining_duration / 60)).zfill(2)
        remaining_seconds = str(int(remaining_duration % 60)).zfill(2)
        return f'{remaining_minutes}:{remaining_seconds}'

    def format_elapsed_work_duration(self, when: datetime.datetime | None = None) -> str:
        if self._pomodoro is None:
            return 'N/A'
        else:
            elapsed_duration = int(self._pomodoro.get_elapsed_work_duration(when))
            td = datetime.timedelta(seconds=elapsed_duration)
            return f'{td}'

    def format_elapsed_rest_duration(self, when: datetime.datetime | None = None) -> str:
        if self._pomodoro is None:
            return 'N/A'
        else:
            elapsed_duration = int(self._pomodoro.get_elapsed_rest_duration(when))
            td = datetime.timedelta(seconds=elapsed_duration)
            return f'{td}'

    def __str__(self) -> str:
        s = 'no pomodoro'
        if self._pomodoro is not None:
            s = 'workitem' + str(self._pomodoro.get_parent())
        return f'Timer for user {self.get_parent().get_identity()}, {s}. ' \
               f'State "{self._state}", ' \
               f'started at {self._last_state_change}, ' \
               f'next ring at {self._next_state_change}'

    def update_remaining_duration(self, when: datetime.datetime | None):
        if self._next_state_change is not None:
            now = when if when is not None else datetime.datetime.now(datetime.timezone.utc)
            if now < self._next_state_change:
                self._remaining_duration = (self._next_state_change - now).total_seconds()
            else:
                self._remaining_duration = 0
        self.item_updated(when)

    def to_dict(self) -> dict:
        d = super().to_dict()
        d['state'] = self._state
        d['pomodoro'] = self._pomodoro.get_uid() if self._pomodoro is not None else None
        d['planned_duration'] = self._planned_duration
        d['remaining_duration'] = self._remaining_duration
        d['last_state_change'] = self._last_state_change
        d['next_state_change'] = self._next_state_change
        return d

    def get_pomodoro_in_series(self) -> int:
        self._refresh_today()
        return self._pomodoro_in_series

1	# Flowkeeper - Pomodoro timer for power users and teams
2	# Copyright (c) 2023 Constantine Kulak
3	#
4	# This program is free software: you can redistribute it and/or modify
5	# it under the terms of the GNU General Public License as published by
6	# the Free Software Foundation; either version 3 of the License, or
7	# (at your option) any later version.
8	#
9	# This program is distributed in the hope that it will be useful,
10	# but WITHOUT ANY WARRANTY; without even the implied warranty of
11	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	# GNU General Public License for more details.
13	#
14	# You should have received a copy of the GNU General Public License
15	# along with this program. If not, see <https://www.gnu.org/licenses/>.
16	import datetime	1✔
17	import logging	1✔
18
19	from fk.core.abstract_data_item import AbstractDataItem, generate_uid	1✔
20	from fk.core.pomodoro import Pomodoro, POMODORO_TYPE_NORMAL	1✔
21	from fk.core.workitem import Workitem	1✔
22
23	logger = logging.getLogger(__name__)	1✔
24
25
26	class TimerData(AbstractDataItem['User']):	1✔
27	# State is one of the following: work, rest, idle
28	_state: str \| None	1✔
29	_pomodoro: Pomodoro \| None	1✔
30	_planned_duration: float	1✔
31	_remaining_duration: float	1✔
32	_last_state_change: datetime.datetime \| None	1✔
33	_next_state_change: datetime.datetime \| None	1✔
34	_last_date: datetime.date	1✔
35	_pomodoro_in_series: int	1✔
36
37	def __init__(self,	1✔
38	user: 'User',
39	create_date: datetime.datetime):
40	super().__init__(uid=generate_uid(), parent=user, create_date=create_date)	1✔
41	self._state = 'idle'	1✔
42	self._pomodoro = None	1✔
43	self._planned_duration = 0	1✔
44	self._remaining_duration = 0	1✔
45	self._last_state_change = None	1✔
46	self._next_state_change = None	1✔
47	self._last_date = datetime.date.today()	1✔
48	self._pomodoro_in_series = 0	1✔
49
50	def get_running_pomodoro(self) -> Pomodoro \| None:	1✔
51	return self._pomodoro	1✔
52
53	def get_running_workitem(self) -> Workitem \| None:	1✔
54	return self._pomodoro.get_parent() if self._pomodoro is not None else None	1✔
55
56	def get_state(self) -> str:	1✔
NEW 57	return self._state	×
58
59	def idle(self, when: datetime.datetime \| None = None) -> None:	1✔
60	self._state = 'idle'	1✔
61	self._pomodoro = None	1✔
62	self._planned_duration = 0	1✔
63	self._remaining_duration = 0	1✔
64	self._last_state_change = datetime.datetime.now(datetime.timezone.utc) if when is None else when	1✔
65	self._next_state_change = None	1✔
66	self.item_updated(when)	1✔
67	logger.debug(f'Timer: Transitioned to idle at {self._last_state_change}')	1✔
68
69	def work(self, pomodoro: Pomodoro, work_duration: float, when: datetime.datetime \| None = None) -> None:	1✔
70	self._state = 'work'	1✔
71	self._pomodoro = pomodoro	1✔
72	self._planned_duration = work_duration	1✔
73	self._remaining_duration = work_duration	1✔
74	self._last_state_change = datetime.datetime.now(datetime.timezone.utc) if when is None else when	1✔
75	if work_duration and pomodoro.get_type() == POMODORO_TYPE_NORMAL: # It might be 0 for tracker workitems	1✔
76	self._next_state_change = self._last_state_change + datetime.timedelta(seconds=work_duration)	1✔
77	else:
78	self._next_state_change = None	1✔
79	self.item_updated(when)	1✔
80	logger.debug(f'Timer: Transitioned to work at {self._last_state_change}. '	1✔
81	f'Next state change: {self._next_state_change}')
82
83	def _refresh_today(self, when: datetime.datetime \| None = None):	1✔
84	if when is None:	1✔
NEW 85	when = datetime.datetime.now()	×
86	today = when.date()	1✔
87	if self._last_date != today:	1✔
88	self._last_date = today	1✔
89	self._pomodoro_in_series = 0	1✔
90	logger.debug('Reset pomodoro series because we started a new day')	1✔
91
92	def rest(self, rest_duration: float, when: datetime.datetime \| None = None) -> None:	1✔
93	self._state = 'rest'	1✔
94	self._planned_duration = rest_duration	1✔
95	self._remaining_duration = rest_duration	1✔
96	self._last_state_change = datetime.datetime.now(datetime.timezone.utc) if when is None else when	1✔
97
98	self._refresh_today(when) # Reset the series, if needed	1✔
99
100	if rest_duration > 0 and self._pomodoro.get_type() == POMODORO_TYPE_NORMAL: # It might be 0 for long / unlimited breaks	1✔
101	self._next_state_change = self._last_state_change + datetime.timedelta(seconds=rest_duration)	1✔
102	self._pomodoro_in_series += 1 # Increment the number of completed pomodoros in series	1✔
103	else:
NEW 104	self._next_state_change = None	×
NEW 105	self._pomodoro_in_series = 0 # We started a long break, can now reset the series	×
106
107	self.item_updated(when)	1✔
108
109	logger.debug(f'Timer: Transitioned to rest at {self._last_state_change}. '	1✔
110	f'Next state change: {self._next_state_change}. '
111	f'Pomodoros in series: {self._pomodoro_in_series}.')
112
113	def is_working(self) -> bool:	1✔
114	return self._state == 'work'	1✔
115
116	def is_resting(self) -> bool:	1✔
117	return self._state == 'rest'	1✔
118
119	def is_idling(self) -> bool:	1✔
120	return self._state == 'idle'	1✔
121
122	def is_ticking(self) -> bool:	1✔
123	return self._state != 'idle'	1✔
124
125	def get_planned_duration(self) -> int:	1✔
126	return self._planned_duration	1✔
127
128	def get_remaining_duration(self) -> float:	1✔
129	return self._remaining_duration	1✔
130
131	def get_next_state_change(self) -> datetime.datetime \| None:	1✔
132	return self._next_state_change	1✔
133
134	# There's no "when" parameter, because it assumes we call update_remaining_duration first
135	def format_remaining_duration(self) -> str:	1✔
136	remaining_duration = self.get_remaining_duration() # This is always >= 0	1✔
137	remaining_minutes = str(int(remaining_duration / 60)).zfill(2)	1✔
138	remaining_seconds = str(int(remaining_duration % 60)).zfill(2)	1✔
139	return f'{remaining_minutes}:{remaining_seconds}'	1✔
140
141	def format_elapsed_work_duration(self, when: datetime.datetime \| None = None) -> str:	1✔
142	if self._pomodoro is None:	1✔
143	return 'N/A'	1✔
144	else:
145	elapsed_duration = int(self._pomodoro.get_elapsed_work_duration(when))	1✔
146	td = datetime.timedelta(seconds=elapsed_duration)	1✔
147	return f'{td}'	1✔
148
149	def format_elapsed_rest_duration(self, when: datetime.datetime \| None = None) -> str:	1✔
NEW 150	if self._pomodoro is None:	×
NEW 151	return 'N/A'	×
152	else:
NEW 153	elapsed_duration = int(self._pomodoro.get_elapsed_rest_duration(when))	×
NEW 154	td = datetime.timedelta(seconds=elapsed_duration)	×
NEW 155	return f'{td}'	×
156
157	def __str__(self) -> str:	1✔
158	s = 'no pomodoro'	1✔
159	if self._pomodoro is not None:	1✔
160	s = 'workitem' + str(self._pomodoro.get_parent())	1✔
161	return f'Timer for user {self.get_parent().get_identity()}, {s}. ' \	1✔
162	f'State "{self._state}", ' \
163	f'started at {self._last_state_change}, ' \
164	f'next ring at {self._next_state_change}'
165
166	def update_remaining_duration(self, when: datetime.datetime \| None):	1✔
167	if self._next_state_change is not None:	1✔
168	now = when if when is not None else datetime.datetime.now(datetime.timezone.utc)	1✔
169	if now < self._next_state_change:	1✔
170	self._remaining_duration = (self._next_state_change - now).total_seconds()	1✔
171	else:
172	self._remaining_duration = 0	1✔
173	self.item_updated(when)	1✔
174
175	def to_dict(self) -> dict:	1✔
176	d = super().to_dict()	1✔
177	d['state'] = self._state	1✔
178	d['pomodoro'] = self._pomodoro.get_uid() if self._pomodoro is not None else None	1✔
179	d['planned_duration'] = self._planned_duration	1✔
180	d['remaining_duration'] = self._remaining_duration	1✔
181	d['last_state_change'] = self._last_state_change	1✔
182	d['next_state_change'] = self._next_state_change	1✔
183	return d	1✔
184
185	def get_pomodoro_in_series(self) -> int:	1✔
NEW 186	self._refresh_today()	×
NEW 187	return self._pomodoro_in_series	×

flowkeeper-org / fk-desktop / 13898387797

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous