gather_data

Compute various statistics and metrics from patch annotation data generated by the diff-annotate script (which has its source in the 'src/diffannotator/annotate.py' file).

This module / script saves extracted insights in a single file; currently only JSON output format is supported. Different subcommands use different schemas and save different data.

It is installed by the build process as diff-gather-stats script; this is defined in 'pyproject.toml' file.

This script provides the following subcommands:

• diff-gather-stats purpose-counter [--output JSON_FILE] DATASETS...: calculate count of purposes from all bugs in provided datasets,
• diff-gather-stats purpose-per-file [OPTIONS] RESULT_JSON DATASETS...: calculate per-file count of purposes from all bugs in provided datasets,
• diff-gather-stats lines-stats [OPTIONS] OUTPUT_FILE DATASETS...: calculate per-bug and per-file count of line types in provided datasets,
• diff-gather-stats timeline [OPTIONS] OUTPUT_FILE DATASETS...: calculate timeline of bugs with per-bug count of different types of lines.

Example (after installing the 'patchscope' package): diff-gather-stats --help

diff-gather-stats --annotations-dir='' purpose-counter         --output ~/example_annotations/tensorflow.purpose-counter.json         ~/example_annotations/tensorflow

diff-gather-stats --annotations-dir='' purpose-per-file         ~/example_annotations/tensorflow.purpose-per-file.json         ~/example_annotations/tensorflow/

diff-gather-stats --annotations-dir='' lines-stats         ~/example_annotations/tensorflow.lines-stats.json         ~/example_annotations/tensorflow/

 diff-gather-stats --annotations-dir='' timeline         --purpose-to-annotation=data         --purpose-to-annotation=documentation         --purpose-to-annotation=markup         --purpose-to-annotation=other         --purpose-to-annotation=project         --purpose-to-annotation=test         ~/example_annotations/tensorflow.timeline.purpose-to-type.json         ~/example_annotations/tensorflow/

AnnotatedBug

Annotated bug class

Source code in src/diffannotator/gather_data.py

class AnnotatedBug:
    """Annotated bug class"""

    def __init__(self, bug_dir: PathLike, annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR):
        """Constructor of the annotated bug

        :param bug_dir: path to the single bug
        """
        self._path = Path(bug_dir)
        self._annotations_path = self._path / annotations_dir

        try:
            self.annotations = [str(d.name) for d in self._annotations_path.iterdir()]
        except Exception as ex:
            print(f"Error in AnnotatedBug for '{self._path}': {ex}")

    def gather_data(self, bug_mapper: Callable[..., T],
                    datastructure_generator: Callable[[], T],
                    **mapper_kwargs) -> T:
        """
        Gathers dataset data via processing each file in current bug using AnnotatedFile class and provided functions

        :param bug_mapper: function to map bug to datastructure
        :param datastructure_generator: function to create empty datastructure to combine results via "+"
        :return: combined datastructure with all files data
        """
        combined_results = datastructure_generator()
        for annotation in self.annotations:
            if '...' in annotation:
                continue
            annotation_file_path = self._annotations_path / annotation
            annotation_file = AnnotatedFile(annotation_file_path)
            file_results = annotation_file.gather_data(bug_mapper, **mapper_kwargs)
            combined_results += file_results
        return combined_results

    def gather_data_dict(self, bug_dict_mapper: Callable[..., dict],
                         **mapper_kwargs) -> dict:
        """
        Gathers dataset data via processing each file in current bug using AnnotatedFile class and provided functions

        :param bug_dict_mapper: function to map diff to dictionary
        :return: combined dictionary of all diffs
        """
        combined_results = {}
        for annotation in self.annotations:
            if '...' in annotation:
                continue
            annotation_file_path = self._annotations_path / annotation
            annotation_file = AnnotatedFile(annotation_file_path)
            diff_file_results = annotation_file.gather_data(bug_dict_mapper, **mapper_kwargs)
            combined_results |= {str(annotation): diff_file_results}
        return combined_results

__init__(bug_dir, annotations_dir=Bug.DEFAULT_ANNOTATIONS_DIR)

Constructor of the annotated bug

:param bug_dir: path to the single bug

Source code in src/diffannotator/gather_data.py

def __init__(self, bug_dir: PathLike, annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR):
    """Constructor of the annotated bug

    :param bug_dir: path to the single bug
    """
    self._path = Path(bug_dir)
    self._annotations_path = self._path / annotations_dir

    try:
        self.annotations = [str(d.name) for d in self._annotations_path.iterdir()]
    except Exception as ex:
        print(f"Error in AnnotatedBug for '{self._path}': {ex}")

gather_data(bug_mapper, datastructure_generator, **mapper_kwargs)

Gathers dataset data via processing each file in current bug using AnnotatedFile class and provided functions

:param bug_mapper: function to map bug to datastructure :param datastructure_generator: function to create empty datastructure to combine results via "+" :return: combined datastructure with all files data

Source code in src/diffannotator/gather_data.py

def gather_data(self, bug_mapper: Callable[..., T],
                datastructure_generator: Callable[[], T],
                **mapper_kwargs) -> T:
    """
    Gathers dataset data via processing each file in current bug using AnnotatedFile class and provided functions

    :param bug_mapper: function to map bug to datastructure
    :param datastructure_generator: function to create empty datastructure to combine results via "+"
    :return: combined datastructure with all files data
    """
    combined_results = datastructure_generator()
    for annotation in self.annotations:
        if '...' in annotation:
            continue
        annotation_file_path = self._annotations_path / annotation
        annotation_file = AnnotatedFile(annotation_file_path)
        file_results = annotation_file.gather_data(bug_mapper, **mapper_kwargs)
        combined_results += file_results
    return combined_results

gather_data_dict(bug_dict_mapper, **mapper_kwargs)

Gathers dataset data via processing each file in current bug using AnnotatedFile class and provided functions

:param bug_dict_mapper: function to map diff to dictionary :return: combined dictionary of all diffs

Source code in src/diffannotator/gather_data.py

def gather_data_dict(self, bug_dict_mapper: Callable[..., dict],
                     **mapper_kwargs) -> dict:
    """
    Gathers dataset data via processing each file in current bug using AnnotatedFile class and provided functions

    :param bug_dict_mapper: function to map diff to dictionary
    :return: combined dictionary of all diffs
    """
    combined_results = {}
    for annotation in self.annotations:
        if '...' in annotation:
            continue
        annotation_file_path = self._annotations_path / annotation
        annotation_file = AnnotatedFile(annotation_file_path)
        diff_file_results = annotation_file.gather_data(bug_dict_mapper, **mapper_kwargs)
        combined_results |= {str(annotation): diff_file_results}
    return combined_results

