Compare commits

...

6 Commits

Author SHA1 Message Date
9c711300a2 Add short form of --force-metadata, -F 2013-08-06 23:07:36 -04:00
74cf34e2ad Update sharon cleanup.cfg 2013-08-06 22:48:06 -04:00
120bf58b85 Support --nometa option for copy_one and copy_wildcard 2013-08-06 22:47:16 -04:00
c26daa9a3b Update crontab 2013-08-03 11:23:43 -04:00
6993f5c886 Fix process termination in nilm-pipewatch 2013-08-03 11:13:30 -04:00
Sharon NILM
dd69f3e51d Update process.sh 2013-08-02 23:19:14 -04:00
8 changed files with 37 additions and 17 deletions

View File

@@ -6,3 +6,4 @@ keep = 2w
[/sharon/sinefit] [/sharon/sinefit]
keep = 1y keep = 1y
decimated = false

View File

@@ -5,5 +5,5 @@
# Run NilmDB processing every 5 minutes # Run NilmDB processing every 5 minutes
*/5 * * * * chronic /home/nilm/data/process.sh */5 * * * * chronic /home/nilm/data/process.sh
# Check the capture process every minute # Try frequently restarting the capture process in case it died
*/1 * * * * chronic /home/nilm/data/capture.sh */5 * * * * chronic /home/nilm/data/capture.sh

View File

@@ -13,16 +13,20 @@ if ! flock -n -x 99 ; then
fi fi
trap 'rm -f "$LOCKFILE"' 0 trap 'rm -f "$LOCKFILE"' 0
# sinefit on phase A voltage # redirect stdout/stderr to log, but keep it on the console too
exec > >(tee /home/nilm/data/process.log)
exec 2> >(tee -a /home/nilm/data/process.log >&2)
echo "sinefit on phase A voltage"
nilm-sinefit -c 5 /sharon/raw /sharon/sinefit nilm-sinefit -c 5 /sharon/raw /sharon/sinefit
# prep on A, B, C with appropriate rotations echo "prep on A, B, C with appropriate rotations"
nilm-prep -c 1 -r 0 /sharon/raw /sharon/sinefit /sharon/prep-a nilm-prep -c 1 -r 0 /sharon/raw /sharon/sinefit /sharon/prep-a
nilm-prep -c 2 -r 120 /sharon/raw /sharon/sinefit /sharon/prep-b nilm-prep -c 2 -r 120 /sharon/raw /sharon/sinefit /sharon/prep-b
nilm-prep -c 3 -r 240 /sharon/raw /sharon/sinefit /sharon/prep-c nilm-prep -c 3 -r 240 /sharon/raw /sharon/sinefit /sharon/prep-c
# decimate raw and prep data echo "decimate raw and prep data"
nilm-decimate-auto /sharon/raw /sharon/prep* nilm-decimate-auto /sharon/raw /sharon/prep*
# run cleanup echo "run cleanup"
nilm-cleanup --yes /home/nilm/data/cleanup.cfg nilm-cleanup --yes /home/nilm/data/cleanup.cfg

View File

@@ -12,6 +12,8 @@ import sys
def main(argv = None): def main(argv = None):
f = nilmtools.filter.Filter() f = nilmtools.filter.Filter()
parser = f.setup_parser("Copy a stream") parser = f.setup_parser("Copy a stream")
parser.add_argument('-n', '--nometa', action='store_true',
help="Don't copy or check metadata")
# Parse arguments # Parse arguments
try: try:
@@ -25,8 +27,9 @@ def main(argv = None):
raise SystemExit(1) raise SystemExit(1)
# Copy metadata # Copy metadata
meta = f.client_src.stream_get_metadata(f.src.path) if not args.nometa:
f.check_dest_metadata(meta) meta = f.client_src.stream_get_metadata(f.src.path)
f.check_dest_metadata(meta)
# Copy all rows of data using the faster Numpy interfaces # Copy all rows of data using the faster Numpy interfaces
extractor = NumpyClient(f.src.url).stream_extract_numpy extractor = NumpyClient(f.src.url).stream_extract_numpy

View File

