diff --git a/config.yaml b/config.yaml index 688349f..d67fa98 100644 --- a/config.yaml +++ b/config.yaml @@ -1,24 +1,13 @@ check-interval: 1 -status-server: - host: 127.0.0.1 - port: 8091 timers: gaming: apps: + - retroarch - minecraft - - xmcl - - mcpelauncher-ui # how many minutes of usage - time-limit: 120 + time-limit: 40 # within how many hours of interval limit-interval: 12 - # warn n minutes before reaching the limit - warn-threshold: 5 - # commands run on GNOME to pre-warn/final-warn - warn-command: > - runuser -l randogoth -c 'DISPLAY=:0 DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus notify-send "App Timer" "Gaming time is almost up ({time_left_int} min left)" -u critical && canberra-gtk-play -i complete' - final-warn-command: > - runuser -l randogoth -c 'DISPLAY=:0 DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus notify-send "App Timer" "Final minute before shutdown ({time_left_seconds}s)" -u critical && canberra-gtk-play -i alarm-clock-elapsed' websurf: apps: - chrome diff --git a/lib/__init__.py b/lib/__init__.py index b3c2ec0..bb236a0 100644 --- a/lib/__init__.py +++ b/lib/__init__.py @@ -1,4 +1,3 @@ -import math import os, time import subprocess @@ -28,7 +27,6 @@ class Usage(object): super(Usage, self).__init__() self.timer = timer self.file = '%s/usage/%s' % (CONFIG_PATH, self.timer.name) - self.final_warning_sent = False @property def current(self): @@ -47,33 +45,7 @@ class Usage(object): def release(self): '''Remove the timer usage to restart the counter''' - if os.path.exists(self.file): - os.remove(self.file) - self.final_warning_sent = False - - def usageStartTimestamp(self): - '''Start timestamp (ctime) for the current interval''' - if not os.path.exists(self.file): - return None - return os.path.getctime(self.file) - - def intervalResetTimestamp(self): - '''Epoch timestamp when the usage interval resets''' - limit_interval = self.timer.limitInterval - if (limit_interval < 0): - return None - usage_started = self.usageStartTimestamp() - if usage_started is None: - return None - usage_expiry = usage_started + (limit_interval * 60 * 60) - return usage_expiry - - def timeUntilIntervalReset(self): - '''Seconds remaining until the interval resets''' - usage_expiry = self.intervalResetTimestamp() - if usage_expiry is None: - return None - return max(0, usage_expiry - time.time()) + os.remove(self.file) def isOffLimit(self): '''Is the timer usage off its limits''' @@ -85,9 +57,13 @@ class Usage(object): def isOffInterval(self): '''Is the timer usage expired''' - usage_expiry = self.intervalResetTimestamp() - if usage_expiry is None: + limit_interval = self.timer.limitInterval + if (limit_interval < 0): return False + if not os.path.exists(self.file): + return False + usage_started = os.path.getctime(self.file) + usage_expiry = usage_started + (limit_interval * 60 * 60) return time.time() >= usage_expiry @@ -123,89 +99,18 @@ class Timer(object): apps = [app.strip() for app in item_apps] return apps - @property - def warnThreshold(self): - threshold = self.item.get('warn-threshold', None) - if threshold is None: - return None - return float(threshold) - - @property - def warnCommand(self): - return self.item.get('warn-command') - - @property - def finalWarnCommand(self): - return self.item.get('final-warn-command') - - def _command_context(self, time_left): - minutes = max(time_left, 0) - seconds = minutes * 60 - return { - 'timer_name': self.name, - 'time_left': minutes, - 'time_left_int': int(math.ceil(minutes)), - 'time_left_floor': int(math.floor(minutes)), - 'time_left_seconds': int(math.ceil(seconds)), - } - - def _prepareCommand(self, command, context): - if isinstance(command, (list, tuple)): - command = ' '.join([str(part) for part in command]) - if not command: - return None - context = context or {} - try: - return command.format(**context) - except (KeyError, IndexError, ValueError) as exc: - print('Timer %s warning command format error (%s), using raw command' % (self.name, exc)) - return command - - def _runCommand(self, command, label, context=None): - cmd = self._prepareCommand(command, context) - if not cmd: - return - print('Timer %s running %s command: %s' % (self.name, label, cmd)) - shell(cmd) - - def maybeWarn(self, check_interval): - if self.timeLimit < 0: - return - current_usage = self.usage.current - time_left = self.timeLimit - current_usage - if time_left <= 0: - if self.usage.final_warning_sent: - self.usage.final_warning_sent = False - return - context = self._command_context(time_left) - if (self.warnThreshold is not None and self.warnCommand and - time_left <= self.warnThreshold): - self._runCommand(self.warnCommand, 'warning', context) - final_command = self.finalWarnCommand - if not final_command: - return - interval = float(check_interval) - if time_left <= interval: - if not self.usage.final_warning_sent: - self._runCommand(final_command, 'final warning', context) - self.usage.final_warning_sent = True - else: - # reset flag if the next loop is no longer the final one - if self.usage.final_warning_sent: - self.usage.final_warning_sent = False - def isRunning(self): running = False for app in self.apps: - cmd = 'pgrep -f "%s"' % app + cmd = 'ps aux | grep -v grep | grep "%s"' % app res = shell(cmd) - if res.strip(): + if len(res): running = True return running def block(self): for app in self.apps: - cmd = 'pkill -f "%s"' % app + cmd = 'killall "%s"' % app shell(cmd) @@ -253,21 +158,3 @@ class Config(object): '''How often the usage check should happen''' interval = self.data.get('check-interval', 1) return int(interval) - - @property - def statusServer(self): - '''Optional host/port overrides for the embedded status server''' - config = self.data.get('status-server') or {} - host = config.get('host') - port = config.get('port') - if port is None: - parsed_port = None - else: - try: - parsed_port = int(port) - except (TypeError, ValueError): - raise ValueError('status-server.port must be an integer') - return { - 'host': host, - 'port': parsed_port, - } diff --git a/readme.md b/readme.md index 5931ad9..4d11f27 100644 --- a/readme.md +++ b/readme.md @@ -1,8 +1,6 @@ -> **Migrated to [code.randogoth.com/randogoth/app-timer](https://code.randogoth.com/randogoth/app-timer)** - # App Timer (Parental Control) -Control apps usage time with a command line application. Forked from [macedd/app-timer](https://github.com/macedd/app-timer) +Control apps usage time with a command line application. The python daemon will watch for the apps executable and store usage time for them. It can block an app if the usage is off limits. @@ -13,10 +11,6 @@ I'm using this at a raspberry pi gaming station (retropie) to enforce usage limi ```yaml # how often should the script check for apps runtime? check-interval: 1 -# optional http status endpoint configuration -status-server: - host: 127.0.0.1 - port: 8090 # timers: list of apps that will be watched timers: # the app name (list key) @@ -29,25 +23,8 @@ timers: time-limit: 60 # within how many hours of interval limit-interval: 12 - # optional: warn N minutes before the limit is reached - warn-threshold: 5 - # run on every loop inside the warning window (anything executable) - warn-command: "notify-send 'App Timer' 'Only {time_left_int} minutes left'" - # run only on the final loop before apps are blocked - final-warn-command: "notify-send 'App Timer' 'Last minute before shutdown'" ``` -`warn-threshold` is measured in minutes and triggers the `warn-command` every loop while there is still time remaining. `final-warn-command` fires once on the last loop before the timer exceeds its limit (based on `check-interval`). All commands are executed through the shell, so they can show desktop notifications, play sounds, etc. You can interpolate values in the command string using `{timer_name}`, `{time_left}`, `{time_left_int}`, `{time_left_floor}`, and `{time_left_seconds}` (any Python format specifiers are supported, e.g. `{time_left:.1f}`). - -If the systemd service runs as `root`, wrap desktop commands so they execute inside your session: - -```yaml -warn-command: > - runuser -l randogoth -c 'DISPLAY=:0 DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus notify-send "App Timer" "{timer_name}: {time_left_int} minutes left"' -``` - -The status server shows timer usage at `http://host:port/` (defaults `127.0.0.1:8090`). Override the bind address or port via the `status-server` block, or keep using the `APP_TIMER_STATUS_HOST` / `APP_TIMER_STATUS_PORT` environment variables if you prefer exporting them in your service unit. - ## Installation There's a systemd service file ready for use. @@ -65,7 +42,3 @@ There's a systemd service file ready for use. # check the service sudo systemctl status app-timer ``` - -You must edit the file `config.yaml` for your timers setup. - -In the code there is some python 3 so this is required. For testing the timer and check it's outputs/configs, run `python3 timer.py`. diff --git a/timer.py b/timer.py index b4ae358..28ba190 100644 --- a/timer.py +++ b/timer.py @@ -1,266 +1,7 @@ -import html import os, time -import threading -from datetime import datetime -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from lib import Config -STATUS_CONTEXT = {'config': None} - - -def _format_duration(seconds): - if seconds is None: - return '—' - seconds = int(seconds) - hours, remainder = divmod(seconds, 3600) - minutes, _ = divmod(remainder, 60) - parts = [] - if hours: - parts.append('%dh' % hours) - if minutes or not parts: - parts.append('%dm' % minutes) - return ' '.join(parts) - - -def _format_minutes(minutes): - if minutes is None: - return '—' - try: - total_minutes = max(0, int(minutes)) - except (TypeError, ValueError): - return '—' - hours, remainder = divmod(total_minutes, 60) - parts = [] - if hours: - parts.append('%dh' % hours) - if remainder or not parts: - parts.append('%dm' % remainder) - return ' '.join(parts) - - -def _format_recharge(hours_value, reset_seconds): - if hours_value is None: - return '—' - if float(hours_value).is_integer(): - base = '%dh' % int(hours_value) - else: - base = '%.1fh' % float(hours_value) - if reset_seconds is None: - return 'Every %s' % base - return 'Every %s (ready in %s)' % (base, _format_duration(reset_seconds)) - - -def _format_usage(used, limit): - if limit is None: - return '%s (no limit)' % _format_minutes(used) - used = max(0, used) - return '%s / %s' % (_format_minutes(used), _format_minutes(limit)) - - -def _format_time_left(time_left, limit): - if limit is None or time_left is None: - return '—' - return _format_minutes(max(0, int(time_left))) - - -def _collect_timer_snapshot(timer): - usage_minutes = timer.usage.current - limit_minutes = timer.timeLimit - if limit_minutes < 0: - limit_minutes = None - time_left = None - if limit_minutes is not None: - time_left = max(limit_minutes - usage_minutes, 0) - interval_hours = timer.limitInterval - if interval_hours < 0: - interval_hours = None - reset_seconds = timer.usage.timeUntilIntervalReset() if interval_hours is not None else None - running = timer.isRunning() - blocked = bool(timer.usage.isOffLimit()) - return { - 'name': timer.name, - 'usage': usage_minutes, - 'limit': limit_minutes, - 'time_left': time_left, - 'interval_hours': interval_hours, - 'reset_in': reset_seconds, - 'running': running, - 'blocked': blocked, - 'apps': timer.apps, - } - - -def _render_timer_row(snapshot): - usage_text = _format_usage(snapshot['usage'], snapshot['limit']) - time_left_text = _format_time_left(snapshot['time_left'], snapshot['limit']) - interval_text = _format_recharge(snapshot['interval_hours'], snapshot['reset_in']) - active_text = 'Yes' if snapshot['running'] else 'No' - blocked_text = 'Yes' if snapshot['blocked'] else 'No' - apps = snapshot['apps'] or [] - apps_text = ', '.join(html.escape(app) for app in apps) if apps else '—' - return '{name}{usage}{time_left}{interval}{active}{blocked}{apps}'.format( - name=html.escape(snapshot['name']), - usage=usage_text, - time_left=time_left_text, - interval=interval_text, - active=active_text, - blocked=blocked_text, - apps=apps_text, - ) - - -def _render_status_page(config): - timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') - timers = config.timers if config else [] - rows = ''.join(_render_timer_row(_collect_timer_snapshot(timer)) for timer in timers) - if not rows: - rows = 'No timers configured' - html_doc = """ - - - - App Timer Status - - - -