AnnotatedBugDataset

Annotated bugs dataset class

Source code in src/diffannotator/gather_data.py

class AnnotatedBugDataset:
    """Annotated bugs dataset class"""

    def __init__(self, dataset_dir: PathLike):
        """Constructor of the annotated bug dataset.

        :param dataset_dir: path to the dataset
        """
        self._path = Path(dataset_dir)
        self.bugs: list[str] = []

        try:
            self.bugs = [str(d.name) for d in self._path.iterdir()
                         if d.is_dir()]
        except Exception as ex:
            print(f"Error in AnnotatedBugDataset for '{self._path}': {ex}")

    def gather_data(self, bug_mapper: Callable[..., T],
                    datastructure_generator: Callable[[], T],
                    annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR,
                    **mapper_kwargs) -> T:
        """
        Gathers dataset data via processing each bug using AnnotatedBug class and provided functions

        :param bug_mapper: function to map bug to datastructure
        :param datastructure_generator: function to create empty datastructure to combine results via "+"
        :param annotations_dir: subdirectory where annotations are; path
            to annotation in a dataset is <bug_id>/<annotations_dir>/<patch_data>.json
        :return: combined datastructure with all bug data
        """
        combined_results = datastructure_generator()

        print(f"Gathering data from bugs/patches in '{self._path}' directory.")
        for bug_id in tqdm.tqdm(self.bugs, desc='bug'):
            # TODO: log info / debug
            #print(bug_id)
            bug_path = self._path / bug_id
            bug = AnnotatedBug(bug_path, annotations_dir=annotations_dir)
            bug_results = bug.gather_data(bug_mapper, datastructure_generator, **mapper_kwargs)
            combined_results += bug_results

        return combined_results

    def gather_data_dict(self, bug_dict_mapper: Callable[..., dict],
                         annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR,
                         **mapper_kwargs) -> dict:
        """
        Gathers dataset data via processing each bug using AnnotatedBug class and provided function

        :param bug_dict_mapper: function to map diff to dictionary
        :param annotations_dir: subdirectory where annotations are; path
            to annotation in a dataset is <bug_id>/<annotations_dir>/<patch_data>.json
        :return: combined dictionary of all bugs
        """
        combined_results = {}
        for bug_id in tqdm.tqdm(self.bugs):
            print(bug_id)
            bug_path = self._path / bug_id
            bug = AnnotatedBug(bug_path, annotations_dir=annotations_dir)
            bug_results = bug.gather_data_dict(bug_dict_mapper, **mapper_kwargs)
            combined_results |= {bug_id: bug_results}
        return combined_results

    def gather_data_list(self, bug_to_dict_mapper: Callable[..., dict],
                         annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR,
                         **mapper_kwargs) -> list:
        """
        Gathers dataset data via processing each bug using AnnotatedBug class and provided function

        :param bug_to_dict_mapper: function to map diff annotations to dictionary
        :param annotations_dir: subdirectory where annotations are; path
            to annotation in a dataset is <bug_id>/<annotations_dir>/<patch_data>.json
        :return: list of bug dictionaries
        """
        combined_results = []
        for bug_id in tqdm.tqdm(self.bugs, desc="patchset", position=2, leave=False):
            bug_path = self._path / bug_id
            bug = AnnotatedBug(bug_path, annotations_dir=annotations_dir)
            bug_results = bug.gather_data_dict(bug_to_dict_mapper, **mapper_kwargs)
            # NOTE: could have used `+=` instead of `.append()`
            for patch_id, patch_data in bug_results.items():
                combined_results.append({
                    'bug_id': bug_id,
                    'patch_id': patch_id,
                    **patch_data
                })

        return combined_results

__init__(dataset_dir)

Constructor of the annotated bug dataset.

:param dataset_dir: path to the dataset

Source code in src/diffannotator/gather_data.py

def __init__(self, dataset_dir: PathLike):
    """Constructor of the annotated bug dataset.

    :param dataset_dir: path to the dataset
    """
    self._path = Path(dataset_dir)
    self.bugs: list[str] = []

    try:
        self.bugs = [str(d.name) for d in self._path.iterdir()
                     if d.is_dir()]
    except Exception as ex:
        print(f"Error in AnnotatedBugDataset for '{self._path}': {ex}")

gather_data(bug_mapper, datastructure_generator, annotations_dir=Bug.DEFAULT_ANNOTATIONS_DIR, **mapper_kwargs)

Gathers dataset data via processing each bug using AnnotatedBug class and provided functions

:param bug_mapper: function to map bug to datastructure :param datastructure_generator: function to create empty datastructure to combine results via "+" :param annotations_dir: subdirectory where annotations are; path to annotation in a dataset is //.json :return: combined datastructure with all bug data

Source code in src/diffannotator/gather_data.py

def gather_data(self, bug_mapper: Callable[..., T],
                datastructure_generator: Callable[[], T],
                annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR,
                **mapper_kwargs) -> T:
    """
    Gathers dataset data via processing each bug using AnnotatedBug class and provided functions

    :param bug_mapper: function to map bug to datastructure
    :param datastructure_generator: function to create empty datastructure to combine results via "+"
    :param annotations_dir: subdirectory where annotations are; path
        to annotation in a dataset is <bug_id>/<annotations_dir>/<patch_data>.json
    :return: combined datastructure with all bug data
    """
    combined_results = datastructure_generator()

    print(f"Gathering data from bugs/patches in '{self._path}' directory.")
    for bug_id in tqdm.tqdm(self.bugs, desc='bug'):
        # TODO: log info / debug
        #print(bug_id)
        bug_path = self._path / bug_id
        bug = AnnotatedBug(bug_path, annotations_dir=annotations_dir)
        bug_results = bug.gather_data(bug_mapper, datastructure_generator, **mapper_kwargs)
        combined_results += bug_results

    return combined_results

gather_data_dict(bug_dict_mapper, annotations_dir=Bug.DEFAULT_ANNOTATIONS_DIR, **mapper_kwargs)

Gathers dataset data via processing each bug using AnnotatedBug class and provided function

:param bug_dict_mapper: function to map diff to dictionary :param annotations_dir: subdirectory where annotations are; path to annotation in a dataset is //.json :return: combined dictionary of all bugs

