Initialize the FileOutput object. Args: output_file_name: name of file to write to. output_dir: path to directory to write to. on_csv_write: "append" or "update", whether or not to append or overwrite a file if it exists Raises:
(
self, output_file_name: str, output_dir: str, on_csv_write: str = "append"
)
| 22 | """ |
| 23 | |
| 24 | def __init__( |
| 25 | self, output_file_name: str, output_dir: str, on_csv_write: str = "append" |
| 26 | ): |
| 27 | """ |
| 28 | Initialize the FileOutput object. |
| 29 | |
| 30 | Args: |
| 31 | output_file_name: name of file to write to. |
| 32 | output_dir: path to directory to write to. |
| 33 | on_csv_write: "append" or "update", whether or not to append or overwrite a file if it exists |
| 34 | |
| 35 | Raises: |
| 36 | ValueError: If the on_csv_write value is invalid. |
| 37 | OSError: If the output directory does not exist. |
| 38 | """ |
| 39 | if on_csv_write not in {"append", "update"}: |
| 40 | raise ValueError( |
| 41 | f"Unknown `on_csv_write` value: {on_csv_write}" |
| 42 | + " (should be one of 'append' or 'update'" |
| 43 | ) |
| 44 | self.output_file_name: str = output_file_name |
| 45 | if not os.path.exists(output_dir): |
| 46 | raise OSError(f"Folder '{output_dir}' doesn't exist !") |
| 47 | self.output_dir: str = output_dir |
| 48 | self.on_csv_write: str = on_csv_write |
| 49 | self.save_file_path = os.path.join(self.output_dir, self.output_file_name) |
| 50 | logger.info( |
| 51 | f"Emissions data (if any) will be saved to file {os.path.abspath(self.save_file_path)}" |
| 52 | ) |
| 53 | |
| 54 | def has_valid_headers(self, data: EmissionsData) -> bool: |
| 55 | """ |
nothing calls this directly
no outgoing calls
no test coverage detected