App Timer Usage

-

Updated {timestamp}

- - - - - - - - - - - - - - {rows} - -
CategoryUsageTime LeftRechargeActiveBlockedApps
- -""".format(timestamp=timestamp, rows=rows) - return html_doc - - -class StatusRequestHandler(BaseHTTPRequestHandler): - def do_GET(self): - if self.path not in ('/', '/status'): - self.send_error(404, 'Not Found') - return - config = STATUS_CONTEXT.get('config') - try: - payload = _render_status_page(config).encode('utf-8') - except Exception as exc: - message = '

Error

%s

' % html.escape(str(exc)) - payload = message.encode('utf-8') - self.send_response(500) - else: - self.send_response(200) - self.send_header('Content-Type', 'text/html; charset=utf-8') - self.send_header('Content-Length', str(len(payload))) - self.end_headers() - self.wfile.write(payload) - - def log_message(self, format, *args): - # Keep HTTP logs concise - print("HTTP %s - %s" % (self.log_date_time_string(), format % args)) - - -def _resolve_status_server_bindings(config): - raw_server_config = getattr(config, 'statusServer', None) - config_data = getattr(config, 'data', None) - if raw_server_config: - server_config = raw_server_config or {} - elif isinstance(config_data, dict): - server_config = config_data.get('status-server') or {} - else: - server_config = {} - env_host = os.environ.get('APP_TIMER_STATUS_HOST') - env_port = os.environ.get('APP_TIMER_STATUS_PORT') - host = env_host or server_config.get('host') or '127.0.0.1' - if env_port is not None: - try: - port = int(env_port) - except ValueError: - raise ValueError('APP_TIMER_STATUS_PORT must be an integer') - else: - server_port = server_config.get('port') - if server_port is not None: - try: - port = int(server_port) - except (TypeError, ValueError): - raise ValueError('status-server.port must be an integer') - else: - port = 8090 - return host, port - - -def start_status_server(host, port): - try: - httpd = ThreadingHTTPServer((host, port), StatusRequestHandler) - except OSError as exc: - print('Failed to start status server on %s:%s (%s)' % (host, port, exc)) - return None - thread = threading.Thread(target=httpd.serve_forever, daemon=True) - thread.start() - print('Status server running on http://%s:%s' % (host, port)) - return httpd - def check_timers(config): '''Will check every timer setup for it's usage and limits''' for timer in config.timers: @@ -271,7 +12,6 @@ def check_timers(config): if not timer.isRunning(): continue print('Timer %s is running' % timer.name) - timer.maybeWarn(config.checkInterval) # check for off limit apps if timer.usage.isOffLimit(): print('Timer %s is off limit' % timer.name) @@ -279,10 +19,8 @@ def check_timers(config): # increment running apps timer timer.usage.increment(config.checkInterval) + config = Config() -STATUS_CONTEXT['config'] = config -status_host, status_port = _resolve_status_server_bindings(config) -status_server = start_status_server(status_host, status_port) while True: # check config changes if config.hasChanges():