Skip to content

unitorch.cli.writer¤

GeneralCsvWriter¤

Tip

core/writer/csv is the section for configuration of GeneralCsvWriter.

Bases: GenericWriter

Write outputs to a CSV/TSV file.

Source code in src/unitorch/cli/writers/__init__.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def __init__(
    self,
    output_file: str,
    nrows_per_sample: Optional[int] = None,
    header: Optional[bool] = None,
    columns: Optional[List[str]] = None,
    sep: Optional[str] = "\t",
    quoting: Optional[int] = 3,
    escapechar: Optional[str] = None,
):
    self.header = header
    self.columns = columns
    self.sep = sep
    self.quoting = quoting
    self.escapechar = escapechar
    has_header = int(header is True)
    self.skip_n_samples = (
        0
        if nrows_per_sample is None or not os.path.exists(output_file)
        else (sum(1 for _ in open(output_file)) - has_header) // nrows_per_sample
    )
    mode = "a" if self.skip_n_samples > 0 else "w"
    self.output_file = open(output_file, mode, encoding="utf-8")

header instance-attribute ¤

header = header

columns instance-attribute ¤

columns = columns

sep instance-attribute ¤

sep = sep

quoting instance-attribute ¤

quoting = quoting

escapechar instance-attribute ¤

escapechar = escapechar

skip_n_samples instance-attribute ¤

skip_n_samples = (
    0
    if (nrows_per_sample is None or not exists(output_file))
    else (sum(1 for _ in (open(output_file))) - has_header)
    // nrows_per_sample
)

output_file instance-attribute ¤

output_file = open(output_file, mode, encoding='utf-8')

from_config classmethod ¤

from_config(config, **kwargs)
Source code in src/unitorch/cli/writers/__init__.py 
104
105
106
107
@classmethod
@config_defaults_init("core/writer/csv")
def from_config(cls, config, **kwargs):
    pass

_write ¤

_write(
    outputs: WriterOutputs, include_header: bool = False
)
Source code in src/unitorch/cli/writers/__init__.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def _write(self, outputs: "WriterOutputs", include_header: bool = False):
    dataframe = outputs.to_pandas()
    if self.columns is not None:
        cols = set(dataframe.columns)
        dataframe = dataframe[[c for c in self.columns if c in cols]]
    self.output_file.write(
        dataframe.to_csv(
            index=False,
            sep=self.sep,
            quoting=self.quoting,
            header=include_header,
            escapechar=self.escapechar,
        )
    )
    self.output_file.flush()

process_start ¤

process_start(outputs: WriterOutputs)
Source code in src/unitorch/cli/writers/__init__.py 
125
126
127
128
def process_start(self, outputs: "WriterOutputs"):
    self._write(
        outputs, include_header=bool(self.header) and self.skip_n_samples == 0
    )

process_chunk ¤

process_chunk(outputs: WriterOutputs)
Source code in src/unitorch/cli/writers/__init__.py 
130
131
def process_chunk(self, outputs: "WriterOutputs"):
    self._write(outputs, include_header=False)

process_end ¤

process_end()
Source code in src/unitorch/cli/writers/__init__.py 
133
134
def process_end(self):
    self.output_file.close()