Skip to content
Snippets Groups Projects
repositories.py 39.7 KiB
Newer Older
  • Learn to ignore specific revisions
  •             # 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_master_logs()
                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):
    
    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.
            """
    
            new_branch_name = self.enter_context()
            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)
    
    
    
    
    class OutputRepo(BaseRepo):
        pass
    
    
    class JupyterInterfaceRepo(ProjectRepo):
        def commit(self, message: str, add_all=True):
            Notebook.save_ipynb()
            super().commit(message, add_all)
    
    
        def commit_nb_output(self, notebook_path: str, results_commit_message: str,
                             force_rerun=False, timeout=600, conversion_formats: list = None):
    
            if not Path(notebook_path).is_absolute():
                notebook_path = self.working_dir / notebook_path
    
            notebook = Notebook(notebook_path)
    
    
            if "nbconvert_call" not in sys.argv:
                # This is reached in the first call of this function
                self.enter_context(force=False)
    
                notebook.check_and_rerun_notebook(force_rerun=force_rerun,
                                                  timeout=timeout)
            else:
                # This is executed during the nbconvert call
                notebook.convert_ipynb(self.output_path, formats=conversion_formats)
                notebook.export_all_figures(self.output_path)
    
                self.exit_context(results_commit_message)