Source code in src/diffannotator/gather_data.py

def gather_data_dict(self, bug_dict_mapper: Callable[..., dict],
                     annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR,
                     **mapper_kwargs) -> dict:
    """
    Gathers dataset data via processing each bug using AnnotatedBug class and provided function

    :param bug_dict_mapper: function to map diff to dictionary
    :param annotations_dir: subdirectory where annotations are; path
        to annotation in a dataset is <bug_id>/<annotations_dir>/<patch_data>.json
    :return: combined dictionary of all bugs
    """
    combined_results = {}
    for bug_id in tqdm.tqdm(self.bugs):
        print(bug_id)
        bug_path = self._path / bug_id
        bug = AnnotatedBug(bug_path, annotations_dir=annotations_dir)
        bug_results = bug.gather_data_dict(bug_dict_mapper, **mapper_kwargs)
        combined_results |= {bug_id: bug_results}
    return combined_results

gather_data_list(bug_to_dict_mapper, annotations_dir=Bug.DEFAULT_ANNOTATIONS_DIR, **mapper_kwargs)

Gathers dataset data via processing each bug using AnnotatedBug class and provided function

:param bug_to_dict_mapper: function to map diff annotations to dictionary :param annotations_dir: subdirectory where annotations are; path to annotation in a dataset is //.json :return: list of bug dictionaries

Source code in src/diffannotator/gather_data.py

def gather_data_list(self, bug_to_dict_mapper: Callable[..., dict],
                     annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR,
                     **mapper_kwargs) -> list:
    """
    Gathers dataset data via processing each bug using AnnotatedBug class and provided function

    :param bug_to_dict_mapper: function to map diff annotations to dictionary
    :param annotations_dir: subdirectory where annotations are; path
        to annotation in a dataset is <bug_id>/<annotations_dir>/<patch_data>.json
    :return: list of bug dictionaries
    """
    combined_results = []
    for bug_id in tqdm.tqdm(self.bugs, desc="patchset", position=2, leave=False):
        bug_path = self._path / bug_id
        bug = AnnotatedBug(bug_path, annotations_dir=annotations_dir)
        bug_results = bug.gather_data_dict(bug_to_dict_mapper, **mapper_kwargs)
        # NOTE: could have used `+=` instead of `.append()`
        for patch_id, patch_data in bug_results.items():
            combined_results.append({
                'bug_id': bug_id,
                'patch_id': patch_id,
                **patch_data
            })

    return combined_results

AnnotatedFile

Annotated single file in specific bug

Source code in src/diffannotator/gather_data.py

class AnnotatedFile:
    """Annotated single file in specific bug"""

    def __init__(self, file_path: PathLike):
        """Constructor of the annotated file of specific bug

        :param file_path: path to the single file
        """
        self._path = Path(file_path)

    def gather_data(self, bug_mapper: Callable[..., T],
                    **mapper_kwargs) -> T:
        """
        Retrieves data from file

        :param bug_mapper: function to map bug to datastructure
        :return: resulting datastructure
        """
        file_format = guess_format_version(self._path, warn_ambiguous=True)
        if file_format is None:
            logger.warning(f"Unknown annotation file format for '{self._path}'")
            file_format = JSONFormat.V1_5
        with self._path.open('r') as json_file:
            data = json.load(json_file)
            return bug_mapper(str(self._path), data,
                              data_format=file_format, **mapper_kwargs)

__init__(file_path)

Constructor of the annotated file of specific bug

:param file_path: path to the single file

Source code in src/diffannotator/gather_data.py

def __init__(self, file_path: PathLike):
    """Constructor of the annotated file of specific bug

    :param file_path: path to the single file
    """
    self._path = Path(file_path)

gather_data(bug_mapper, **mapper_kwargs)

Retrieves data from file

:param bug_mapper: function to map bug to datastructure :return: resulting datastructure

Source code in src/diffannotator/gather_data.py

def gather_data(self, bug_mapper: Callable[..., T],
                **mapper_kwargs) -> T:
    """
    Retrieves data from file

    :param bug_mapper: function to map bug to datastructure
    :return: resulting datastructure
    """
    file_format = guess_format_version(self._path, warn_ambiguous=True)
    if file_format is None:
        logger.warning(f"Unknown annotation file format for '{self._path}'")
        file_format = JSONFormat.V1_5
    with self._path.open('r') as json_file:
        data = json.load(json_file)
        return bug_mapper(str(self._path), data,
                          data_format=file_format, **mapper_kwargs)

ListAddedLinesResults

Example class to gather added lines from each hunk

Override this datastructure to gather results

Source code in src/diffannotator/gather_data.py

class ListAddedLinesResults:
    """Example class to gather added lines from each hunk

    Override this datastructure to gather results"""

    def __init__(self, processed_files, added_lines):
        self._processed_files = processed_files
        self._added_lines = added_lines

    def __add__(self, other):
        if isinstance(other, ListAddedLinesResults):
            new_instance = ListAddedLinesResults(
                self._processed_files + other._processed_files,
                self._added_lines + other._added_lines)
            return new_instance

    def __repr__(self):
        return f"ListAddedLinesResults(_processed_files={self._processed_files!r}, _added_lines={self._added_lines!r}"

    @staticmethod
    def default():
        """
        Constructs empty datastructure to work as 0 for addition via "+"

        :return: empty datastructure
        """
        return ListAddedLinesResults([], [])

    @staticmethod
    def create(file_path, data):
        """
        Override this function for single annotation handling

        :param file_path: path to processed file
        :param data: dictionary with annotations (file content)
        :return: datastructure instance
        """
        added_lines = []
        for hunk in data:
            print(hunk)
            print(data[hunk]['purpose'])
            if '+' in data[hunk]:
                added_lines.extend(data[hunk]['+'])
        return ListAddedLinesResults([file_path], added_lines)

create(file_path, data) staticmethod

Override this function for single annotation handling

:param file_path: path to processed file :param data: dictionary with annotations (file content) :return: datastructure instance

Source code in src/diffannotator/gather_data.py

@staticmethod
def create(file_path, data):
    """
    Override this function for single annotation handling

    :param file_path: path to processed file
    :param data: dictionary with annotations (file content)
    :return: datastructure instance
    """
    added_lines = []
    for hunk in data:
        print(hunk)
        print(data[hunk]['purpose'])
        if '+' in data[hunk]:
            added_lines.extend(data[hunk]['+'])
    return ListAddedLinesResults([file_path], added_lines)

