From ad01f60d1d56ce714ea975a9e6bc185e75853730 Mon Sep 17 00:00:00 2001 From: Alan Date: Thu, 13 May 2021 11:03:16 -0400 Subject: [PATCH 1/6] Add Job JSON, and Status one-shot new commands: `json` to output jobs in JSON format and `status` to do a one-time view instead of normal view mode --- manager.py | 10 ++-- plotmanager/library/utilities/commands.py | 63 +++++++++++++++++++++-- plotmanager/library/utilities/print.py | 57 +++++++++++++++++++- 3 files changed, 121 insertions(+), 9 deletions(-) diff --git a/manager.py b/manager.py index 6622fa7..47e1e62 100644 --- a/manager.py +++ b/manager.py @@ -1,7 +1,7 @@ import argparse from plotmanager.library.utilities.exceptions import InvalidArgumentException -from plotmanager.library.utilities.commands import start_manager, stop_manager, view, analyze_logs +from plotmanager.library.utilities.commands import start_manager, stop_manager, view, jsonout, analyze_logs parser = argparse.ArgumentParser(description='This is the central manager for Swar\'s Chia Plot Manager.') @@ -33,10 +33,14 @@ elif args.action == 'stop': stop_manager() elif args.action == 'view': - view() + view(False) +elif args.action == 'json': + jsonout() +elif args.action == 'status': + view(True) elif args.action == 'analyze_logs': analyze_logs() else: - error_message = 'Invalid action provided. The valid options are "start", "restart", "stop", "view", and ' \ + error_message = 'Invalid action provided. The valid options are "start", "restart", "stop", "view", "status", "json" and ' \ '"analyze_logs".' raise InvalidArgumentException(error_message) diff --git a/plotmanager/library/utilities/commands.py b/plotmanager/library/utilities/commands.py index 71a20a2..c5e70fe 100644 --- a/plotmanager/library/utilities/commands.py +++ b/plotmanager/library/utilities/commands.py @@ -12,13 +12,13 @@ from plotmanager.library.utilities.jobs import load_jobs from plotmanager.library.utilities.log import analyze_log_dates, check_log_progress, analyze_log_times from plotmanager.library.utilities.notifications import send_notifications -from plotmanager.library.utilities.print import print_view +from plotmanager.library.utilities.print import print_view, print_json from plotmanager.library.utilities.processes import is_windows, get_manager_processes, get_running_plots, start_process def start_manager(): if get_manager_processes(): - raise ManagerError('Manager is already running.') + raise ManagerError('Manger is already running.') directory = pathlib.Path().resolve() stateless_manager_path = os.path.join(directory, 'stateless-manager.py') @@ -68,8 +68,61 @@ def stop_manager(): raise TerminationException("Failed to stop manager processes.") print("Successfully stopped manager processes.") +def jsonout(): + chia_location, log_directory, config_jobs, manager_check_interval, max_concurrent, progress_settings, \ + notification_settings, debug_level, view_settings = get_config_info() + view_check_interval = view_settings['check_interval'] + analysis = {'files': {}} + drives = {'temp': [], 'temp2': [], 'dest': []} + jobs = load_jobs(config_jobs) + for job in jobs: + drive = job.temporary_directory.split('\\')[0] + drives['temp'].append(drive) + directories = { + 'dest': job.destination_directory, + 'temp2': job.temporary2_directory, + } + for key, directory_list in directories.items(): + if directory_list is None: + continue + if isinstance(directory_list, list): + for directory in directory_list: + drive = directory.split('\\')[0] + if drive in drives[key]: + continue + drives[key].append(drive) + else: + drive = directory_list.split('\\')[0] + if drive in drives[key]: + continue + drives[key].append(drive) -def view(): + running_work = {} + + analysis = analyze_log_dates(log_directory=log_directory, analysis=analysis) + jobs = load_jobs(config_jobs) + jobs, running_work = get_running_plots(jobs=jobs, running_work=running_work) + check_log_progress(jobs=jobs, running_work=running_work, progress_settings=progress_settings, + notification_settings=notification_settings, view_settings=view_settings) + print_json(jobs=jobs, running_work=running_work, analysis=analysis, drives=drives, + next_log_check=datetime.now() + timedelta(seconds=60), view_settings=view_settings) + + has_file = False + if len(running_work.values()) == 0: + has_file = True + for work in running_work.values(): + if not work.log_file: + continue + has_file = True + break + if not has_file: + print("Restarting view due to psutil going stale...") + system_args = [f'"{sys.executable}"'] + sys.argv + os.execv(sys.executable, system_args) + + exit() + +def view(status = False): chia_location, log_directory, config_jobs, manager_check_interval, max_concurrent, progress_settings, \ notification_settings, debug_level, view_settings = get_config_info() view_check_interval = view_settings['check_interval'] @@ -107,7 +160,9 @@ def view(): check_log_progress(jobs=jobs, running_work=running_work, progress_settings=progress_settings, notification_settings=notification_settings, view_settings=view_settings) print_view(jobs=jobs, running_work=running_work, analysis=analysis, drives=drives, - next_log_check=datetime.now() + timedelta(seconds=60), view_settings=view_settings) + next_log_check=datetime.now() + timedelta(seconds=60), view_settings=view_settings, viewstatus = status) + if status: + break time.sleep(view_check_interval) has_file = False if len(running_work.values()) == 0: diff --git a/plotmanager/library/utilities/print.py b/plotmanager/library/utilities/print.py index ef6cf4c..87218f0 100644 --- a/plotmanager/library/utilities/print.py +++ b/plotmanager/library/utilities/print.py @@ -1,10 +1,35 @@ import os import psutil +import json from datetime import datetime, timedelta from plotmanager.library.utilities.processes import get_manager_processes, get_chia_drives +def _get_json(pid, running_work, view_settings): + work = running_work[pid] + phase_times = work.phase_times + elapsed_time = (datetime.now() - work.datetime_start) + elapsed_time = pretty_print_time(elapsed_time.seconds) + phase_time_log = [] + for i in range(1, 5): + if phase_times.get(i): + phase_time_log.append(phase_times.get(i)) + + row = [ + work.job.name if work.job else '?', + work.k_size, + pid, + work.datetime_start.strftime(view_settings['datetime_format']), + elapsed_time, + work.current_phase, + ' / '.join(phase_time_log), + work.progress, + pretty_print_bytes(work.temp_file_size, 'gb', 0, " GiB"), + ] + + return (row) + def _get_row_info(pid, running_work, view_settings): work = running_work[pid] @@ -65,6 +90,31 @@ def pretty_print_table(rows): console.append(separator) return "\n".join(console) +def get_job_json(jobs, running_work, view_settings): + rows = [] + rows2 = [] + #headers = ['num', 'job', 'k', 'pid', 'start', 'elapsed_time', 'phase', 'phase_times', 'progress', 'temp_size'] + added_pids = [] + for job in jobs: + for pid in job.running_work: + if pid not in running_work: + continue + #rows.append(_get_json(pid, running_work, view_settings)) + rows2.append(_get_json(pid, running_work, view_settings)) + added_pids.append(pid) + for pid in running_work.keys(): + if pid in added_pids: + continue + rows2.append(_get_json(pid, running_work, view_settings)) + added_pids.append(pid) + rows2.sort(key=lambda x: (x[4]), reverse=True) + for i in range(len(rows2)): + rows2[i] = [str(i + 1)] + rows2[i] + #rows = [headers] + rows + rows2 = '{ "jobs": ' + json.dumps(rows2) + ' }' + print(rows2) + return(rows2) + #return pretty_print_table(rows2) def get_job_data(jobs, running_work, view_settings): rows = [] @@ -103,8 +153,10 @@ def get_drive_data(drives): str(chia_drives[drive_type].get(drive, '?'))]) return pretty_print_table(rows) +def print_json(jobs, running_work, analysis, drives, next_log_check, view_settings): + job_data2 = get_job_json(jobs=jobs, running_work=running_work, view_settings=view_settings) -def print_view(jobs, running_work, analysis, drives, next_log_check, view_settings): +def print_view(jobs, running_work, analysis, drives, next_log_check, view_settings, viewstatus): # Job Table job_data = get_job_data(jobs=jobs, running_work=running_work, view_settings=view_settings) @@ -136,5 +188,6 @@ def print_view(jobs, running_work, analysis, drives, next_log_check, view_settin print(f'Plots Completed Yesterday: {analysis["summary"].get(datetime.now().date() - timedelta(days=1), 0)}') print(f'Plots Completed Today: {analysis["summary"].get(datetime.now().date(), 0)}') print() - print(f"Next log check at {next_log_check.strftime('%Y-%m-%d %H:%M:%S')}") + if viewstatus == False: + print(f"Next log check at {next_log_check.strftime('%Y-%m-%d %H:%M:%S')}") print() From 081792bb7d3f72d2881ad4e58e1a86ce2988416f Mon Sep 17 00:00:00 2001 From: Swar Date: Sat, 15 May 2021 01:48:16 -0400 Subject: [PATCH 2/6] Renaming jsonout to json_output --- manager.py | 4 ++-- plotmanager/library/utilities/commands.py | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/manager.py b/manager.py index 8f4fd26..301f45f 100644 --- a/manager.py +++ b/manager.py @@ -1,7 +1,7 @@ import argparse from plotmanager.library.utilities.exceptions import InvalidArgumentException -from plotmanager.library.utilities.commands import start_manager, stop_manager, view, jsonout, analyze_logs +from plotmanager.library.utilities.commands import start_manager, stop_manager, view, json_output, analyze_logs parser = argparse.ArgumentParser(description='This is the central manager for Swar\'s Chia Plot Manager.') @@ -35,7 +35,7 @@ elif args.action == 'view': view(False) elif args.action == 'json': - jsonout() + json_output() elif args.action == 'status': view(True) elif args.action == 'analyze_logs': diff --git a/plotmanager/library/utilities/commands.py b/plotmanager/library/utilities/commands.py index e18fe07..016be8b 100644 --- a/plotmanager/library/utilities/commands.py +++ b/plotmanager/library/utilities/commands.py @@ -74,7 +74,8 @@ def stop_manager(): raise TerminationException("Failed to stop manager processes.") print("Successfully stopped manager processes.") -def jsonout(): + +def json_output(): chia_location, log_directory, config_jobs, manager_check_interval, max_concurrent, progress_settings, \ notification_settings, debug_level, view_settings = get_config_info() view_check_interval = view_settings['check_interval'] From 80da3669b742d3cbeff0eebca8e58509b2ae3794 Mon Sep 17 00:00:00 2001 From: Swar Date: Sat, 15 May 2021 01:49:47 -0400 Subject: [PATCH 3/6] Fixing typo --- plotmanager/library/utilities/commands.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plotmanager/library/utilities/commands.py b/plotmanager/library/utilities/commands.py index 016be8b..3f1bbf2 100644 --- a/plotmanager/library/utilities/commands.py +++ b/plotmanager/library/utilities/commands.py @@ -19,7 +19,7 @@ def start_manager(): if get_manager_processes(): - raise ManagerError('Manger is already running.') + raise ManagerError('Manager is already running.') directory = pathlib.Path().resolve() stateless_manager_path = os.path.join(directory, 'stateless-manager.py') From de0ba9b623703051aa6fd40fe941021e4c1867c9 Mon Sep 17 00:00:00 2001 From: Swar Date: Sat, 15 May 2021 01:56:04 -0400 Subject: [PATCH 4/6] Changing kwarg to loop --- manager.py | 4 ++-- plotmanager/library/utilities/commands.py | 7 ++++--- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/manager.py b/manager.py index 301f45f..299a311 100644 --- a/manager.py +++ b/manager.py @@ -33,11 +33,11 @@ elif args.action == 'stop': stop_manager() elif args.action == 'view': - view(False) + view() elif args.action == 'json': json_output() elif args.action == 'status': - view(True) + view(loop=False) elif args.action == 'analyze_logs': analyze_logs() else: diff --git a/plotmanager/library/utilities/commands.py b/plotmanager/library/utilities/commands.py index 3f1bbf2..c065f4c 100644 --- a/plotmanager/library/utilities/commands.py +++ b/plotmanager/library/utilities/commands.py @@ -77,7 +77,7 @@ def stop_manager(): def json_output(): chia_location, log_directory, config_jobs, manager_check_interval, max_concurrent, progress_settings, \ - notification_settings, debug_level, view_settings = get_config_info() + notification_settings, debug_level, view_settings, instrumentation_settings = get_config_info() view_check_interval = view_settings['check_interval'] analysis = {'files': {}} drives = {'temp': [], 'temp2': [], 'dest': []} @@ -129,7 +129,8 @@ def json_output(): exit() -def view(status=False): + +def view(loop=False): chia_location, log_directory, config_jobs, manager_check_interval, max_concurrent, progress_settings, \ notification_settings, debug_level, view_settings, instrumentation_settings = get_config_info() view_check_interval = view_settings['check_interval'] @@ -169,7 +170,7 @@ def view(status=False): instrumentation_settings=instrumentation_settings) print_view(jobs=jobs, running_work=running_work, analysis=analysis, drives=drives, next_log_check=datetime.now() + timedelta(seconds=view_check_interval), view_settings=view_settings, viewstatus=status) - if status: + if not loop: break time.sleep(view_check_interval) has_file = False From cb21a06b48a8c5bab83e36e194a48bb246c9a657 Mon Sep 17 00:00:00 2001 From: Swar Date: Sat, 15 May 2021 02:11:53 -0400 Subject: [PATCH 5/6] Cleanup --- plotmanager/library/utilities/commands.py | 32 +++++++++--------- plotmanager/library/utilities/print.py | 40 +++++++++++------------ 2 files changed, 36 insertions(+), 36 deletions(-) diff --git a/plotmanager/library/utilities/commands.py b/plotmanager/library/utilities/commands.py index c065f4c..7551e19 100644 --- a/plotmanager/library/utilities/commands.py +++ b/plotmanager/library/utilities/commands.py @@ -14,7 +14,8 @@ from plotmanager.library.utilities.log import analyze_log_dates, check_log_progress, analyze_log_times from plotmanager.library.utilities.notifications import send_notifications from plotmanager.library.utilities.print import print_view, print_json -from plotmanager.library.utilities.processes import is_windows, get_manager_processes, get_running_plots, start_process +from plotmanager.library.utilities.processes import is_windows, get_manager_processes, get_running_plots, \ + start_process, identify_drive, get_system_drives def start_manager(): @@ -78,7 +79,9 @@ def stop_manager(): def json_output(): chia_location, log_directory, config_jobs, manager_check_interval, max_concurrent, progress_settings, \ notification_settings, debug_level, view_settings, instrumentation_settings = get_config_info() - view_check_interval = view_settings['check_interval'] + + all_drives = get_system_drives() + analysis = {'files': {}} drives = {'temp': [], 'temp2': [], 'dest': []} jobs = load_jobs(config_jobs) @@ -92,14 +95,10 @@ def json_output(): for key, directory_list in directories.items(): if directory_list is None: continue - if isinstance(directory_list, list): - for directory in directory_list: - drive = directory.split('\\')[0] - if drive in drives[key]: - continue - drives[key].append(drive) - else: - drive = directory_list.split('\\')[0] + if not isinstance(directory_list, list): + directory_list = [directory_list] + for directory in directory_list: + drive = identify_drive(file_path=directory, drives=all_drives) if drive in drives[key]: continue drives[key].append(drive) @@ -108,11 +107,12 @@ def json_output(): analysis = analyze_log_dates(log_directory=log_directory, analysis=analysis) jobs = load_jobs(config_jobs) - jobs, running_work = get_running_plots(jobs=jobs, running_work=running_work) + jobs, running_work = get_running_plots(jobs=jobs, running_work=running_work, + instrumentation_settings=instrumentation_settings) check_log_progress(jobs=jobs, running_work=running_work, progress_settings=progress_settings, - notification_settings=notification_settings, view_settings=view_settings) - print_json(jobs=jobs, running_work=running_work, analysis=analysis, drives=drives, - next_log_check=datetime.now() + timedelta(seconds=60), view_settings=view_settings) + notification_settings=notification_settings, view_settings=view_settings, + instrumentation_settings=instrumentation_settings) + print_json(jobs=jobs, running_work=running_work, view_settings=view_settings) has_file = False if len(running_work.values()) == 0: @@ -126,7 +126,6 @@ def json_output(): print("Restarting view due to psutil going stale...") system_args = [f'"{sys.executable}"'] + sys.argv os.execv(sys.executable, system_args) - exit() @@ -169,7 +168,8 @@ def view(loop=False): notification_settings=notification_settings, view_settings=view_settings, instrumentation_settings=instrumentation_settings) print_view(jobs=jobs, running_work=running_work, analysis=analysis, drives=drives, - next_log_check=datetime.now() + timedelta(seconds=view_check_interval), view_settings=view_settings, viewstatus=status) + next_log_check=datetime.now() + timedelta(seconds=view_check_interval), + view_settings=view_settings, loop=loop) if not loop: break time.sleep(view_check_interval) diff --git a/plotmanager/library/utilities/print.py b/plotmanager/library/utilities/print.py index 1208bf6..bb00fb5 100644 --- a/plotmanager/library/utilities/print.py +++ b/plotmanager/library/utilities/print.py @@ -4,7 +4,8 @@ from datetime import datetime, timedelta -from plotmanager.library.utilities.processes import get_manager_processes, get_chia_drives +from plotmanager.library.utilities.processes import get_manager_processes + def _get_json(pid, running_work, view_settings): work = running_work[pid] @@ -28,7 +29,7 @@ def _get_json(pid, running_work, view_settings): pretty_print_bytes(work.temp_file_size, 'gb', 0, " GiB"), ] - return (row) + return row def _get_row_info(pid, running_work, view_settings): @@ -90,31 +91,28 @@ def pretty_print_table(rows): console.append(separator) return "\n".join(console) + def get_job_json(jobs, running_work, view_settings): rows = [] - rows2 = [] - #headers = ['num', 'job', 'k', 'pid', 'start', 'elapsed_time', 'phase', 'phase_times', 'progress', 'temp_size'] added_pids = [] for job in jobs: for pid in job.running_work: if pid not in running_work: continue - #rows.append(_get_json(pid, running_work, view_settings)) - rows2.append(_get_json(pid, running_work, view_settings)) + rows.append(_get_json(pid, running_work, view_settings)) added_pids.append(pid) for pid in running_work.keys(): if pid in added_pids: continue - rows2.append(_get_json(pid, running_work, view_settings)) + rows.append(_get_json(pid, running_work, view_settings)) added_pids.append(pid) - rows2.sort(key=lambda x: (x[4]), reverse=True) - for i in range(len(rows2)): - rows2[i] = [str(i + 1)] + rows2[i] - #rows = [headers] + rows - rows2 = '{ "jobs": ' + json.dumps(rows2) + ' }' - print(rows2) - return(rows2) - #return pretty_print_table(rows2) + rows.sort(key=lambda x: (x[4]), reverse=True) + for i in range(len(rows)): + rows[i] = [str(i + 1)] + rows[i] + jobs = dict(jobs=rows) + print(json.dumps(jobs, separators=(',', ':'))) + return rows + def get_job_data(jobs, running_work, view_settings): rows = [] @@ -219,10 +217,12 @@ def get_drive_data(drives, running_work, job_data): rows = [headers] + rows return pretty_print_table(rows) -def print_json(jobs, running_work, analysis, drives, next_log_check, view_settings): - job_data2 = get_job_json(jobs=jobs, running_work=running_work, view_settings=view_settings) -def print_view(jobs, running_work, analysis, drives, next_log_check, view_settings, viewstatus): +def print_json(jobs, running_work, view_settings): + get_job_json(jobs=jobs, running_work=running_work, view_settings=view_settings) + + +def print_view(jobs, running_work, analysis, drives, next_log_check, view_settings, loop): # Job Table job_data = get_job_data(jobs=jobs, running_work=running_work, view_settings=view_settings) @@ -254,6 +254,6 @@ def print_view(jobs, running_work, analysis, drives, next_log_check, view_settin print(f'Plots Completed Yesterday: {analysis["summary"].get(datetime.now().date() - timedelta(days=1), 0)}') print(f'Plots Completed Today: {analysis["summary"].get(datetime.now().date(), 0)}') print() - if viewstatus == False: - print(f"Next log check at {next_log_check.strftime('%Y-%m-%d %H:%M:%S')}") + if loop: + print(f"Next log check at {next_log_check.strftime('%Y-%m-%d %H:%M:%S')}") print() From 7ecf8eb80a503a9d13b1fcc4fbce25d8b4cb0f7c Mon Sep 17 00:00:00 2001 From: Swar Date: Sat, 15 May 2021 02:16:09 -0400 Subject: [PATCH 6/6] Updating description --- manager.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/manager.py b/manager.py index 299a311..7f448a3 100644 --- a/manager.py +++ b/manager.py @@ -7,11 +7,12 @@ parser = argparse.ArgumentParser(description='This is the central manager for Swar\'s Chia Plot Manager.') help_description = ''' -There are a few different actions that you can use: "start", "restart", "stop", "view", and "analyze_logs". "start" will -start a manager process. If one already exists, it will display an error message. "restart" will try to kill any -existing manager and start a new one. "stop" will terminate the manager, but all existing plots will be completed. -"view" can be used to display an updating table that will show the progress of your plots. Once a manager has started it -will always be running in the background unless an error occurs. This field is case-sensitive. +There are a few different actions that you can use: "start", "restart", "stop", "view", "status", "json", and +"analyze_logs". "start" will start a manager process. If one already exists, it will display an error message. +"restart" will try to kill any existing manager and start a new one. "stop" will terminate the manager, but all +existing plots will be completed. "view" can be used to display an updating table that will show the progress of your +plots. Once a manager has started it will always be running in the background unless an error occurs. This field is +case-sensitive. "analyze_logs" is a helper command that will scan all the logs in your log_directory to get your custom settings for the progress settings in the YAML file.