Skip to content
Snippets Groups Projects
repositories.py 41.3 KiB
Newer Older
Ronald Jäpel's avatar
Ronald Jäpel committed
        """
        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.
Ronald Jäpel's avatar
Ronald Jäpel committed
        :param message:
            Commit message for the output repository commit.
        """
        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")
        self.output_repo.add(".")
            # 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.copy_data_to_cache()
            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()
            # self.remove_cached_files()
            self._is_in_context_manager = False
            self._on_context_enter_commit_hash = None
    def track_results(self, results_commit_message: str, debug=False, force=False):
Ronald Jäpel's avatar
Ronald Jäpel committed
        """
        Context manager to be used when running project code that produces output that should
Ronald Jäpel's avatar
Ronald Jäpel committed
        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.
Ronald Jäpel's avatar
Ronald Jäpel committed
        """
        new_branch_name = self.enter_context(force=force)
        try:
            yield new_branch_name
        except Exception as e:
Ronald Jäpel's avatar
Ronald Jäpel committed
            self.capture_error(e)
            self.output_repo.delete_active_branch_if_branch_is_empty()
            raise e
        else:
            self.exit_context(message=results_commit_message)

Ronald Jäpel's avatar
Ronald Jäpel committed
    def capture_error(self, error):
        print(traceback.format_exc())
        write_lines_to_file(self.output_path / "error.stack", traceback.format_exc())

Ronald Jäpel's avatar
Ronald Jäpel committed

    def print_data_log(self):
        self.checkout("main")
Ronald Jäpel's avatar
Ronald Jäpel committed

        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

    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:
Ronald Jäpel's avatar
Ronald Jäpel committed
            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)