data/data_collection.py
def collect_signals(self) -> None:
        """Collects pull request signals and save to CSV file.

        Returns: None
        """
        print("Collecting signals for %s" % self._repo_name)
        if not os.path.exists(
                './%s_pull_requests_signals.csv' % self._repo_name):
            with open('./%s_pull_requests_signals.csv' % self._repo_name, 'a') \
                    as file:
                writer = csv.writer(file)
                writer.writerow(["repo name", "pull request id", "author",
                                 "pull request created time",
                                 "pull request closed time",
                                 "pull request review time",
                                 "reverted pull request id",
                                 "pull request revert time",
                                 "num review comments", "review comments msg",
                                 "num issue comments", "issue comments msg",
                                 "num approved reviewers", "approved reviewers",
                                 "num commits", "num line changes",
                                 "files changes",
                                 "file versions", "check run results"])
        if self._find_all:
            print("Retrieving all pull requests for %s" % self._repo_name)
            if not os.path.exists(
                    './%s_pull_requests.txt' % self._repo_name):
                pull_requests = get_all_pull_requests(
                    self._repo_name, self._start_date, self._end_date, 'closed',
                    self._auth)
                if pull_requests is None:
                    return
                save_pull_requests(self._repo_name, pull_requests)
            else:
                print("Reading from file")
                pull_requests = []
                with open('./%s_pull_requests.txt' % self._repo_name, 'r') as f:
                    for datum in f:
                        pull_requests.append(eval(datum))
        else:
            print("Retrieving pull requests on page %s for %s"
                  % (self._page, self._repo_name))
            pull_requests = get_pull_requests_by_page(
                self._page, self._repo_name, self._start_date, self._end_date,
                'closed', self._auth)
            if pull_requests is None:
                return

        collected_pr_signals = pd.read_csv('./%s_pull_requests_signals.csv'
                                           % self._repo_name)
        pr_ids = set(collected_pr_signals['pull request id'].tolist())

        for pull_request_info in pull_requests:
            if not pull_request_info:
                continue
            pull_request_number = pull_request_info['number']
            if pull_request_number in pr_ids:
                continue
            closed_time = pull_request_info['closed_at']
            merged_time = pull_request_info['merged_at']
            if not merged_time:
                continue
            if not (to_timestamp(self._start_date) <=
                    to_timestamp(closed_time) <= to_timestamp(self._end_date)):
                continue
            datum = self._collect_signals_for_one_pull_request(
                pull_request_info)
            if not datum:
                continue
            with open('./%s_pull_requests_signals.csv' % self._repo_name, 'a') \
                    as file:
                writer = csv.writer(file)
                writer.writerow(datum)
Collects pull request signals and save to CSV file.

        Returns: None
        
