Add migration notice

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

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
+        usage_expiry = self.intervalResetTimestamp()
-        if (limit_interval < 0):
+        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,
+        }

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`.

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 '<tr><td>{name}</td><td>{usage}</td><td>{time_left}</td><td>{interval}</td><td>{active}</td><td>{blocked}</td><td>{apps}</td></tr>'.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 = '<tr><td colspan="7">No timers configured</td></tr>'
+    html_doc = """<!DOCTYPE html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <title>App Timer Status</title>
+  <style>
+    :root {{
+      color-scheme: dark;
+      --bg: #0f172a;
+      --surface: #1e293b;
+      --surface-alt: #162033;
+      --border: rgba(148, 163, 184, 0.3);
+      --text: #e2e8f0;
+      --muted: #94a3b8;
+      --accent: #38bdf8;
+    }}
+    * {{ box-sizing: border-box; }}
+    body {{
+      font-family: "Inter", "Segoe UI", system-ui, sans-serif;
+      padding: 2rem clamp(1rem, 3vw, 3rem);
+      background: radial-gradient(circle at top, rgba(56, 189, 248, 0.15), transparent 55%), var(--bg);
+      color: var(--text);
+      min-height: 100vh;
+      margin: 0;
+    }}
+    h1 {{
+      margin-top: 0;
+      font-weight: 600;
+      letter-spacing: 0.02em;
+    }}
+    p {{ color: var(--muted); margin-top: 0.3rem; }}
+    table {{
+      border-collapse: collapse;
+      width: 100%;
+      background: var(--surface);
+      border-radius: 0.75rem;
+      overflow: hidden;
+      box-shadow: 0 20px 50px rgba(15, 23, 42, 0.55);
+      border: 1px solid var(--border);
+    }}
+    th, td {{
+      border-bottom: 1px solid var(--border);
+      padding: 0.85rem 1rem;
+      text-align: left;
+    }}
+    th {{
+      background: rgba(15, 23, 42, 0.65);
+      font-size: 0.9rem;
+      text-transform: uppercase;
+      letter-spacing: 0.08em;
+      color: var(--muted);
+    }}
+    tbody tr:last-child td {{ border-bottom: none; }}
+    tr:nth-child(even) {{ background: var(--surface-alt); }}
+    tr:hover {{ background: rgba(56, 189, 248, 0.08); transition: background 0.15s ease-in-out; }}
+  </style>
+</head>
+<body>
+  <h1>App Timer Usage</h1>
+  <p>Updated {timestamp}</p>
+  <table>
+    <thead>
+      <tr>
+        <th>Category</th>
+        <th>Usage</th>
+        <th>Time Left</th>
+        <th>Recharge</th>
+        <th>Active</th>
+        <th>Blocked</th>
+        <th>Apps</th>
+      </tr>
+    </thead>
+    <tbody>
+      {rows}
+    </tbody>
+  </table>
+</body>
+</html>""".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 = '<html><body><h1>Error</h1><p>%s</p></body></html>' % 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():