Method out

codecarbon/output_methods/file.py:74–130

Save the emissions data from a whole run to a CSV file. * If the file does not exist, then create it. * If the file already exists but has invalid headers, then back it up and replace with new data. * If the file already exists and has valid headers: * I

(self, total: EmissionsData, _)

            return sorted(data.values.keys()) == list_of_column_names

    def out(self, total: EmissionsData, _):
        """
        Save the emissions data from a whole run to a CSV file.

        * If the file does not exist, then create it.
        * If the file already exists but has invalid headers, then back it up and replace with new data.
        * If the file already exists and has valid headers:
            * If it has no rows with a matching run ID, append the new data.
            * If it has one row with a matching run ID, then replace that row with the new data.
            * If it has > one row with a matching run ID, append the new data

        Args:
            total: data to save.


        """
        file_exists: bool = os.path.isfile(self.save_file_path)
        if file_exists and os.path.getsize(self.save_file_path) == 0:
            logger.warning(
                f"File {self.save_file_path} exists but is empty. Treating as new file."
            )
            file_exists = False
        if file_exists and not self.has_valid_headers(total):
            logger.warning("The CSV format has changed, backing up old emission file.")
            backup(self.save_file_path)
            file_exists = False
        new_df = pd.DataFrame.from_records([dict(total.values)])
        if not file_exists:
            df = new_df
        elif self.on_csv_write == "append":
            df = pd.read_csv(self.save_file_path)
            # Filter out empty or all-NA columns only from new_df, to avoid warnings from Pandas,
            # see https://github.com/pandas-dev/pandas/issues/55928
            new_df = new_df.dropna(axis=1, how="all")
            df = pd.concat([df, new_df])
        else:
            df = pd.read_csv(self.save_file_path)
            df_run = df.loc[df.run_id == total.run_id]
            if len(df_run) < 1:
                df = pd.concat([df, new_df])
            elif len(df_run) > 1:
                logger.warning(
                    f"CSV contains more than 1 ({len(df_run)})"
                    + f" rows with current run ID ({total.run_id})."
                    + "Appending instead of updating."
                )
                df = pd.concat([df, new_df])
            else:
                update_values = {}
                for col, val in dict(total.values).items():
                    # Explicitly cast new values to prevent warnings about incompatible dtypes.
                    update_values[col] = df[col].dtype.type(val)
                df.loc[df.run_id == total.run_id, update_values.keys()] = (
                    update_values.values()
                )

        df.to_csv(self.save_file_path, index=False)