Collects pull request signals and save to CSV file.
Returns: None
Collects pull request signals and save to CSV file . Returns : None
def collect_signals(self) -> None:
        print("Collecting signals for %s" % self._repo_name)
        if not os.path.exists(
                './%s_pull_requests_signals.csv' % self._repo_name):
            with open('./%s_pull_requests_signals.csv' % self._repo_name, 'a') \
                    as file:
                writer = csv.writer(file)
                writer.writerow(["repo name", "pull request id", "author",
                                 "pull request created time",
                                 "pull request closed time",
                                 "pull request review time",
                                 "reverted pull request id",
                                 "pull request revert time",
                                 "num review comments", "review comments msg",
                                 "num issue comments", "issue comments msg",
                                 "num approved reviewers", "approved reviewers",
                                 "num commits", "num line changes",
                                 "files changes",
                                 "file versions", "check run results"])
        if self._find_all:
            print("Retrieving all pull requests for %s" % self._repo_name)
            if not os.path.exists(
                    './%s_pull_requests.txt' % self._repo_name):
                pull_requests = get_all_pull_requests(
                    self._repo_name, self._start_date, self._end_date, 'closed',
                    self._auth)
                if pull_requests is None:
                    return
                save_pull_requests(self._repo_name, pull_requests)
            else:
                print("Reading from file")
                pull_requests = []
                with open('./%s_pull_requests.txt' % self._repo_name, 'r') as f:
                    for datum in f:
                        pull_requests.append(eval(datum))
        else:
            print("Retrieving pull requests on page %s for %s"
                  % (self._page, self._repo_name))
            pull_requests = get_pull_requests_by_page(
                self._page, self._repo_name, self._start_date, self._end_date,
                'closed', self._auth)
            if pull_requests is None:
                return
        collected_pr_signals = pd.read_csv('./%s_pull_requests_signals.csv'
                                           % self._repo_name)
        pr_ids = set(collected_pr_signals['pull request id'].tolist())
        for pull_request_info in pull_requests:
            if not pull_request_info:
                continue
            pull_request_number = pull_request_info['number']
            if pull_request_number in pr_ids:
                continue
            closed_time = pull_request_info['closed_at']
            merged_time = pull_request_info['merged_at']
            if not merged_time:
                continue
            if not (to_timestamp(self._start_date) <=
                    to_timestamp(closed_time) <= to_timestamp(self._end_date)):
                continue
            datum = self._collect_signals_for_one_pull_request(
                pull_request_info)
            if not datum:
                continue
            with open('./%s_pull_requests_signals.csv' % self._repo_name, 'a') \
                    as file:
                writer = csv.writer(file)
                writer.writerow(datum)
def collect_signals ( self ) -> None : print ( "Collecting signals for %s" % self . _repo_name ) if not os . path . exists ( './%s_pull_requests_signals.csv' % self . _repo_name ) : with open ( './%s_pull_requests_signals.csv' % self . _repo_name , 'a' ) as file : writer = csv . writer ( file ) writer . writerow ( [ "repo name" , "pull request id" , "author" , "pull request created time" , "pull request closed time" , "pull request review time" , "reverted pull request id" , "pull request revert time" , "num review comments" , "review comments msg" , "num issue comments" , "issue comments msg" , "num approved reviewers" , "approved reviewers" , "num commits" , "num line changes" , "files changes" , "file versions" , "check run results" ] ) if self . _find_all : print ( "Retrieving all pull requests for %s" % self . _repo_name ) if not os . path . exists ( './%s_pull_requests.txt' % self . _repo_name ) : pull_requests = get_all_pull_requests ( self . _repo_name , self . _start_date , self . _end_date , 'closed' , self . _auth ) if pull_requests is None : return save_pull_requests ( self . _repo_name , pull_requests ) else : print ( "Reading from file" ) pull_requests = [ ] with open ( './%s_pull_requests.txt' % self . _repo_name , 'r' ) as f : for datum in f : pull_requests . append ( eval ( datum ) ) else : print ( "Retrieving pull requests on page %s for %s" % ( self . _page , self . _repo_name ) ) pull_requests = get_pull_requests_by_page ( self . _page , self . _repo_name , self . _start_date , self . _end_date , 'closed' , self . _auth ) if pull_requests is None : return collected_pr_signals = pd . read_csv ( './%s_pull_requests_signals.csv' % self . _repo_name ) pr_ids = set ( collected_pr_signals [ 'pull request id' ] . tolist ( ) ) for pull_request_info in pull_requests : if not pull_request_info : continue pull_request_number = pull_request_info [ 'number' ] if pull_request_number in pr_ids : continue closed_time = pull_request_info [ 'closed_at' ] merged_time = pull_request_info [ 'merged_at' ] if not merged_time : continue if not ( to_timestamp ( self . _start_date ) <= to_timestamp ( closed_time ) <= to_timestamp ( self . _end_date ) ) : continue datum = self . _collect_signals_for_one_pull_request ( pull_request_info ) if not datum : continue with open ( './%s_pull_requests_signals.csv' % self . _repo_name , 'a' ) as file : writer = csv . writer ( file ) writer . writerow ( datum )
Collects pull request signals and save to CSV file.
Collects pull request signals and save to CSV file .
"""Collects pull request signals and save to CSV file.

        Returns: None
        """
self