Skip to content

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 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
class AnnotatedBug:
    """Annotated bug class"""

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

        Parameters
        ----------
        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

        Parameters
        ----------
        bug_mapper
            function to map bug to datastructure
        datastructure_generator
            function to create empty datastructure to combine results
            via "+"

        Returns
        -------
        T
            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

        Parameters
        ----------
        bug_dict_mapper
            function to map diff to dictionary

        Returns
        -------
        dict
            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__ 

__init__(
    bug_dir: PathLike,
    annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR,
)

Constructor of the annotated bug

PARAMETER DESCRIPTION
bug_dir

path to the single bug

TYPE: PathLike

Source code in src/diffannotator/gather_data.py 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def __init__(self, bug_dir: PathLike, annotations_dir: str = Bug.DEFAULT_ANNOTATIONS_DIR):
    """Constructor of the annotated bug

    Parameters
    ----------
    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 

gather_data(
    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

PARAMETER DESCRIPTION
bug_mapper

function to map bug to datastructure

TYPE: Callable[..., T]

datastructure_generator

function to create empty datastructure to combine results via "+"

TYPE: Callable[[], T]

RETURNS DESCRIPTION
T

combined datastructure with all files data
Source code in src/diffannotator/gather_data.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
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

    Parameters
    ----------
    bug_mapper
        function to map bug to datastructure
    datastructure_generator
        function to create empty datastructure to combine results
        via "+"

    Returns
    -------
    T
        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 

gather_data_dict(
    bug_dict_mapper: Callable[..., dict], **mapper_kwargs
) -> dict

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

PARAMETER DESCRIPTION
bug_dict_mapper

function to map diff to dictionary

TYPE: Callable[..., dict]

RETURNS DESCRIPTION
dict

combined dictionary of all diffs
Source code in src/diffannotator/gather_data.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
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

    Parameters
    ----------
    bug_dict_mapper
        function to map diff to dictionary

    Returns
    -------
    dict
        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 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
class AnnotatedBugDataset:
    """Annotated bugs dataset class"""

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

        Parameters
        ----------
        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

        Parameters
        ----------
        bug_mapper
            function to map bug to datastructure
        datastructure_generator
            function to create empty datastructure to combine results
            via "+"
        annotations_dir
            subdirectory where annotations are; path to annotation in a
            dataset is <bug_id>/<annotations_dir>/<patch_data>.json

        Returns
        -------
        T
            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

        Parameters
        ----------
        bug_dict_mapper
            function to map diff to dictionary
        annotations_dir
            subdirectory where annotations are; path to annotation in a
            dataset is <bug_id>/<annotations_dir>/<patch_data>.json

        Returns
        -------
        dict
            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

        Parameters
        ----------
        bug_to_dict_mapper
            function to map diff annotations to dictionary
        annotations_dir
            subdirectory where annotations are; path to annotation in a
            dataset is <bug_id>/<annotations_dir>/<patch_data>.json

        Returns
        -------
        list
            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__ 

__init__(dataset_dir: PathLike)

Constructor of the annotated bug dataset.

PARAMETER DESCRIPTION
dataset_dir

path to the dataset

TYPE: PathLike

Source code in src/diffannotator/gather_data.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def __init__(self, dataset_dir: PathLike):
    """Constructor of the annotated bug dataset.

    Parameters
    ----------
    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 