default() staticmethod

Constructs empty datastructure to work as 0 for addition via "+"

:return: empty datastructure

Source code in src/diffannotator/gather_data.py

@staticmethod
def default():
    """
    Constructs empty datastructure to work as 0 for addition via "+"

    :return: empty datastructure
    """
    return ListAddedLinesResults([], [])

MaybeChanges

Bases: NamedTuple

Changes data, maybe intermixed with other data (see check_it)

Source code in src/diffannotator/gather_data.py

class MaybeChanges(NamedTuple):
    """Changes data, maybe intermixed with other data (see check_it)"""
    changes: dict[str, Union[dict, int]]
    check_it: bool = False

PurposeCounterResults

Example class to count purposes of each hunk

Override this datastructure to gather results

Source code in src/diffannotator/gather_data.py

class PurposeCounterResults:
    """Example class to count purposes of each hunk

    Override this datastructure to gather results"""

    def __init__(self, processed_files: list,
                 hunk_purposes: Counter[str], added_line_purposes: Counter[str], removed_line_purposes: Counter[str]):
        self._processed_files = processed_files
        self._hunk_purposes = hunk_purposes
        self._added_line_purposes = added_line_purposes
        self._removed_line_purposes = removed_line_purposes

    def __add__(self, other: 'PurposeCounterResults') -> 'PurposeCounterResults':
        if isinstance(other, PurposeCounterResults):
            new_instance = PurposeCounterResults(
                self._processed_files + other._processed_files,
                self._hunk_purposes + other._hunk_purposes,
                self._added_line_purposes + other._added_line_purposes,
                self._removed_line_purposes + other._removed_line_purposes)
            return new_instance

    def __repr__(self) -> str:
        return f"PurposeCounterResults(_processed_files={self._processed_files!r}, " \
               f"_hunk_purposes={self._hunk_purposes!r}, " \
               f"_added_line_purposes={self._added_line_purposes!r}, " \
               f"_removed_line_purposes)={self._removed_line_purposes!r})"

    def to_dict(self) -> dict:
        return {
            "processed_files": self._processed_files,
            "hunk_purposes": self._hunk_purposes,
            "added_line_purposes": self._added_line_purposes,
            "removed_line_purposes": self._removed_line_purposes,
        }

    @staticmethod
    def default() -> 'PurposeCounterResults':
        """
        Constructs empty datastructure to work as 0 for addition via "+"

        :return: empty datastructure
        """
        return PurposeCounterResults([], Counter(), Counter(), Counter())

    @staticmethod
    def create(file_path: str, data: dict,
               data_format: JSONFormat = JSONFormat.V1_5) -> 'PurposeCounterResults':
        """
        Override this function for single annotation handling

        :param file_path: path to processed file
        :param data: dictionary with annotations (file content)
        :param data_format: version of data schema used by annotation file
        :return: datastructure instance
        """
        file_purposes = Counter()
        added_line_purposes = Counter()
        removed_line_purposes = Counter()
        ## DEBUG
        #print(f"PurposeCounterResults.create({file_path=}, {data.keys()=}, {data_format=})")
        maybe_changes = _extract_maybe_changes(data, data_format=data_format)

        for change_file, change_data in maybe_changes.changes.items():
            if (maybe_changes.check_it and
                _is_not_changes(change_file, change_data,
                                data_format=data_format)):
                # this is not changed file information
                continue

            # TODO: log info / debug
            #print(f"PurposeCounterResults.create: {change_file=}, {change_data.keys()=}")
            file_purposes[change_data['purpose']] += 1
            if '+' in change_data:
                added_lines = change_data['+']
                for added_line in added_lines:
                    added_line_purposes[added_line['purpose']] += 1
            if '-' in change_data:
                removed_lines = change_data['-']
                for removed_line in removed_lines:
                    removed_line_purposes[removed_line['purpose']] += 1
        return PurposeCounterResults([file_path], file_purposes, added_line_purposes, removed_line_purposes)

create(file_path, data, data_format=JSONFormat.V1_5) staticmethod

Override this function for single annotation handling

:param file_path: path to processed file :param data: dictionary with annotations (file content) :param data_format: version of data schema used by annotation file :return: datastructure instance

Source code in src/diffannotator/gather_data.py

@staticmethod
def create(file_path: str, data: dict,
           data_format: JSONFormat = JSONFormat.V1_5) -> 'PurposeCounterResults':
    """
    Override this function for single annotation handling

    :param file_path: path to processed file
    :param data: dictionary with annotations (file content)
    :param data_format: version of data schema used by annotation file
    :return: datastructure instance
    """
    file_purposes = Counter()
    added_line_purposes = Counter()
    removed_line_purposes = Counter()
    ## DEBUG
    #print(f"PurposeCounterResults.create({file_path=}, {data.keys()=}, {data_format=})")
    maybe_changes = _extract_maybe_changes(data, data_format=data_format)

    for change_file, change_data in maybe_changes.changes.items():
        if (maybe_changes.check_it and
            _is_not_changes(change_file, change_data,
                            data_format=data_format)):
            # this is not changed file information
            continue

        # TODO: log info / debug
        #print(f"PurposeCounterResults.create: {change_file=}, {change_data.keys()=}")
        file_purposes[change_data['purpose']] += 1
        if '+' in change_data:
            added_lines = change_data['+']
            for added_line in added_lines:
                added_line_purposes[added_line['purpose']] += 1
        if '-' in change_data:
            removed_lines = change_data['-']
            for removed_line in removed_lines:
                removed_line_purposes[removed_line['purpose']] += 1
    return PurposeCounterResults([file_path], file_purposes, added_line_purposes, removed_line_purposes)

default() staticmethod

Constructs empty datastructure to work as 0 for addition via "+"

:return: empty datastructure

Source code in src/diffannotator/gather_data.py

@staticmethod
def default() -> 'PurposeCounterResults':
    """
    Constructs empty datastructure to work as 0 for addition via "+"

    :return: empty datastructure
    """
    return PurposeCounterResults([], Counter(), Counter(), Counter())

lines_stats(ctx, output_file, datasets, purpose_to_annotation=None)

Calculate per-bug and per-file count of line types in provided datasets

Each dataset is expected to be existing directory with the following structure:

<dataset_directory>/<bug_directory>/annotation/<patch_file>.json

Each dataset can consist of many BUGs, each BUG should include patch of annotated *diff.json file in 'annotation/' subdirectory.