@@ -16,6 +16,8 @@ def main(argv = None):
Example: %(prog)s -u http://host1/nilmdb -U http://host2/nilmdb /sharon/* Example: %(prog)s -u http://host1/nilmdb -U http://host2/nilmdb /sharon/*
""", skip_paths = True) """, skip_paths = True)
parser.add_argument('-n', '--nometa', action='store_true',
help="Don't copy or check metadata")
parser.add_argument("path", action="store", nargs="+", parser.add_argument("path", action="store", nargs="+",
help='Wildcard paths to copy') help='Wildcard paths to copy')
args = parser.parse_args(argv) args = parser.parse_args(argv)
@@ -56,6 +58,8 @@ def main(argv = None):
new_argv.extend(["--end", "@" + repr(args.end)]) new_argv.extend(["--end", "@" + repr(args.end)])
if args.dry_run: if args.dry_run:
new_argv.extend(["--dry-run"]) new_argv.extend(["--dry-run"])
if args.nometa:
new_argv.extend(["--nometa"])
if args.force_metadata: if args.force_metadata:
new_argv.extend(["--force-metadata"]) new_argv.extend(["--force-metadata"])
new_argv.extend([stream[0], stream[0]]) new_argv.extend([stream[0], stream[0]])

View File

@@ -21,9 +21,9 @@ def main(argv = None):
parser.add_argument("-u", "--url", action="store", parser.add_argument("-u", "--url", action="store",
default="http://localhost/nilmdb/", default="http://localhost/nilmdb/",
help="NilmDB server URL (default: %(default)s)") help="NilmDB server URL (default: %(default)s)")
parser.add_argument('-f', '--factor', action='store', default=4, type=int, parser.add_argument("-f", "--factor", action="store", default=4, type=int,
help='Decimation factor (default: %(default)s)') help='Decimation factor (default: %(default)s)')
parser.add_argument("--force-metadata", action="store_true", parser.add_argument("-F", "--force-metadata", action="store_true",
default = False, default = False,
help="Force metadata changes if the dest " help="Force metadata changes if the dest "
"doesn't match") "doesn't match")

View File

@@ -206,7 +206,7 @@ class Filter(object):
default = False, default = False,
help="Just print intervals that would be " help="Just print intervals that would be "
"processed") "processed")
group.add_argument("--force-metadata", action="store_true", group.add_argument("-F", "--force-metadata", action="store_true",
default = False, default = False,
help="Force metadata changes if the dest " help="Force metadata changes if the dest "
"doesn't match") "doesn't match")

View File

@@ -84,13 +84,16 @@ def pipewatch(args):
bufsize = -1, close_fds = True, bufsize = -1, close_fds = True,
stdin = devnull, stdin = devnull,
stdout = subprocess.PIPE, stdout = subprocess.PIPE,
stderr = None) stderr = None,
preexec_fn = os.setpgrp)
consumer = subprocess.Popen(args.consumer, shell = True, consumer = subprocess.Popen(args.consumer, shell = True,
bufsize = -11, close_fds = True, bufsize = -11, close_fds = True,
stdin = subprocess.PIPE, stdin = subprocess.PIPE,
stdout = None, stderr = None) stdout = None,
stderr = None,
preexec_fn = os.setpgrp)
queue = Queue.Queue(maxsize = 32) queue = Queue.Queue(maxsize = 4)
reader = threading.Thread(target = reader_thread, reader = threading.Thread(target = reader_thread,
args = (queue, generator.stdout.fileno())) args = (queue, generator.stdout.fileno()))
reader.start() reader.start()
@@ -125,16 +128,21 @@ def pipewatch(args):
return proc.poll() return proc.poll()
try: try:
if poll_timeout(proc, 0.5) is None: if poll_timeout(proc, 0.5) is None:
proc.terminate() os.killpg(proc.pid, signal.SIGTERM)
if poll_timeout(proc, 0.5) is None: if poll_timeout(proc, 0.5) is None:
proc.kill() os.killpg(proc.pid, signal.SIGKILL)
except OSError: except OSError:
pass pass
return poll_timeout(proc, 0.5) return poll_timeout(proc, 0.5)
# Wait for them to die, or kill them # Wait for them to die, or kill them
gret = kill(generator)
cret = kill(consumer) cret = kill(consumer)
gret = kill(generator)
# Consume all remaining data in the queue until the reader
# and watcher threads are done
while reader.is_alive() or watcher.is_alive():
queue.get(True, 0.1)
fprintf(sys.stderr, "pipewatch: generator returned %d, " + fprintf(sys.stderr, "pipewatch: generator returned %d, " +
"consumer returned %d\n", gret, cret) "consumer returned %d\n", gret, cret)