make sure we have enough samples before checking termination inactivity threshold

README.md

@@ -0,0 +1,77 @@
+# impatient
+
+A general purpose utility for estimating when a task will complete
+
+## Overview
+
+`impatient` is sort of like `pv` for tasks where you can't insert
+extra commands into a pipeline, and sort of like `progress` for tasks
+that aren't reading/writing a specific file on the local system.  If
+you give it any shell command that outputs a number (possibly with a
+K/M/G/T suffix), it will repeatedly run the command and show you how
+fast that number is changing.  If you give it the expected final value
+of the number, it'll also estimate how much longer you have to wait.
+It also has a specific option for tracking the total size of a
+file or directory as given by `du`.
+
+For example, here's `impatient` tracking an in-progress `zfs send` to
+a remote machine:
+
+```
+$ impatient -f "$(zfs list -pH -o used pool/home)" -c 'ssh remote-server zfs list -pH -o used remote-pool/backups/pool/home' -i 60
+109.8G - 2.8M/s - 263.7G total - 41.6% complete - 15h03m remaining - ETA 2021-11-23 11:34
+```
+
+Note that `-f` is given the value output by `zfs list`; we could
+give it a number directly, this just simplifies things in this
+particular scenario.  The `-p` flag for `zfs list` isn't strictly
+necessary, `impatient` can parse values like `50.7G`, but
+using the exact byte value provides more precision.
+
+`-c`, on the other hand, is given a quoted string containing a
+command, which will output the current progress value each time it's
+run. The command given to `-c` can be an arbitrary shell one-liner,
+use responsibly.
+
+## Usage
+
+```
+usage: impatient [-h] [-f FINAL] [-i INTERVAL] [-w WINDOW] [-d DECAY]
+                 [-V FRACTION] [-I COUNT] [-l PATH] (-p PATH | -c COMMAND)
+
+Display progress and time estimates for arbitrary tasks
+
+optional arguments:
+  -h, --help            show this help message and exit
+  -f FINAL, --final FINAL
+                        Expected final size/value at completion, optionally
+                        with a K/M/G/T suffix
+  -i INTERVAL, --interval INTERVAL
+                        Interval in seconds between samples (default 10)
+  -w WINDOW, --window WINDOW
+                        Number of samples to keep for the sliding window
+                        (default 100)
+  -d DECAY, --decay DECAY
+                        Decay coefficient for older samples (default 1). Must
+                        be between 0 and 1 inclusive. The lower this is, the
+                        more responsive/swingy the estimate will be.
+  -V FRACTION, --termination-value-threshold FRACTION
+                        Fraction of the expected final value that must be
+                        reached in order to terminate (default 0.95). Reaching
+                        this threshold is necessary but not sufficient, see
+                        also -I
+  -I COUNT, --termination-inactivity-threshold COUNT
+                        Number of consecutive unchanged samples that must be
+                        observed in order to terminate (default 10). Reaching
+                        this threshold is necessary but not sufficient, see
+                        also -V
+  -l PATH, --log-file PATH
+                        File to log the time series to. Will be saved as a
+                        csv, with columns for timestamp and for value. Will
+                        append data if the file already exists.
+  -p PATH, --path PATH  Track total disk usage of a given path
+  -c COMMAND, --command COMMAND
+                        Track value returned by a shell command; this should
+                        return a single number, optionally with a K/M/G/T
+                        suffix
+```

impatient

@@ -2,30 +2,49 @@
 
 import argparse
 import datetime
+import math
 import subprocess
 import sys
 import time
 
-WINDOW_SIZE = 100
-
 SUFFIXES = 'KMGT'
 
 parser = argparse.ArgumentParser(description='Display progress and time estimates for arbitrary tasks')
-parser.add_argument('-f', '--final', type=str, help='Expected final size/value at completion.')
+parser.add_argument('-f', '--final', type=str, help='Expected final size/value at completion, optionally with a K/M/G/T suffix')
 parser.add_argument('-i', '--interval', type=int, default=10, help='Interval in seconds between samples (default 10)')
+parser.add_argument('-w', '--window', type=int, default=100, help='Number of samples to keep for the sliding window (default 100)')
+parser.add_argument('-d', '--decay', type=float, default=1, help='Decay coefficient for older samples (default 1). Must be between 0 and 1 inclusive. The lower this is, the more responsive/swingy the estimate will be.')
+parser.add_argument('-V', '--termination-value-threshold', type=float, default=0.95, metavar='FRACTION', help='Fraction of the expected final value that must be reached in order to terminate (default 0.95). Reaching this threshold is necessary but not sufficient, see also -I')
+parser.add_argument('-I', '--termination-inactivity-threshold', type=int, default=10, metavar='COUNT', help='Number of consecutive unchanged samples that must be observed in order to terminate (default 10). Reaching this threshold is necessary but not sufficient, see also -V')
+parser.add_argument('-l', '--log-file', type=str, metavar='PATH', help='File to log the time series to. Will be saved as a csv, with columns for timestamp and for value. Will append data if the file already exists.')
 tracker_types = parser.add_mutually_exclusive_group(required=True)
 tracker_types.add_argument('-p', '--path', type=str, help='Track total disk usage of a given path')