Source code in src/diffannotator/gather_data.py

@app.command()
def lines_stats(
    ctx: typer.Context,
    output_file: Annotated[
        Path,
        typer.Argument(
            dir_okay=False,
            help="JSON file to write gathered results to"
        )
    ],
    datasets: Annotated[
        list[Path],
        typer.Argument(
            exists=True,
            file_okay=False,
            dir_okay=True,
            readable=True,
            writable=False,
            help="list of dirs with datasets to process"
        )
    ],
    # TODO: make it a common option, ~~or share it with lines_stats()~~
    purpose_to_annotation: Annotated[
        # see https://github.com/fastapi/typer/issues/387#issuecomment-1927465075
        Optional[list[click.Tuple]],
        typer.Option(
            help="""Mapping from file PURPOSE to line type LINE_TYPE.
                    Each line of such file will be treated as if it had given type.
                    As a shortcut, giving PURPOSE is the same as PURPOSE:PURPOSE.
                    Can be given multiple times.""",
            metavar="PURPOSE:LINE_TYPE|PURPOSE",
            # `parser` and `click_type` may not both be provided
            #click_type=click.Tuple([str, str]),
            parser=parse_colon_separated_pair,
        )
    ] = None,
) -> None:
    """Calculate per-bug and per-file count of line types in provided datasets

    Each dataset is expected to be existing directory with the following
    structure:

        <dataset_directory>/<bug_directory>/annotation/<patch_file>.json

    Each dataset can consist of many BUGs, each BUG should include patch
    of annotated *diff.json file in 'annotation/' subdirectory.
    """
    result = {}
    # often there is only one dataset
    for dataset in tqdm.tqdm(datasets, desc='dataset'):
        tqdm.tqdm.write(f"Dataset {dataset}")
        annotated_bugs = AnnotatedBugDataset(dataset)
        data = annotated_bugs.gather_data_dict(map_diff_to_lines_stats,
                                               annotations_dir=ctx.obj.annotations_dir,
                                               purpose_to_annotation=purpose_to_annotation)

        result[str(dataset)] = data

    save_result(result, output_file)

list_added_lines(datasets)

List added lines from all bugs in provided datasets

Each dataset is expected to be existing directory with the following structure:

<dataset_directory>/<bug_directory>/annotation/<patch_file>.json

Each dataset can consist of many bugs, each bug should include patch of annotated *diff.json file in 'annotation/' subdirectory.

Source code in src/diffannotator/gather_data.py

@app.command()
def list_added_lines(datasets: Annotated[
    list[Path],
    typer.Argument(
        exists=True,
        file_okay=False,
        dir_okay=True,
        readable=True,
        writable=False,
    )
]):
    """List added lines from all bugs in provided datasets

    Each dataset is expected to be existing directory with the following
    structure:

        <dataset_directory>/<bug_directory>/annotation/<patch_file>.json

    Each dataset can consist of many bugs, each bug should include patch
    of annotated *diff.json file in 'annotation/' subdirectory.
    """
    for dataset in datasets:
        print(f"Dataset {dataset}")
        annotated_bugs = AnnotatedBugDataset(dataset)
        data = annotated_bugs.gather_data(ListAddedLinesResults.create, ListAddedLinesResults.default)
        print(data)

map_diff_to_lines_stats(annotation_file_basename, annotation_data, data_format=JSONFormat.V1_5, purpose_to_annotation=None)

Mapper passed by line_stats() to *.gather_data_dict() method

It gathers information about file, and counts information about changed lines (in pre-image i.e. "-", in post-image i.e. "+",...).

:param annotation_file_basename: name of JSON file with annotation data :param annotation_data: parsed annotations data, retrieved from annotation_file_basename file. :param data_format: version of data schema used by annotation file

Source code in src/diffannotator/gather_data.py

def map_diff_to_lines_stats(annotation_file_basename: str,
                            annotation_data: dict,
                            data_format: JSONFormat = JSONFormat.V1_5,
                            purpose_to_annotation: Optional[list] = None) -> dict:
    """Mapper passed by line_stats() to *.gather_data_dict() method

    It gathers information about file, and counts information about
    changed lines (in pre-image i.e. "-", in post-image i.e. "+",...).

    :param annotation_file_basename: name of JSON file with annotation data
    :param annotation_data: parsed annotations data, retrieved from
        `annotation_file_basename` file.
    :param data_format: version of data schema used by annotation file
    """
    # Example fragment of annotation file:
    #
    # {
    #   "third_party/xla/xla/service/gpu/ir_emitter_unnested.cc": {
    #     "language": "C++",
    #     "type": "programming",
    #     "purpose": "programming",
    #     "+": [
    #       {
    #         "id": 4,
    #         "type": "code",
    #         "purpose": "programming",
    #         "tokens": […],
    #       },
    #       {"id":…},
    #     ],
    #     "-": […],
    #   },…
    # }
    result = {}
    # TODO: replace commented out DEBUG lines with logging (info or debug)
    # DEBUG
    #print(f"map_diff_to_lines_stats('{annotation_file_basename}', {{...}}):")
    # TODO: reduce code duplication wrt. purpose_to_annotation, if possible
    if purpose_to_annotation is None:
        purpose_to_annotation = []
    purpose_to_type_dict = dict([elem
                                 for elem in purpose_to_annotation
                                 if len(elem) == 2])

    maybe_changes = _extract_maybe_changes(annotation_data, data_format=data_format)

    for filename, file_data in maybe_changes.changes.items():
        if (maybe_changes.check_it and
            _is_not_changes(filename, file_data,
                            data_format=data_format)):
            # this is not changed file information
            continue

        # NOTE: each file should be present only once for given patch/commit
        if filename in result:
            print(f"Warning: '{filename}' file present more than once in '{annotation_file_basename}'")

        if filename not in result:
            # per-file data
            result[filename] = {
                key: value for key, value in file_data.items()
                if key in {"language", "type", "purpose"}
            }
            # DEBUG
            #print(f"  {result[filename]=}")
            # summary of per-line data
            result[filename].update({
                "+": Counter(),
                "-": Counter(),
                "+/-": Counter(),  # probably not necessary
            })
            # DEBUG
            #print(f"  {result[filename]=}")

        # DEBUG
        #print(f"  {type(file_data)=}, {file_data.keys()=}")

        for line_type in "+-":  # str used as iterable
            # diff might have removed lines, or any added lines
            if line_type not in file_data:
                continue

            for line in file_data[line_type]:
                result[filename][line_type]["count"] += 1  # count of added/removed lines

                for data_type in ["type", "purpose"]:  # ignore "id" and "tokens" fields
                    # handle --purpose-to-annotation PURPOSE:LINE_TYPE
                    if data_type == "type" and file_data["purpose"] in purpose_to_type_dict:
                        line_data = purpose_to_type_dict[file_data["purpose"]]
                    else:
                        line_data = line[data_type]

                    result[filename][line_type][f"{data_type}.{line_data}"] += 1
                    result[filename]["+/-"][f"{data_type}.{line_data}"] += 1

    return result

