diff --git a/config.yaml b/config.yaml index d67fa98..688349f 100644 --- a/config.yaml +++ b/config.yaml @@ -1,13 +1,24 @@ 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: 40 + time-limit: 120 # 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 bb236a0..b3c2ec0 100644 --- a/lib/__init__.py +++ b/lib/__init__.py @@ -1,3 +1,4 @@ +import math import os, time import subprocess @@ -27,6 +28,7 @@ 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): @@ -45,7 +47,33 @@ class Usage(object): def release(self): '''Remove the timer usage to restart the counter''' - os.remove(self.file) + 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()) def isOffLimit(self): '''Is the timer usage off its limits''' @@ -57,13 +85,9 @@ class Usage(object): def isOffInterval(self): '''Is the timer usage expired''' - limit_interval = self.timer.limitInterval - if (limit_interval < 0): + usage_expiry = self.intervalResetTimestamp() + if usage_expiry is None: 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 @@ -99,18 +123,89 @@ 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 = 'ps aux | grep -v grep | grep "%s"' % app + cmd = 'pgrep -f "%s"' % app res = shell(cmd) - if len(res): + if res.strip(): running = True return running def block(self): for app in self.apps: - cmd = 'killall "%s"' % app + cmd = 'pkill -f "%s"' % app shell(cmd) @@ -158,3 +253,21 @@ 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 4d11f27..5931ad9 100644 --- a/readme.md +++ b/readme.md @@ -1,6 +1,8 @@ +> **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. +Control apps usage time with a command line application. Forked from [macedd/app-timer](https://github.com/macedd/app-timer) 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. @@ -11,6 +13,10 @@ 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) @@ -23,8 +29,25 @@ 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. @@ -42,3 +65,7 @@ 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 28ba190..b4ae358 100644 --- a/timer.py +++ b/timer.py @@ -1,7 +1,266 @@ +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: @@ -12,6 +271,7 @@ 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) @@ -19,8 +279,10 @@ 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():