Skip to content

Writer Strategy

WriterStrategy

Bases: ABC

Abstract base class for data file writers.

The Writer class defines the interface for writing sensor measurement data to files in various formats (CSV, Parquet, JSON, etc.). Implementations handle format-specific serialization and file operations while maintaining a consistent interface for the SensorLogger.

This class uses the Strategy pattern to allow runtime selection of data output formats without modifying the logger code.

Methods:

Name Description
write

Write a single data point to the specified file.
get_extension

Return the file extension for this writer's format.

Examples:

Implementing a custom writer:

>>> from pathlib import Path
>>> from typing import Dict, Any
>>>
>>> class CustomWriter(Writer):
...     def write(self, filepath: Path, data: Dict[str, Any], is_first: bool):
...         with open(filepath, 'a') as f:
...             f.write(f"{data}\n")
...
...     def get_extension(self) -> str:
...         return 'txt'
>>>
>>> writer = CustomWriter()
>>> writer.write(Path('output.txt'), {'value': 42}, True)
Source code in atmospyre/loggers/strategies/writer.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class WriterStrategy(ABC):
    """Abstract base class for data file writers.

    The Writer class defines the interface for writing sensor measurement data
    to files in various formats (CSV, Parquet, JSON, etc.). Implementations
    handle format-specific serialization and file operations while maintaining
    a consistent interface for the SensorLogger.

    This class uses the Strategy pattern to allow runtime selection of data
    output formats without modifying the logger code.

    Methods
    -------
    write(filepath: Path, data: Dict[str, Any], is_first: bool)
        Write a single data point to the specified file.
    get_extension() -> str
        Return the file extension for this writer's format.

    Examples
    --------
    Implementing a custom writer:

    >>> from pathlib import Path
    >>> from typing import Dict, Any
    >>>
    >>> class CustomWriter(Writer):
    ...     def write(self, filepath: Path, data: Dict[str, Any], is_first: bool):
    ...         with open(filepath, 'a') as f:
    ...             f.write(f"{data}\\n")
    ...
    ...     def get_extension(self) -> str:
    ...         return 'txt'
    >>>
    >>> writer = CustomWriter()
    >>> writer.write(Path('output.txt'), {'value': 42}, True)
    """

    @abstractmethod
    def write(self, filepath: Path, data: Dict[str, Any], is_first: bool):
        """Write data point to file.

        Parameters
        ----------
        filepath : Path
            Path to the output file. Parent directories must exist.
        data : Dict[str, Any]
            Dictionary containing the data point to write. Keys are column/field
            names, values are the measurement values. Typically includes a
            'timestamp' key with ISO format datetime string.
        is_first : bool
            True if this is the first write to the file today, False otherwise.
            Used to determine whether to write headers/schema.

        Notes
        -----
        Implementations should:

        - Create the file if it doesn't exist
        - Append to existing files
        - Write headers/schema when ``is_first`` is True
        - Handle data type conversion as needed for the format
        - Ensure atomic writes to prevent data corruption

        Examples
        --------
        >>> data = {
        ...     'timestamp': '2025-10-24T14:30:00.123456',
        ...     'CO2': 415.2,
        ...     'TEMPERATURE': 23.5
        ... }
        >>> writer.write(Path('./data/20251024/data.csv'), data, is_first=True)
        """

    @abstractmethod
    def get_extension(self) -> str:
        """Get file extension for this writer's format.

        Returns
        -------
        str
            File extension without leading dot (e.g., 'csv', 'parquet', 'json').

        Examples
        --------
        >>> writer = CSVWriter()
        >>> writer.get_extension()
        'csv'
        """

Functions

write(filepath, data, is_first) abstractmethod

Write data point to file.

Parameters:

Name Type Description Default
filepath Path

Path to the output file. Parent directories must exist.

 required
data Dict[str, Any]

Dictionary containing the data point to write. Keys are column/field names, values are the measurement values. Typically includes a 'timestamp' key with ISO format datetime string.

 required
is_first bool

True if this is the first write to the file today, False otherwise. Used to determine whether to write headers/schema.

 required
Notes

Implementations should:

  • Create the file if it doesn't exist
  • Append to existing files
  • Write headers/schema when is_first is True
  • Handle data type conversion as needed for the format
  • Ensure atomic writes to prevent data corruption

Examples:

>>> data = {
...     'timestamp': '2025-10-24T14:30:00.123456',
...     'CO2': 415.2,
...     'TEMPERATURE': 23.5
... }
>>> writer.write(Path('./data/20251024/data.csv'), data, is_first=True)
Source code in atmospyre/loggers/strategies/writer.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@abstractmethod
def write(self, filepath: Path, data: Dict[str, Any], is_first: bool):
    """Write data point to file.

    Parameters
    ----------
    filepath : Path
        Path to the output file. Parent directories must exist.
    data : Dict[str, Any]
        Dictionary containing the data point to write. Keys are column/field
        names, values are the measurement values. Typically includes a
        'timestamp' key with ISO format datetime string.
    is_first : bool
        True if this is the first write to the file today, False otherwise.
        Used to determine whether to write headers/schema.

    Notes
    -----
    Implementations should:

    - Create the file if it doesn't exist
    - Append to existing files
    - Write headers/schema when ``is_first`` is True
    - Handle data type conversion as needed for the format
    - Ensure atomic writes to prevent data corruption

    Examples
    --------
    >>> data = {
    ...     'timestamp': '2025-10-24T14:30:00.123456',
    ...     'CO2': 415.2,
    ...     'TEMPERATURE': 23.5
    ... }
    >>> writer.write(Path('./data/20251024/data.csv'), data, is_first=True)
    """

get_extension() abstractmethod

Get file extension for this writer's format.

Returns:

Type Description
str

File extension without leading dot (e.g., 'csv', 'parquet', 'json').

Examples:

>>> writer = CSVWriter()
>>> writer.get_extension()
'csv'
Source code in atmospyre/loggers/strategies/writer.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@abstractmethod
def get_extension(self) -> str:
    """Get file extension for this writer's format.

    Returns
    -------
    str
        File extension without leading dot (e.g., 'csv', 'parquet', 'json').

    Examples
    --------
    >>> writer = CSVWriter()
    >>> writer.get_extension()
    'csv'
    """