gather_data(
    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

PARAMETER DESCRIPTION
bug_mapper

function to map bug to datastructure

TYPE: Callable[..., T]

datastructure_generator

function to create empty datastructure to combine results via "+"

TYPE: Callable[[], T]

annotations_dir

subdirectory where annotations are; path to annotation in a dataset is //.json

TYPE: str DEFAULT: DEFAULT_ANNOTATIONS_DIR

RETURNS DESCRIPTION
T

combined datastructure with all bug data
Source code in src/diffannotator/gather_data.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
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

    Parameters
    ----------
    bug_mapper
        function to map bug to datastructure
    datastructure_generator
        function to create empty datastructure to combine results
        via "+"
    annotations_dir
        subdirectory where annotations are; path to annotation in a
        dataset is <bug_id>/<annotations_dir>/<patch_data>.json

    Returns
    -------
    T
        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 

gather_data_dict(
    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

PARAMETER DESCRIPTION
bug_dict_mapper

function to map diff to dictionary

TYPE: Callable[..., dict]

annotations_dir

subdirectory where annotations are; path to annotation in a dataset is //.json

TYPE: str DEFAULT: DEFAULT_ANNOTATIONS_DIR

RETURNS DESCRIPTION
dict

combined dictionary of all bugs
Source code in src/diffannotator/gather_data.py 
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
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

    Parameters
    ----------
    bug_dict_mapper
        function to map diff to dictionary
    annotations_dir
        subdirectory where annotations are; path to annotation in a
        dataset is <bug_id>/<annotations_dir>/<patch_data>.json

    Returns
    -------
    dict
        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 

gather_data_list(
    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

PARAMETER DESCRIPTION
bug_to_dict_mapper

function to map diff annotations to dictionary

TYPE: Callable[..., dict]

annotations_dir

subdirectory where annotations are; path to annotation in a dataset is //.json

TYPE: str DEFAULT: DEFAULT_ANNOTATIONS_DIR

RETURNS DESCRIPTION
list

list of bug dictionaries
Source code in src/diffannotator/gather_data.py 
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
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

    Parameters
    ----------
    bug_to_dict_mapper
        function to map diff annotations to dictionary
    annotations_dir
        subdirectory where annotations are; path to annotation in a
        dataset is <bug_id>/<annotations_dir>/<patch_data>.json

    Returns
    -------
    list
        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 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
class AnnotatedFile:
    """Annotated single file in specific bug"""

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

        Parameters
        ----------
        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

        Parameters
        ----------
        bug_mapper
            function to map bug to datastructure

        Returns
        -------
        T
            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__ 

__init__(file_path: PathLike)

Constructor of the annotated file of specific bug

PARAMETER DESCRIPTION
file_path

path to the single file

TYPE: PathLike

Source code in src/diffannotator/gather_data.py 
292
293
294
295
296
297
298
299
300
def __init__(self, file_path: PathLike):
    """Constructor of the annotated file of specific bug

    Parameters
    ----------
    file_path
        path to the single file
    """
    self._path = Path(file_path)

gather_data 

gather_data(
    bug_mapper: Callable[..., T], **mapper_kwargs
) -> T

Retrieves data from file

PARAMETER DESCRIPTION
bug_mapper

function to map bug to datastructure

TYPE: Callable[..., T]

RETURNS DESCRIPTION
T

resulting datastructure
Source code in src/diffannotator/gather_data.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
def gather_data(self, bug_mapper: Callable[..., T],
                **mapper_kwargs) -> T:
    """Retrieves data from file

    Parameters
    ----------
    bug_mapper
        function to map bug to datastructure

    Returns
    -------
    T
        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 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
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 "+"

        Returns
        -------
        ListAddedLinesResults
            empty datastructure
        """
        return ListAddedLinesResults([], [])

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

        Parameters
        ----------
        file_path
            path to processed file
        data
            dictionary with annotations (file content)

        Returns
        -------
        ListAddedLinesResults
            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 staticmethod 

create(file_path, data)

Override this function for single annotation handling

PARAMETER DESCRIPTION
file_path

path to processed file
data

dictionary with annotations (file content)
RETURNS DESCRIPTION
ListAddedLinesResults

datastructure instance
Source code in src/diffannotator/gather_data.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
@staticmethod
def create(file_path, data):
    """Override this function for single annotation handling

    Parameters
    ----------
    file_path
        path to processed file
    data
        dictionary with annotations (file content)

    Returns
    -------
    ListAddedLinesResults
        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 

default()

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

RETURNS DESCRIPTION
ListAddedLinesResults

empty datastructure
Source code in src/diffannotator/gather_data.py 
253
254
255
256
257
258
259
260
261
262
@staticmethod
def default():
    """Constructs empty datastructure to work as 0 for addition via "+"

    Returns
    -------
    ListAddedLinesResults
        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 
111
112
113
114
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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
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 "+"

        Returns
        -------
        PurposeCounterResults
            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

        Parameters
        ----------
        file_path
            path to processed file
        data
            dictionary with annotations (file content)
        data_format
            version of data schema used by annotation file

        Returns
        -------
        PurposeCounterResults
            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 staticmethod 

create(
    file_path: str,
    data: dict,
    data_format: JSONFormat = JSONFormat.V1_5,
) -> PurposeCounterResults

Override this function for single annotation handling

PARAMETER DESCRIPTION
file_path

path to processed file

TYPE: str

data

dictionary with annotations (file content)

TYPE: dict

data_format

version of data schema used by annotation file

TYPE: JSONFormat DEFAULT: V1_5

RETURNS DESCRIPTION
PurposeCounterResults

datastructure instance
Source code in src/diffannotator/gather_data.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@staticmethod
def create(file_path: str, data: dict,
           data_format: JSONFormat = JSONFormat.V1_5) -> 'PurposeCounterResults':
    """Override this function for single annotation handling

    Parameters
    ----------
    file_path
        path to processed file
    data
        dictionary with annotations (file content)
    data_format
        version of data schema used by annotation file

    Returns
    -------
    PurposeCounterResults
        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 

default() -> PurposeCounterResults

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

RETURNS DESCRIPTION
PurposeCounterResults

empty datastructure
Source code in src/diffannotator/gather_data.py 
175
176
177
178
179
180
181
182
183
184
@staticmethod
def default() -> 'PurposeCounterResults':
    """Constructs empty datastructure to work as 0 for addition via "+"

    Returns
    -------
    PurposeCounterResults
        empty datastructure
    """
    return PurposeCounterResults([], Counter(), Counter(), Counter())

common 

common(
    ctx: Context,
    annotations_dir: Annotated[
        str,
        Option(
            metavar=DIR_NAME,
            help="Subdirectory to read annotations from; use '' to do without such",
        ),
    ] = Bug.DEFAULT_ANNOTATIONS_DIR,
) -> None

Generate a configurable report or a summary of annotation results. Each summary is saved as a single JSON file. Various subcommands compute different types of statistics.

To create annotation results, run the diff-annotate command.

Source code in src/diffannotator/gather_data.py 
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
@app.callback()
def common(
    ctx: typer.Context,
    annotations_dir: Annotated[
        str,
        typer.Option(
            metavar="DIR_NAME",
            help="Subdirectory to read annotations from; use '' to do without such"
        )
    ] = Bug.DEFAULT_ANNOTATIONS_DIR,
) -> None:
    """Generate a configurable report or a summary of annotation results.
    Each summary is saved as a single JSON file.  Various subcommands
    compute different types of statistics.

    To create annotation results, run the `diff-annotate` command.
    """
    # if anything is printed by this function, it needs to check the context
    # to not break installed shell completion for the command
    # see https://typer.tiangolo.com/tutorial/options/callback-and-context/#fix-completion-using-the-context
    if ctx.resilient_parsing:
        return

    # pass to subcommands via context
    # TODO: use this technique for other scripts
    ctx.obj = SimpleNamespace(
        annotations_dir=annotations_dir,
    )

lines_stats 

lines_stats(
    ctx: Context,
    output_file: Annotated[
        Path,
        Argument(
            dir_okay=False,
            help="JSON file to write gathered results to",
        ),
    ],
    datasets: Annotated[
        list[Path],
        Argument(
            exists=True,
            file_okay=False,
            dir_okay=True,
            readable=True,
            writable=False,
            help="list of dirs with datasets to process",
        ),
    ],
    purpose_to_annotation: Annotated[
        Optional[list[Tuple]],
        Option(
            help="Mapping from file PURPOSE to line type LINE_TYPE.\n                    Each line of such file will be treated as if it had given type.\n                    As a shortcut, giving PURPOSE is the same as PURPOSE:PURPOSE.\n                    Can be given multiple times.",
            metavar="PURPOSE:LINE_TYPE|PURPOSE",
            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.

Source code in src/diffannotator/gather_data.py 
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
@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 

list_added_lines(
    datasets: Annotated[
        list[Path],
        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.

Source code in src/diffannotator/gather_data.py 
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
@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 

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. "+",...).

PARAMETER DESCRIPTION
annotation_file_basename

name of JSON file with annotation data

TYPE: str

annotation_data

parsed annotations data, retrieved from annotation_file_basename file.

TYPE: dict

data_format

version of data schema used by annotation file

TYPE: JSONFormat DEFAULT: V1_5

Source code in src/diffannotator/gather_data.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
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. "+",...).

    Parameters
    ----------
    annotation_file_basename
        name of JSON file with annotation data
    annotation_data
        parsed annotations data, retrieved from
        `annotation_file_basename` file.
    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 

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'],
}
PARAMETER DESCRIPTION
_diff_file_path

file path containing diff, ignored

TYPE: str

data

dictionary loaded from file

TYPE: dict

data_format

version of data schema used by annotation file

TYPE: JSONFormat DEFAULT: V1_5

RETURNS DESCRIPTION
dict

dictionary with file purposes
Source code in src/diffannotator/gather_data.py 
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
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'],
        }

    Parameters
    ----------
    _diff_file_path
        file path containing diff, ignored
    data
        dictionary loaded from file
    data_format
        version of data schema used by annotation file

    Returns
    -------
    dict
        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 

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. "+",...).

PARAMETER DESCRIPTION
annotation_file_basename

name of JSON file with annotation data

TYPE: str

annotation_data

parsed annotations data, retrieved from annotation_file_basename file.

TYPE: dict

data_format

version of data schema used by annotation file

TYPE: JSONFormat DEFAULT: V1_5

purpose_to_annotation

list of pairs (, ) to treat each line of file with given purpose to have given type annotation.

TYPE: Optional[list] DEFAULT: None

Source code in src/diffannotator/gather_data.py 
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
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. "+",...).

    Parameters
    ----------
    annotation_file_basename
        name of JSON file with annotation data
    annotation_data
        parsed annotations data, retrieved from
        `annotation_file_basename` file.
    data_format
        version of data schema used by annotation file
    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 

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')])

PARAMETER DESCRIPTION
value

string with colon-separated values, 'KEY:VALUE', or stringwithout colon, 'STR'

TYPE: str

RETURNS DESCRIPTION
(str, str)

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 
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
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'}

    Parameters
    ----------
    value
        string with colon-separated values, 'KEY:VALUE', or
        stringwithout colon, 'STR'

    Returns
    -------
    (str, str)
        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 

purpose_counter(
    ctx: Context,
    datasets: Annotated[
        list[Path],
        Argument(
            exists=True,
            file_okay=False,
            dir_okay=True,
            readable=True,
            writable=False,
        ),
    ],
    result_json: Annotated[
        Optional[Path],
        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.

Source code in src/diffannotator/gather_data.py 
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
@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 

purpose_per_file(
    ctx: Context,
    result_json: Annotated[
        Path,
        Argument(
            dir_okay=False,
            help="JSON file to write gathered results to",
        ),
    ],
    datasets: Annotated[
        list[Path],
        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.

Source code in src/diffannotator/gather_data.py 
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
@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 

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
PARAMETER DESCRIPTION
result

data to serialize and save

TYPE: Any

result_json

path to JSON file to save result to

TYPE: Path

Source code in src/diffannotator/gather_data.py 
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
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

    Parameters
    ----------
    result
        data to serialize and save
    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 

timeline(
    ctx: Context,
    output_file: Annotated[
        Path,
        Argument(
            dir_okay=False,
            help="file to write gathered results to",
        ),
    ],
    datasets: Annotated[
        list[Path],
        Argument(
            exists=True,
            file_okay=False,
            dir_okay=True,
            readable=True,
            writable=False,
            help="list of dirs with datasets to process",
        ),
    ],
    purpose_to_annotation: Annotated[
        Optional[list[Tuple]],
        Option(
            help="Mapping from file PURPOSE to line type LINE_TYPE.\n                    Each line of such file will be treated as if it had given type.\n                    As a shortcut, giving PURPOSE is the same as PURPOSE:PURPOSE.\n                    Can be given multiple times.",
            metavar="PURPOSE:LINE_TYPE|PURPOSE",
            parser=parse_colon_separated_pair,
        ),
    ] = None,
) -> 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 
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
@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)