map_diff_to_purpose_dict(_diff_file_path, data, data_format=JSONFormat.V1_5)

Extracts file purposes of changed file in a diff annotation

Returns mapping from file name (of a changed file) to list (???) of file purposes for that file.

Example:

{
    'keras/engine/training_utils.py': ['programming'],
    'tests/keras/engine/test_training.py': ['test'],
}

:param _diff_file_path: file path containing diff, ignored :param data: dictionary loaded from file :param data_format: version of data schema used by annotation file :return: dictionary with file purposes

Source code in src/diffannotator/gather_data.py

def map_diff_to_purpose_dict(_diff_file_path: str, data: dict,
                             data_format: JSONFormat = JSONFormat. V1_5) -> dict:
    """Extracts file purposes of changed file in a diff annotation

    Returns mapping from file name (of a changed file) to list (???)
    of file purposes for that file.

    Example:

        {
            'keras/engine/training_utils.py': ['programming'],
            'tests/keras/engine/test_training.py': ['test'],
        }

    :param _diff_file_path: file path containing diff, ignored
    :param data: dictionary loaded from file
    :param data_format: version of data schema used by annotation file
    :return: dictionary with file purposes
    """
    result = {}
    maybe_changes = _extract_maybe_changes(data, data_format=data_format)

    for change_file, change_data in maybe_changes.changes.items():
        if (maybe_changes.check_it and
            _is_not_changes(change_file, change_data,
                            data_format=data_format)):
            # this is not changed file information
            continue

        #print(change_file)
        #print(change_data['purpose'])
        if change_file not in result:
            result[change_file] = []
        result[change_file].append(change_data['purpose'])

    #print(f"{_diff_file_path}:{result=}")
    return result

map_diff_to_timeline(annotation_file_basename, annotation_data, data_format=JSONFormat.V1_5, purpose_to_annotation=None)

Mapper passed by timeline() to *.gather_data_dict() method

It gathers information about file, and counts information about changed lines (in pre-image i.e. "-", in post-image i.e. "+",...).

:param annotation_file_basename: name of JSON file with annotation data :param annotation_data: parsed annotations data, retrieved from annotation_file_basename file. :param data_format: version of data schema used by annotation file :param purpose_to_annotation: list of pairs (, ) to treat each line of file with given purpose to have given type annotation.

Source code in src/diffannotator/gather_data.py

def map_diff_to_timeline(annotation_file_basename: str,
                         annotation_data: dict,
                         data_format: JSONFormat = JSONFormat.V1_5,
                         purpose_to_annotation: Optional[list] = None) -> dict:
    """Mapper passed by timeline() to *.gather_data_dict() method

    It gathers information about file, and counts information about
    changed lines (in pre-image i.e. "-", in post-image i.e. "+",...).

    :param annotation_file_basename: name of JSON file with annotation data
    :param annotation_data: parsed annotations data, retrieved from
        `annotation_file_basename` file.
    :param data_format: version of data schema used by annotation file
    :param purpose_to_annotation: list of pairs (<file purpose>, <line type annotation>)
        to treat each line of file with given purpose to have given type annotation.
    """
    # Example fragment of annotation file:
    #
    # {
    #   "commit_metadata": {
    #     "id": "e54746bdf7d5c831eabe4dcea76a7626f1de73df",
    #     "parents": ["93b61589b0bdb3845ee839e9c2a4e1adb06bd483"],
    #     "tree": "262d65e6c945adfa2d64bfe51e70c09d2e1d7d06",
    #     "author": {
    #       "author": "Patrick Cloke <clokep@users.noreply.github.com>",
    #       "name": "Patrick Cloke",
    #       "email": "clokep@users.noreply.github.com",
    #       "timestamp": 1611763190,
    #       "tz_info": "-0500"
    #     },
    #     "committer": {
    #       "committer": "GitHub <noreply@github.com>",
    #       "name": "GitHub",
    #       "email": "noreply@github.com",
    #       "timestamp": 1611763190,
    #       "tz_info": "-0500"
    #     },
    #   },
    #
    #   "n_files": 1,
    #   "hunk_span_src": 108,
    #   "hunk_span_dst": 114,
    #   "hunk_spread_src": 82,
    #   "hunk_spread_dst": 82,
    #   "n_hunks": 3,
    #   "n_lines_added": 9,
    #   "n_lines_removed": 3,
    #   "n_lines_all": 38,
    #   "n_mod": 2,
    #   "n_add": 7,
    #   "n_rem": 1,
    #   "n_groups": 5,
    #   "patch_size": 10,
    #   "groups_spread": 100,
    #   "spread_inner": 8,
    #
    #   "third_party/xla/xla/service/gpu/ir_emitter_unnested.cc": {
    #     "language": "C++",
    #     "type": "programming",
    #     "purpose": "programming",
    #     "+": [
    #       {
    #         "id": 4,
    #         "type": "code",
    #         "purpose": "programming",
    #         "tokens": […],
    #       },
    #       {"id":…},
    #     ],
    #     "-": […],
    #   },…
    # }

    # TODO: add logging (info or debug)
    result = Counter()
    per_commit_info = {}
    if purpose_to_annotation is None:
        purpose_to_annotation = []
    purpose_to_type_dict = dict([elem
                                 for elem in purpose_to_annotation
                                 if len(elem) == 2])
    #print(f"{purpose_to_annotation=}")
    #print(f"{purpose_to_type_dict=}")

    # gather diff metadata for v2
    if data_format == JSONFormat.V2:
        if 'diff_metadata' in annotation_data:
            for metric, count in annotation_data['diff_metadata'].items():
                per_commit_info[f"diff.{metric}"] = count

    # gather commit metadata for v1.5 and v2
    if ('commit_metadata' in annotation_data and
        (data_format == JSONFormat.V1_5 or
         data_format == JSONFormat.V2)):

        commit_metadata = annotation_data['commit_metadata']

        for metadata_key in ('author', 'committer'):
            if metadata_key not in commit_metadata:
                continue
            authorship_data = commit_metadata[metadata_key]
            for authorship_key in ('timestamp', 'tz_info', 'name', 'email'):
                if authorship_key in authorship_data:
                    per_commit_info[f"{metadata_key}.{authorship_key}"] = \
                        commit_metadata[metadata_key][authorship_key]

        if 'parents' in commit_metadata:
            per_commit_info['n_parents'] = len(commit_metadata['parents'])

        if data_format == JSONFormat.V1_5 and 'purpose' not in commit_metadata:
            # cannot be an ordinary file
            del annotation_data['commit_metadata']

    # extract changes data, required for v2
    if data_format == JSONFormat.V2:
        if 'changes' in annotation_data:
            changes_data = annotation_data['changes']
        else:
            changes_data = {}
    else:
        changes_data = annotation_data

    # gather summary data from all changed files
    for filename, file_data in changes_data.items():
        # handle the case of commit and diff metadata intermixed with changes data
        if data_format == JSONFormat.V1_5:
            # handle case of file named 'commit_metadata'
            # the commit metadata got extracted before the loop
            if filename == 'commit_metadata':
                # this might be changed file information, but commit metadata mixed in
                # at least for v1.5 annotations file format (file schema version)
                if 'purpose' not in file_data:
                    # commit metadata, skip processing it as a file
                    continue
                else:
                    # TODO: use logging
                    print(f"  warning: found file named 'commit_metadata' in {annotation_file_basename}")

            # handle the case of diff metadata intermixed with changes data
            if _is_diff_metadata(filename, file_data):
                per_commit_info[f"diff.{filename}"] = file_data
                # diff metadata, skip processing it as a file
                continue

        # NOTE: each file should be present only once for given patch/commit
        result['file_names'] += 1

        # gather per-file information, and aggregate it
        per_file_data = {
            key: value for key, value in file_data.items()
            if key in ("language", "type", "purpose")
        }
        per_file_data.update({
            "+": Counter(),
            "-": Counter(),
        })

        for line_type in "+-":  # str used as iterable
            # diff might have removed lines, or any added lines
            if line_type not in file_data:
                continue

            for line in file_data[line_type]:
                per_file_data[line_type]["count"] += 1  # count of added/removed lines

                for data_type in ["type", "purpose"]:  # ignore "id" and "tokens" fields
                    # handle --purpose-to-annotation PURPOSE:LINE_TYPE
                    if data_type == "type" and file_data["purpose"] in purpose_to_type_dict:
                        line_data = purpose_to_type_dict[file_data["purpose"]]
                    else:
                        line_data = line[data_type]

                    per_file_data[line_type][f"{data_type}.{line_data}"] += 1

        for key, value in per_file_data.items():
            if isinstance(value, (dict, defaultdict, Counter)):
                for sub_key, sub_value in value.items():
                    # don't expect anything deeper
                    result[f"{key}:{sub_key}"] += sub_value
            elif isinstance(value, int):
                result[key] += value
            else:
                result[f"{key}:{value}"] += 1

    result = dict(result, **per_commit_info)

    return result

