data_file_writer

frog.hardware.data_file_writer

Provides a class for writing sensor data to a CSV file.

Classes

DataFileWriter()

A class for writing sensor data to a CSV file.

This class is a singleton. Every time a new file needs to be written this instance is reused.

Create a new DataFileWriter.

Source code in frog/hardware/data_file_writer.py

def __init__(self) -> None:
    """Create a new DataFileWriter."""
    self._writer: Writer
    """The CSV writer."""

    # Listen to open/close messages
    pub.subscribe(self.open, "data_file.open")
    pub.subscribe(self.close, "data_file.close")

    # Close data file if GUI closes unexpectedly
    pub.subscribe(self.close, "window.closed")

Functions

close()

Close the current file handle.

Source code in frog/hardware/data_file_writer.py

def close(self) -> None:
    """Close the current file handle."""
    pub.unsubscribe(
        self.write, f"device.{config.TEMPERATURE_MONITOR_TOPIC}.data.response"
    )
    pub.unsubscribe(_on_error_occurred, "data_file.error")

    if hasattr(self, "_writer"):
        logging.info("Closing data file")

        # Ensure data is written to disk
        os.fsync(self._writer._file.fileno())

        self._writer.close()
        del self._writer

open(path)

Open a file at the specified path for writing.

Parameters:

Name	Type	Description	Default
path	Path	The path of the file to write to	required

Source code in frog/hardware/data_file_writer.py

@pubsub_errors("data_file.error")
def open(self, path: Path) -> None:
    """Open a file at the specified path for writing.

    Args:
        path: The path of the file to write to
    """
    logging.info(f"Opening data file at {path}")
    self._writer = _create_writer(path)

    # Listen to temperature monitor messages
    pub.subscribe(
        self.write, f"device.{config.TEMPERATURE_MONITOR_TOPIC}.data.response"
    )

    # Call close() if writing errors occur
    pub.subscribe(_on_error_occurred, "data_file.error")

    # Send message to indicate that file has opened successfully
    pub.sendMessage("data_file.opened")

write(time, temperatures)

Write temperature readings to the CSV file.

Source code in frog/hardware/data_file_writer.py

@pubsub_errors("data_file.error")
def write(self, time: datetime, temperatures: list[Decimal]) -> None:
    """Write temperature readings to the CSV file."""
    # Also include timestamp as seconds since midnight
    midnight = datetime(time.year, time.month, time.day)
    secs_since_midnight = floor((time - midnight).total_seconds())

    angle, is_moving = _get_stepper_motor_angle()
    self._writer.writerow(
        (
            time.strftime("%Y%m%d"),
            time.strftime("%H:%M:%S"),
            *(round(t, config.TEMPERATURE_PRECISION) for t in temperatures),
            secs_since_midnight,
            angle,
            int(is_moving),
            _get_hot_bb_power(),
        )
    )

    pub.sendMessage("data_file.writing")

Functions