Sam
3a34759094
Some checks failed
Build Simulation and Test / Run All Tests (push) Failing after 8m17s
Major rewrite.
140 lines
4.2 KiB
Python
140 lines
4.2 KiB
Python
"""File writer for output data."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, List, Optional
|
|
from .base_writer import BaseWriter
|
|
|
|
|
|
class FileWriter(BaseWriter):
|
|
"""Writes data to files."""
|
|
|
|
def __init__(self, output_dir: str = "simulation_output", create_dirs: bool = True):
|
|
self.output_dir = Path(output_dir)
|
|
self.create_dirs = create_dirs
|
|
self.file_handles = {}
|
|
self.ready = False
|
|
|
|
if self.create_dirs:
|
|
self.output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
self.ready = self.output_dir.exists()
|
|
|
|
def write(self, data: Any, filename: Optional[str] = None, data_type: str = "data") -> bool:
|
|
"""Write data to file."""
|
|
if not self.ready:
|
|
return False
|
|
|
|
try:
|
|
if filename is None:
|
|
# Generate filename based on data type and timestamp
|
|
import time
|
|
timestamp = int(time.time())
|
|
filename = f"{data_type}_{timestamp}.txt"
|
|
|
|
filepath = self.output_dir / filename
|
|
|
|
# Handle different data types
|
|
if isinstance(data, str):
|
|
# Already formatted data
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
f.write(data)
|
|
elif isinstance(data, (list, dict)):
|
|
# Convert to string representation
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
f.write(str(data))
|
|
else:
|
|
# Single value
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
f.write(str(data))
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error writing to file {filename}: {e}")
|
|
return False
|
|
|
|
def write_batch(self, data_list: List[Any], filename: str, data_type: str = "batch") -> bool:
|
|
"""Write multiple data items to a single file."""
|
|
if not self.ready or not data_list:
|
|
return False
|
|
|
|
try:
|
|
filepath = self.output_dir / filename
|
|
|
|
# Combine all data items
|
|
if isinstance(data_list[0], str):
|
|
# Already formatted strings
|
|
content = '\n'.join(data_list)
|
|
else:
|
|
# Convert to string representation
|
|
content = '\n'.join(str(item) for item in data_list)
|
|
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
f.write(content)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error writing batch to file {filename}: {e}")
|
|
return False
|
|
|
|
def append(self, data: Any, filename: str) -> bool:
|
|
"""Append data to existing file."""
|
|
if not self.ready:
|
|
return False
|
|
|
|
try:
|
|
filepath = self.output_dir / filename
|
|
|
|
with open(filepath, 'a', encoding='utf-8') as f:
|
|
f.write(str(data))
|
|
f.write('\n') # Add newline
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error appending to file {filename}: {e}")
|
|
return False
|
|
|
|
def get_file_path(self, filename: str) -> Path:
|
|
"""Get full path for a filename."""
|
|
return self.output_dir / filename
|
|
|
|
def list_files(self, pattern: str = "*") -> List[str]:
|
|
"""List files in output directory."""
|
|
if not self.ready:
|
|
return []
|
|
|
|
try:
|
|
return [f.name for f in self.output_dir.glob(pattern) if f.is_file()]
|
|
except Exception:
|
|
return []
|
|
|
|
def delete_file(self, filename: str) -> bool:
|
|
"""Delete a file."""
|
|
if not self.ready:
|
|
return False
|
|
|
|
try:
|
|
filepath = self.output_dir / filename
|
|
if filepath.exists():
|
|
filepath.unlink()
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
print(f"Error deleting file {filename}: {e}")
|
|
return False
|
|
|
|
def close(self):
|
|
"""Close file handles and cleanup."""
|
|
for handle in self.file_handles.values():
|
|
try:
|
|
handle.close()
|
|
except:
|
|
pass
|
|
self.file_handles.clear()
|
|
|
|
def is_ready(self) -> bool:
|
|
"""Check if writer is ready for writing."""
|
|
return self.ready |