-tracker_types.add_argument('-c', '--command', type=str, help='Track value returned by a shell command; this should return a single number, optionally followed by K/M/G/T')
+tracker_types.add_argument('-c', '--command', type=str, help='Track value returned by a shell command; this should return a single number, optionally with a K/M/G/T suffix')
 
 args = parser.parse_args()
+if args.interval < 1:
+    print('interval must be at least 1', file=sys.stderr)
+    exit(1)
+if args.window < 1:
+    print('window size must be at least 1', file=sys.stderr)
+    exit(1)
+if args.decay < 0 or args.decay > 1:
+    print('decay coefficient must be between 0 and 1', file=sys.stderr)
+    exit(1)
+if args.termination_value_threshold < 0 or args.termination_value_threshold > 1:
+    print('termination value threshold must be between 0 and 1', file=sys.stderr)
+    exit(1)
+if args.termination_inactivity_threshold < 0:
+    print('termination inactivity threshold must be nonnegative', file=sys.stderr)
+    exit(1)
 
 def parse_value(v):
     suffix = v[-1].upper()
     if suffix in SUFFIXES:
         exponent = 3*(SUFFIXES.find(suffix)+1)
-        return int(v[:-1])*(10**exponent)
+        return float(v[:-1])*(10**exponent)
     else:
-        return int(v)
+        return float(v)
 
 def display_value(v):
     suffix = ''
@@ -41,36 +60,57 @@ def display_timedelta(d):
     if d.days != 0:
         result += '{}d'.format(d.days)
     result += '{}h'.format(d.seconds // 3600)
-    result += '{}m'.format((d.seconds % 3600) // 60)
+    result += '{:02}m'.format((d.seconds % 3600) // 60)
     return result
 
 if args.path:
-    def current_val():
+    def current_val_helper():
         du = subprocess.run(['du', '--bytes', '--summarize', args.path], capture_output=True, text=True).stdout
         return parse_value(du.split()[0])
 else:
-    def current_val():
+    def current_val_helper():
         result = subprocess.run(args.command, shell=True, capture_output=True, text=True).stdout
         return parse_value(result.strip())
 
+if args.log_file:
+    log_file = open(args.log_file, mode='a', buffering=1)
+else:
+    log_file = None
+
+def current_val():
+    result = current_val_helper()
+    if log_file:
+        print('{},{}'.format(datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S"), result), file=log_file)
+    return result
+
 if args.final:
     final = parse_value(args.final)
+    if final <= 0:
+        print('final value must be positive', file=sys.stderr)
+        exit(1)
 else:
     final = None
 
-samples = [current_val()]
+deltas = []
+current = current_val()
 while True:
     time.sleep(args.interval)
 
-    current = current_val()
+    new = current_val()
 
-    samples.append(current)
-    if len(samples) < 2:
-        continue
-    if len(samples) > WINDOW_SIZE:
-        samples = samples[-WINDOW_SIZE:]
+    deltas.append(new - current)
+    current = new
+    if len(deltas) > args.window:
+        deltas = deltas[-args.window:]
 
-    rate = (current - samples[0])/((len(samples)-1)*args.interval)
+    total = 0
+    divisor = 0
+    coeff = 1
+    for d in deltas[::-1]:
+        total += coeff*d
+        divisor += coeff
+        coeff *= args.decay
+    rate = total/(divisor*args.interval)
 
     print('\033[2K\r', end='')
     print('{} - {}/s'.format(display_value(current), display_value(rate)), end='')
@@ -78,8 +118,13 @@ while True:
         fraction = current / final
         value_remaining = final - current
         print(' - {} total - {:.1f}% complete'.format(display_value(final), 100*fraction), end='')
-        if rate > 0:
-            time_remaining = datetime.timedelta(seconds=(value_remaining / rate))
+        seconds_remaining = (value_remaining / rate) if rate > 0 else math.inf
+        if seconds_remaining < 0:
+            print(' - unknown time remaining', end='')
+        elif seconds_remaining > 864000:
+            print(' - stalled', end='')
+        else:
+            time_remaining = datetime.timedelta(seconds=seconds_remaining)
             eta = datetime.datetime.now() + time_remaining
             print(' - {} remaining - ETA {}'.format(
                 display_timedelta(time_remaining),
@@ -87,6 +132,11 @@ while True:
             ), end='')
     sys.stdout.flush()
 
-    if final and current >= final:
+    if (
+        final
+        and current >= args.termination_value_threshold*final
+        and len(deltas) >= args.termination_inactivity_threshold
+        and all(d == 0 for d in deltas[-args.termination_inactivity_threshold:])
+    ):
         print()
         break