Newer
Older
traceback.print_exc()
def exit_context(self, message):
"""
After running all project code, this prepares the commit of the results to the output repository. This includes
- Ensure no uncommitted changes in the project repository
- Stage all changes in the output repository
- Commit all changes in the output repository with the given commit message.
- Update the log files in the main branch of the output repository.
:param message:
Commit message for the output repository commit.
"""
self.test_for_uncommitted_changes()
if self._on_context_enter_commit_hash != self.current_commit_hash:
raise RuntimeError("Code has changed since starting the context. Don't do that.")
print("Completed computations, commiting results")
# This has to be using ._git.commit to raise an error if no results have been written.
commit_return = self.output_repo._git.commit("-m", message)
self.update_output_main_logs()
self.copy_data_to_cache("main")
print("\n" + commit_return + "\n")
except git.exc.GitCommandError as e:
self.output_repo.delete_active_branch_if_branch_is_empty()
raise e
finally:
self._is_in_context_manager = False
self._on_context_enter_commit_hash = None
@contextlib.contextmanager
def track_results(self, results_commit_message: str, debug=False, force=False):
Context manager to be used when running project code that produces output that should
be tracked in the output repository.
:param results_commit_message:
Commit message for the commit of the output repository.
:param debug:
Perform calculations without tracking output.
:param force:
Skip confirmation and force tracking of results.
if debug:
yield "debug"
return
new_branch_name = self.enter_context(force=force)
try:
yield new_branch_name
except Exception as e:
self.output_repo.delete_active_branch_if_branch_is_empty()
raise e
else:
self.exit_context(message=results_commit_message)
def capture_error(self, error):
print(traceback.format_exc())
write_lines_to_file(self.output_path / "error.stack", traceback.format_exc())
class OutputRepo(BaseRepo):
tsv_filepath = self.working_dir / "log.tsv"
with open(tsv_filepath, "r") as filehandle:
lines = filehandle.readlines()
line_array = [line.replace("\n", "").split("\t") for line in lines]
# Print
print(tabulate(line_array[1:], headers=line_array[0]))
self.checkout(self._most_recent_branch)
class JupyterInterfaceRepo(ProjectRepo):
def commit(self, message: str, add_all=True):
if "nbconvert_call" in sys.argv:
print("Not committing during nbconvert.")
return
Notebook.save_ipynb()
super().commit(message, add_all)
def commit_nb_output(self, notebook_path: str, results_commit_message: str,
force_rerun=True, timeout=600, conversion_formats: list = None):
if "nbconvert_call" in sys.argv:
return
# This is reached in the first call of this function
if not Path(notebook_path).is_absolute():
notebook_path = self.working_dir / notebook_path
notebook = Notebook(notebook_path)
with self.track_results(results_commit_message, force=True):
notebook.check_and_rerun_notebook(force_rerun=force_rerun,
timeout=timeout)
# This is executed after the nbconvert call
notebook.convert_ipynb(self.output_path, formats=conversion_formats)
notebook.export_all_figures(self.output_path)