parse_colon_separated_pair(value)

Parse colon separated pair 'A:B' string into ('A', 'B') tuple

As a shortcut, parse 'A' into ('A', 'A') tuple (if 'A' does not contain the colon ':').

Examples:

parse_colon_separated_pair('a:b') ('a', 'b') parse_colon_separated_pair('a') ('a','a') dict([parse_colon_separated_pair('key:value')])

:param value: string with colon-separated values, 'KEY:VALUE', or stringwithout colon, 'STR' :return: 2-element tuple with KEY and VALUE: ('KEY', 'VALUE'), or 2-element tuple ('STR', 'STR') if str does not include ':'

Source code in src/diffannotator/gather_data.py

def parse_colon_separated_pair(value: str) -> tuple[str, str]:
    """Parse colon separated pair 'A:B' string into ('A', 'B') tuple

    As a shortcut, parse 'A' into ('A', 'A') tuple
    (if 'A' does not contain the colon ':').

    Examples:

    >>> parse_colon_separated_pair('a:b')
    ('a', 'b')
    >>> parse_colon_separated_pair('a')
    ('a','a')
    >>> dict([parse_colon_separated_pair('key:value')])
    {'key': 'value'}

    :param value: string with colon-separated values, 'KEY:VALUE',
        or stringwithout colon, 'STR'
    :return: 2-element tuple with KEY and VALUE: ('KEY', 'VALUE'),
        or 2-element tuple ('STR', 'STR') if `str` does not include ':'
    """
    result = tuple(value.split(sep=':', maxsplit=2))  # type is Union[tuple[str], tuple[str, str]]
    if len(result) == 1:  # len it is always > 0
        # type of result variable is tuple[str]
        result = result * 2

    # noinspection PyTypeChecker
    return result  # type: tuple[str, str]

purpose_counter(ctx, datasets, result_json=None)

Calculate count of purposes from all bugs in provided datasets

Each dataset is expected to be existing directory with the following structure:

<dataset_directory>/<bug_directory>/annotation/<patch_file>.json

Each dataset can consist of many bugs, each bug should include patch of annotated *diff.json file in 'annotation/' subdirectory.

Source code in src/diffannotator/gather_data.py

@app.command()
def purpose_counter(
    ctx: typer.Context,
    datasets: Annotated[
        list[Path],
        typer.Argument(
            exists=True,
            file_okay=False,
            dir_okay=True,
            readable=True,
            writable=False
        )
    ],
    result_json: Annotated[
        Optional[Path],
        typer.Option(
            "--output", "-o",
            dir_okay=False,
            metavar="JSON_FILE",
            help="JSON file to write gathered results to",
        )
    ] = None,
) -> None:
    """Calculate count of purposes from all bugs in provided datasets

    Each dataset is expected to be existing directory with the following
    structure:

        <dataset_directory>/<bug_directory>/annotation/<patch_file>.json

    Each dataset can consist of many bugs, each bug should include patch
    of annotated *diff.json file in 'annotation/' subdirectory.
    """
    result = {}
    for dataset in datasets:
        print(f"Dataset {dataset}")
        annotated_bugs = AnnotatedBugDataset(dataset)
        data = annotated_bugs.gather_data(PurposeCounterResults.create,
                                          PurposeCounterResults.default,
                                          annotations_dir=ctx.obj.annotations_dir)
        result[dataset] = data

    if result_json is None:
        print(result)
    else:
        save_result({
                        str(key): value.to_dict()
                        for key, value in result.items()
                    },
                    result_json)

purpose_per_file(ctx, result_json, datasets)

Calculate per-file count of purposes from all bugs in provided datasets

Each dataset is expected to be existing directory with the following structure:

<dataset_directory>/<bug_directory>/annotation/<patch_file>.json

Each dataset can consist of many BUGs, each BUG should include patch of annotated *diff.json file in 'annotation/' subdirectory.

Source code in src/diffannotator/gather_data.py

@app.command()
def purpose_per_file(
    ctx: typer.Context,
    result_json: Annotated[
        Path,
        typer.Argument(
            dir_okay=False,
            help="JSON file to write gathered results to"
        )
    ],
    datasets: Annotated[
        list[Path],
        typer.Argument(
            exists=True,
            file_okay=False,
            dir_okay=True,
            readable=True,
            writable=False,
            help="list of dirs with datasets to process"
        )
    ],
) -> None:
    """Calculate per-file count of purposes from all bugs in provided datasets

    Each dataset is expected to be existing directory with the following
    structure:

        <dataset_directory>/<bug_directory>/annotation/<patch_file>.json

    Each dataset can consist of many BUGs, each BUG should include patch
    of annotated *diff.json file in 'annotation/' subdirectory.
    """
    result = {}
    for dataset in datasets:
        print(f"Dataset {dataset}")
        annotated_bugs = AnnotatedBugDataset(dataset)
        data = annotated_bugs.gather_data_dict(map_diff_to_purpose_dict,
                                               annotations_dir=ctx.obj.annotations_dir)
        result[str(dataset)] = data

    #print(result)
    save_result(result, result_json)

save_result(result, result_json)

Serialize result and save it in result_json JSON file

Side effects:

• prints progress information to stdout
• creates parent directory if it does not exist

:param result: data to serialize and save :param result_json: path to JSON file to save result to

Source code in src/diffannotator/gather_data.py

def save_result(result: Any, result_json: Path) -> None:
    """Serialize `result` and save it in `result_json` JSON file

    Side effects:

    - prints progress information to stdout
    - creates parent directory if it does not exist

    :param result: data to serialize and save
    :param result_json: path to JSON file to save `result` to
    """
    print(f"Saving results to '{result_json}' JSON file")

    # ensure that parent directory exists, so we can save the file
    parent_dir = result_json.parent
    if not parent_dir.exists():
        print(f"- creating '{parent_dir}' directory")
        parent_dir.mkdir(parents=True, exist_ok=True)  # exist_ok=True for race condition

    with result_json.open(mode='w') as result_f:  # type: SupportsWrite[str]
        json.dump(result, result_f, indent=4)

timeline(ctx, output_file, datasets, purpose_to_annotation=None)

Calculate timeline of bugs with per-bug count of different types of lines

For each bug (bugfix commit), compute the count of lines removed and added by the patch (commit) in all changed files, keeping separate counts for lines with different types, and (separately) with different purposes.

The gathered data is then saved in a format easy to load into dataframe.

Each DATASET is expected to be generated by annotating dataset or creating annotations from a repository, and should be an existing directory with the following structure:

<dataset_directory>/<bug_directory>/annotation/<patch_file>.json

Each dataset can consist of many BUGs, each BUG should include JSON file with its diff/patch annotations as *.json file in 'annotation/' subdirectory (by default).

Saves gathered timeline results to the OUTPUT_FILE.

Source code in src/diffannotator/gather_data.py

@app.command()
def timeline(
    ctx: typer.Context,  # common arguments like --annotations-dir
    output_file: Annotated[
        Path,
        typer.Argument(
            dir_okay=False,
            help="file to write gathered results to"
        )
    ],
    datasets: Annotated[
        list[Path],
        typer.Argument(
            exists=True,
            file_okay=False,
            dir_okay=True,
            readable=True,
            writable=False,
            help="list of dirs with datasets to process"
        )
    ],
    # TODO: make it a common option, or share it with lines_stats()
    purpose_to_annotation: Annotated[
        # see https://github.com/fastapi/typer/issues/387#issuecomment-1927465075
        Optional[list[click.Tuple]],
        typer.Option(
            help="""Mapping from file PURPOSE to line type LINE_TYPE.
                    Each line of such file will be treated as if it had given type.
                    As a shortcut, giving PURPOSE is the same as PURPOSE:PURPOSE.
                    Can be given multiple times.""",
            metavar="PURPOSE:LINE_TYPE|PURPOSE",
            # `parser` and `click_type` may not both be provided
            #click_type=click.Tuple([str, str]),
            parser=parse_colon_separated_pair,
        )
    ] = None,
) -> None:
    # TODO: extract common part of the command description
    """Calculate timeline of bugs with per-bug count of different types of lines

    For each bug (bugfix commit), compute the count of lines removed and added
    by the patch (commit) in all changed files, keeping separate counts for
    lines with different types, and (separately) with different purposes.

    The gathered data is then saved in a format easy to load into dataframe.

    Each DATASET is expected to be generated by annotating dataset or creating
    annotations from a repository, and should be an existing directory with
    the following structure:

        <dataset_directory>/<bug_directory>/annotation/<patch_file>.json

    Each dataset can consist of many BUGs, each BUG should include JSON
    file with its diff/patch annotations as *.json file in 'annotation/'
    subdirectory (by default).

    Saves gathered timeline results to the OUTPUT_FILE.
    """
    result = {}
    #print(f"{type(purpose_to_annotation)=}, {purpose_to_annotation=}")
    # TODO: check if there were values without ':' among --purpose-to-annotation

    # often there is only one dataset, therefore joblib support is not needed
    for dataset in tqdm.tqdm(datasets, desc='dataset'):
        tqdm.tqdm.write(f"Dataset {dataset}")
        annotated_bugs = AnnotatedBugDataset(dataset)
        data = annotated_bugs.gather_data_list(map_diff_to_timeline,
                                               annotations_dir=ctx.obj.annotations_dir,
                                               purpose_to_annotation=purpose_to_annotation)

        # sanity check
        if not data:
            tqdm.tqdm.write("  warning: no data extracted from this dataset")
        else:
            if 'author.timestamp' not in data[0]:
                tqdm.tqdm.write("  warning: dataset does not include time information")

        result[dataset.name] = data

    # TODO: support other formats than JSON
    save_result(result, output_file)