### Install cfinterface from GitHub Source: https://rjmalves.github.io/cfinterface/_sources/install/install.rst.txt Installation method for testing development branches directly from the Git repository. ```bash pip install git+https://github.com/rjmalves/cfinterface.git ``` -------------------------------- ### Install development dependencies Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/contributing.rst.txt Commands to clone the repository and install necessary development tools using uv. ```bash git clone https://github.com/rjmalves/cfinterface.git cd cfinterface uv sync --extra dev uv run pre-commit install ``` -------------------------------- ### Install cfinterface via PyPi Source: https://rjmalves.github.io/cfinterface/_sources/install/install.rst.txt Standard installation method using the pip package manager. ```bash pip install cfinterface ``` -------------------------------- ### Example File Content Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Sample text file structure for section-based parsing. ```none Date Index Value ---------------------------- 2020/01 1 1000.0 2020/02 1 2000.0 2020/01 2 3000.0 2020/02 2 4000.0 2020/01 3 5000.0 2020/02 3 6000.0 ---------------------------- Username Last Login ------------------------- admin 1996/01/01 sunshine 2000/01/01 pineapple 1996/01/01 admin 1996/01/01 ------------------------- ``` -------------------------------- ### Install cfinterface with pandas support Source: https://rjmalves.github.io/cfinterface/_sources/guides/migration-v1.9.rst.txt Updates the installation command to use the [pandas] extra for DataFrame integration. ```bash # pandas was automatically installed with cfinterface pip install cfinterface ``` ```bash # Install the [pandas] extra only if you need DataFrame integration pip install cfinterface[pandas] ``` -------------------------------- ### Example Output Source: https://rjmalves.github.io/cfinterface/examples/plot_versioned_file.html Expected console output from reading and validating the versioned file. ```text === Version 2.0 Read === DefaultBlock: ConstantBlock: My Versioned File v2 DefaultBlock: END_FILE_HEADER NewDataBlock: [[1, 'Alice', 95.5], [2, 'Bob', 87.25]] === Version Validation === Matched: False Expected: ['ConstantBlock', 'NewDataBlock'] Found: ['ConstantBlock', 'NewDataBlock'] Default ratio: 50.0% === resolve_version('1.5') === Resolved to v1.0 components: ['ConstantBlock', 'OldDataBlock'] ``` -------------------------------- ### Example File Content Source: https://rjmalves.github.io/cfinterface/getting_started/tutorial.html A sample text file structure containing two distinct tables to be modeled. ```text 1Date Index Value 2---------------------------- 32020/01 1 1000.0 42020/02 1 2000.0 52020/01 2 3000.0 62020/02 2 4000.0 72020/01 3 5000.0 82020/02 3 6000.0 9---------------------------- 10 11 Username Last Login 12------------------------- 13 admin 1996/01/01 14 sunshine 2000/01/01 15 pineapple 1996/01/01 16 admin 1996/01/01 17------------------------- ``` -------------------------------- ### Example Register File Content Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Sample content of a file structured with line-based registers. ```none DATA_HIGH ID001 sudo.user 10/20/2025 901.25 DATA_HIGH ID002 sudo.user 10/21/2025 100.20 DATA_HIGH ID003 test.user 10/30/2025 100.20 DATA_LOW 01/01/2024 105.23 DATA_LOW 01/02/2024 29.15 DATA_LOW 01/03/2024 5.05 ``` -------------------------------- ### Install pandas integration Source: https://rjmalves.github.io/cfinterface/_sources/guides/faq.rst.txt Install the optional pandas dependency using pip. ```bash pip install cfinterface[pandas] ``` -------------------------------- ### Example Text File Structure Source: https://rjmalves.github.io/cfinterface/getting_started/tutorial.html A sample text file demonstrating multiple blocks defined by specific beginning and ending patterns. ```text 1MY_FIRST_BLOCK_BEGINNING_PATTERN 2I have some raw text for describing my content, because I was mean for being irectly read by someone. 3 4Now I have some data. After which I will be done. 5 6Date Index Value 72020/01 1 1000.0 82020/02 1 2000.0 92020/01 2 3000.0 102020/02 2 4000.0 112020/01 3 5000.0 122020/02 3 6000.0 13MY_FIRST_BLOCK_ENDING_PATTERN 14 15MY_SECOND_BLOCK_BEGINNING_PATTERN 16My content is completely different from the previous block... 17 Username Last Login 18 admin 1996/01/01 19 sunshine 2000/01/01 20 pineapple 1996/01/01 21 admin 1996/01/01 22MY_SECOND_BLOCK_ENDING_PATTERN 23 24MY_FIRST_BLOCK_BEGINNING_PATTERN 25... 26MY_FIRST_BLOCK_ENDING_PATTERN 27 28MY_FIRST_BLOCK_BEGINNING_PATTERN 29... 30MY_FIRST_BLOCK_ENDING_PATTERN 31 32MY_SECOND_BLOCK_BEGINNING_PATTERN 33... 34MY_SECOND_BLOCK_ENDING_PATTERN 35... ``` -------------------------------- ### Example File Content for RegisterFile Source: https://rjmalves.github.io/cfinterface/getting_started/tutorial.html Sample data structure representing a file with single-line blocks suitable for the RegisterFile abstraction. ```text DATA_HIGH ID001 sudo.user 10/20/2025 901.25 DATA_HIGH ID002 sudo.user 10/21/2025 100.20 DATA_HIGH ID003 test.user 10/30/2025 100.20 DATA_LOW 01/01/2024 105.23 DATA_LOW 01/02/2024 29.15 DATA_LOW 01/03/2024 5.05 ``` -------------------------------- ### BinaryRepository(content, wrap_io=False, *args) Source: https://rjmalves.github.io/cfinterface/reference/adapters/files/reading.html Initializes a repository specifically for binary content. ```APIDOC ## BinaryRepository ### Signature `class BinaryRepository(content: str | bytes, wrap_io: bool = False, *args: Any)` ### Description Repository adapter designed for handling binary data. ``` -------------------------------- ### Build documentation locally Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/contributing.rst.txt Generates the project documentation in HTML format using Sphinx. ```bash uv run sphinx-build -W -b html docs/source docs/build/html ``` -------------------------------- ### Write and Read Sample File Source: https://rjmalves.github.io/cfinterface/_downloads/3f729c894635516594e180c04c0a30c4/plot_register_file.ipynb Perform a round-trip by writing sample content to a temporary file, reading it back, and verifying the output. ```python import tempfile from pathlib import Path # Field positions (0-indexed in the full line): # DataHigh: ID@11(8) User@19(11) Date@30(12) Value@42(8) # DataLow: Date@11(12) Value@23(10) SAMPLE_CONTENT = ( "DATA_HIGH ID001 sudo.user 10/20/2025 901.25\n" "DATA_HIGH ID002 sudo.user 10/21/2025 100.20\n" "DATA_HIGH ID003 test.user 10/30/2025 100.20\n" "\n" "DATA_LOW 01/01/2024 105.23\n" "DATA_LOW 01/02/2024 29.15\n" "DATA_LOW 01/03/2024 5.05\n" ) with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "registers.txt" path.write_text(SAMPLE_CONTENT) file = MyRegisterFile.read(str(path)) # Inspect parsed registers print(f"DataHigh count: {len(file.data_high)}") for reg in file.data_high: print(f" {reg.field_id} {reg.user} {reg.date} {reg.value}") print(f"\nDataLow count: {len(file.data_low)}") for reg in file.data_low: print(f" {reg.date} {reg.value}") # Write back and verify round-trip out_path = Path(tmpdir) / "registers_out.txt" file.write(str(out_path)) print(f"\nRound-trip successful: {out_path.exists()}") ``` -------------------------------- ### Repository(content, wrap_io=False, *args) Source: https://rjmalves.github.io/cfinterface/reference/adapters/files/reading.html Initializes a generic repository for reading content. ```APIDOC ## Repository ### Signature `class Repository(content: str | bytes, wrap_io: bool = False, *args: Any)` ### Description Base class for repository adapters used to read content. ``` -------------------------------- ### Define a LiteralField Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Example of defining a LiteralField to extract a specific data value from a line based on size and position. ```none | Username (max. 30 chars) | ... | FooBar | ... ``` ```python from cfinterface import LiteralField username = LiteralField(size=30, starting_position=1) value = username.read("| FooBar | ...") # The content "FooBar" can be accessed by both value and username.value assert value == "FooBar" assert username.value == "FooBar" ``` -------------------------------- ### Parse Delimiter-Separated Data Source: https://rjmalves.github.io/cfinterface/_downloads/6435b8aa02809e17ffeb2541166ddaaa/plot_tabular_parsing.ipynb Parse CSV-style data by providing a delimiter to the TabularParser. Starting positions are ignored in this mode. ```python delimited_columns = [ ColumnDef(name="Name", field=LiteralField(size=20, starting_position=0)), ColumnDef( name="Score", field=FloatField(size=10, starting_position=0, decimal_digits=2), ), ColumnDef(name="Grade", field=LiteralField(size=2, starting_position=0)), ] csv_parser = TabularParser(delimited_columns, delimiter=",") csv_lines = [ "Alice,95.50,A\n", "Bob,87.25,B\n", "Charlie,72.00,C\n", ] csv_data = csv_parser.parse_lines(csv_lines) print("\n=== Delimiter-Separated Parsing ===") for i, name in enumerate(csv_data["Name"]): print( f" {name}: score={csv_data['Score'][i]}, grade={csv_data['Grade'][i]}" ) ``` -------------------------------- ### Handle pandas dependency in code Source: https://rjmalves.github.io/cfinterface/_sources/guides/migration-v1.9.rst.txt Ensures pandas is explicitly declared or installed via the extra when using DataFrame integration. ```python # Before: implicit import that worked because cfinterface brought in pandas import pandas as pd df = pd.DataFrame(file.data) # After: install cfinterface[pandas] or declare pandas as your own # project dependency import pandas as pd # requires pip install cfinterface[pandas] df = pd.DataFrame(file.data) ``` -------------------------------- ### Reading and Writing Block Files Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Demonstrates the basic workflow for reading a file, asserting data types, and writing the content back. ```python file = MyBlockFile.read("/path/to/file_describe_above.txt") assert type(file.first_block_data) == pd.DataFrame file.write("/path/to/some_other_desired_file.txt") # The content of the written file should be the same # as the source file ``` -------------------------------- ### Defining Block Data Access Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Example of accessing block data as a pandas DataFrame within a custom block class. ```python @property def first_block_data(self) -> pd.DataFrame: block_dfs = [b.data for b in self.data.get_blocks_of_type(FirstBlock)] return pd.concat(block_dfs, ignore_index=True) ``` -------------------------------- ### Implement Block and BlockFile Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Shows how to subclass Block to define custom parsing logic and use BlockFile to manage multiple block types. ```python from typing import IO import pandas as pd from cfinterface import IntegerField, FloatField, DatetimeField, LiteralField from cfinterface import Line, Block, BlockFile class FirstBlock(Block): __slots__ = [ "__header_lines", "__line_model", ] BEGIN_PATTERN = "MY_FIRST_BLOCK_BEGINNING_PATTERN" END_PATTERN = "MY_FIRST_BLOCK_ENDING_PATTERN" NUM_HEADER_LINES = 5 def __init__(self, previous=None, next=None, data=None) -> None: super().__init__(previous, next, data) self.__header_lines = [] date_field = DatetimeField(size=7, starting_position=0, format="%y/%m") index_field = IntegerField(size=4, starting_position=11) value_field = FloatField(size=6, starting_position=19) self.__line_model = Line([date_field, index_field, value_field]) def __eq__(self, o: object) -> bool: if not isinstance(o, FirstBlock): return False block: FirstBlock = o if not all( [ isinstance(self.data, pd.DataFrame), isinstance(block.data, pd.DataFrame), ] ): return False else: return self.data.equals(block.data) # Override def read(self, file: IO, *args, **kwargs) -> bool: # Discards the line with the beginning pattern file.readline() # Reads header lines for writing later for _ in range(self.__class__.NUM_HEADER_LINES): self.__header_lines.append(file.readline()) # Reads the data content dates = [] indices = [] values = [] while True: line = file.readline() if FirstBlock.ends(line): self.data = pd.DataFrame({"Date": dates, "Index": indices, "Value": values}) break date, index, value = self.__line_model.read(line) dates.append(date) indices.append(index) values.append(value) # Override def write(self, file: IO, *args, **kwargs): file.write(self.__class__.BEGIN_PATTERN + "\n") # Writes header lines for line in self.__header_lines: file.write(line) # Writes data lines for _, line in self.data.iterrows(): self.__line_model.write([line["Date"], line["Index"], line["Value"]]) file.write(self.__class__.END_PATTERN + "\n") class SecondBlock(Block): # Implement in a similar way for the second block specifics class MyBlockFile(BlockFile): BLOCKS = [ FirstBlock, SecondBlock, ] # All the reading and writing logic is done by the framework, # finding when each block begins and calling their implemetations. # The user can implement some properties for better suiting its use cases: ``` -------------------------------- ### Raw Data Block Structure Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Example of a file format containing repeating blocks identified by specific beginning and ending patterns. ```none MY_FIRST_BLOCK_BEGINNING_PATTERN I have some raw text for describing my content, because I was mean for being irectly read by someone. Now I have some data. After which I will be done. Date Index Value 2020/01 1 1000.0 2020/02 1 2000.0 2020/01 2 3000.0 2020/02 2 4000.0 2020/01 3 5000.0 2020/02 3 6000.0 MY_FIRST_BLOCK_ENDING_PATTERN MY_SECOND_BLOCK_BEGINNING_PATTERN My content is completely different from the previous block... Username Last Login admin 1996/01/01 sunshine 2000/01/01 pineapple 1996/01/01 admin 1996/01/01 MY_SECOND_BLOCK_ENDING_PATTERN MY_FIRST_BLOCK_BEGINNING_PATTERN ... MY_FIRST_BLOCK_ENDING_PATTERN MY_FIRST_BLOCK_BEGINNING_PATTERN ... MY_FIRST_BLOCK_ENDING_PATTERN MY_SECOND_BLOCK_BEGINNING_PATTERN ... MY_SECOND_BLOCK_ENDING_PATTERN ... ``` -------------------------------- ### TextualRepository(content, wrap_io=False, encoding='utf-8', *args) Source: https://rjmalves.github.io/cfinterface/reference/adapters/files/reading.html Initializes a repository for textual content with encoding support. ```APIDOC ## TextualRepository ### Signature `class TextualRepository(content: str, wrap_io: bool = False, encoding: str = 'utf-8', *args: Any)` ### Description Repository adapter designed for handling textual data with configurable encoding. ``` -------------------------------- ### Define and use a TabularParser Source: https://rjmalves.github.io/cfinterface/_sources/guides/architecture.rst.txt Demonstrates creating a parser with specific column definitions and parsing a list of strings into a dictionary. ```python from cfinterface.components.tabular import TabularParser, ColumnDef from cfinterface.components.literalfield import LiteralField from cfinterface.components.floatfield import FloatField columns = [ ColumnDef(name="name", field=LiteralField(size=20, starting_position=0)), ColumnDef(name="value", field=FloatField(size=10, starting_position=20, decimal_digits=2)), ] parser = TabularParser(columns) lines = [ "Product A 12.50 ", "Product B 7.99 ", ] data = parser.parse_lines(lines) # data == {"name": ["Product A", "Product B"], "value": [12.5, 7.99]} ``` -------------------------------- ### BinaryRepository Implementation Source: https://rjmalves.github.io/cfinterface/_modules/cfinterface/adapters/components/line/repository.html Concrete implementation for handling binary-based lines. ```python [docs] class BinaryRepository(Repository): def read( self, line: Any, delimiter: str | bytes | None = None, ) -> list[Any]: line_bytes: bytes = ( line if isinstance(line, bytes) else line.encode("utf-8") ) for field in self._fields: field.read(line_bytes) # type: ignore[arg-type] return self.values def write( self, values: list[Any], delimiter: str | bytes | None = None ) -> bytes: line = b"" self.values = values for field in self._fields: line = field.write(line) return line ``` -------------------------------- ### Execute Read/Write Round-Trip Source: https://rjmalves.github.io/cfinterface/examples/plot_register_file.html Parse sample content from a file and verify that writing the parsed data back produces identical output. ```python import tempfile from pathlib import Path # Field positions (0-indexed in the full line): # DataHigh: ID@11(8) User@19(11) Date@30(12) Value@42(8) # DataLow: Date@11(12) Value@23(10) SAMPLE_CONTENT = ( "DATA_HIGH ID001 sudo.user 10/20/2025 901.25\n" "DATA_HIGH ID002 sudo.user 10/21/2025 100.20\n" "DATA_HIGH ID003 test.user 10/30/2025 100.20\n" "\n" "DATA_LOW 01/01/2024 105.23\n" "DATA_LOW 01/02/2024 29.15\n" "DATA_LOW 01/03/2024 5.05\n" ) with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "registers.txt" path.write_text(SAMPLE_CONTENT) file = MyRegisterFile.read(str(path)) # Inspect parsed registers print(f"DataHigh count: {len(file.data_high)}") for reg in file.data_high: print(f" {reg.field_id} {reg.user} {reg.date} {reg.value}") print(f"\nDataLow count: {len(file.data_low)}") for reg in file.data_low: print(f" {reg.date} {reg.value}") # Write back and verify round-trip out_path = Path(tmpdir) / "registers_out.txt" file.write(str(out_path)) print(f"\nRound-trip successful: {out_path.exists()}") ``` ```text DataHigh count: 3 ID001 sudo.user 2025-10-20 00:00:00 901.25 ID002 sudo.user 2025-10-21 00:00:00 100.2 ID003 test.user 2025-10-30 00:00:00 100.2 DataLow count: 3 2024-01-01 00:00:00 105.23 2024-01-02 00:00:00 29.15 2024-01-03 00:00:00 5.05 Round-trip successful: True ``` -------------------------------- ### Implement File Versioning Source: https://rjmalves.github.io/cfinterface/_sources/guides/faq.rst.txt Map version keys to component lists in VERSIONS to support multiple schema versions within one class. ```python from cfinterface.files.registerfile import RegisterFile from cfinterface.storage import StorageType class VersionedFile(RegisterFile): REGISTERS = [RecordV2A, RecordV2B] # default schema (most recent) VERSIONS = { "1.0": [RecordV1A], "2.0": [RecordV2A, RecordV2B], } STORAGE = StorageType.TEXT # Selects the schema of the most recent version <= "1.5", i.e., "1.0" file = VersionedFile.read("/path/to/file.txt", version="1.5") ``` -------------------------------- ### Define and Read a Line Model Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Demonstrates how to define a line structure with specific fields and read values from a formatted string. ```python line = Line(fields=[username, signup_date, age, balance]) values = line.read("| FooBar | 2020-05-20 | 18 | 99.90 | ...") assert values == ["FooBar", datetime(2020, 5, 20), 18, 99.90] ``` -------------------------------- ### Implement and use versioned file classes Source: https://rjmalves.github.io/cfinterface/guides/faq.html Define a class inheriting from RegisterFile with a VERSIONS dictionary to map version keys to record types. Use the read method with a version argument to select the appropriate schema. ```python from cfinterface.files.registerfile import RegisterFile from cfinterface.storage import StorageType class VersionedFile(RegisterFile): REGISTERS = [RecordV2A, RecordV2B] # default schema (most recent) VERSIONS = { "1.0": [RecordV1A], "2.0": [RecordV2A, RecordV2B], } STORAGE = StorageType.TEXT # Selects the schema of the most recent version <= "1.5", i.e., "1.0" file = VersionedFile.read("/path/to/file.txt", version="1.5") # Validates whether the read content matches the expected schema result = file.validate(version="1.0") print(result.matched) # True if all expected types were found print(result.missing_types) # expected types that did not appear in the file ``` -------------------------------- ### Use factory methods Source: https://rjmalves.github.io/cfinterface/reference/adapters/files/line.html Factory methods to instantiate specific repository types based on the provided kind. ```python factory(_kind : Literal['TEXT']_) → type[TextualRepository][source] ``` ```python factory(_kind : Literal['BINARY']_) → type[BinaryRepository] ``` ```python factory(_kind : str | StorageType_) → type[Repository] ``` -------------------------------- ### Define the versioned file Source: https://rjmalves.github.io/cfinterface/_downloads/29750da81eab0b9bf38b43fc5a450c6e/plot_versioned_file.ipynb Maps version keys to component lists and sets the default schema using the VERSIONS and BLOCKS attributes. ```python class MyVersionedFile(BlockFile): VERSIONS = { "1.0": [ConstantBlock, OldDataBlock], "2.0": [ConstantBlock, NewDataBlock], } BLOCKS = [ConstantBlock, NewDataBlock] ``` -------------------------------- ### Define the File Model Source: https://rjmalves.github.io/cfinterface/_downloads/3f729c894635516594e180c04c0a30c4/plot_register_file.ipynb Create a RegisterFile subclass that lists the expected register types. ```python class MyRegisterFile(RegisterFile): REGISTERS = [DataHigh, DataLow] @property def data_high(self) -> list[DataHigh] | None: return self.data.get_registers_of_type(DataHigh) @property def data_low(self) -> list[DataLow] | None: return self.data.get_registers_of_type(DataLow) ``` -------------------------------- ### TextualRepository Implementation Source: https://rjmalves.github.io/cfinterface/_modules/cfinterface/adapters/components/line/repository.html Concrete implementation for handling text-based lines, supporting both positional and delimited formats. ```python [docs] class TextualRepository(Repository): def __positional_to_delimited_field(self, f: Field) -> Field: f.ending_position = f.size f.starting_position = 0 return f def __positional_reading(self, line: str) -> list[Any]: for field in self._fields: field.read(line) return self.values def __delimted_reading(self, line: str, delimiter: str) -> list[Any]: fields = [self.__positional_to_delimited_field(f) for f in self._fields] values = [v.strip() for v in line.split(delimiter)] for field, value in zip(fields, values, strict=False): field.read(value) return self.values def read( self, line: Any, delimiter: str | bytes | None = None, ) -> list[Any]: line_str: str = line if isinstance(line, str) else line.decode("utf-8") if isinstance(delimiter, str): return self.__delimted_reading(line_str, delimiter) return self.__positional_reading(line_str) def __positional_writing(self, values: list[Any]) -> str: line = "" self.values = values for field in self._fields: line = field.write(line) return line + "\n" def __delimted_writing(self, values: list[Any], delimiter: str) -> str: fields = [self.__positional_to_delimited_field(f) for f in self._fields] self.values = values separated = [field.write("").strip() for field in fields] return delimiter.join(separated) + "\n" def write( self, values: list[Any], delimiter: str | bytes | None = None ) -> str: if isinstance(delimiter, str): return self.__delimted_writing(values, delimiter) return self.__positional_writing(values) ``` -------------------------------- ### Implement a RegisterFile Source: https://rjmalves.github.io/cfinterface/_sources/guides/architecture.rst.txt Define a file class by specifying the registers and storage configuration, then perform read/write operations. ```python from cfinterface.files.registerfile import RegisterFile from cfinterface.storage import StorageType class MyFile(RegisterFile): REGISTERS = [MonthlyValue] STORAGE = StorageType.TEXT ENCODING = "utf-8" file = MyFile.read("/path/to/file.txt") file.write("/path/to/output.txt") ``` -------------------------------- ### Implement Versioned Files Source: https://rjmalves.github.io/cfinterface/_sources/guides/architecture.rst.txt Define a file class with multiple versions to allow schema selection at read time without creating separate classes. ```python from cfinterface.files.registerfile import RegisterFile from cfinterface.storage import StorageType class VersionedFile(RegisterFile): REGISTERS = [MonthlyValueV2] VERSIONS = { "1.0": [MonthlyValueV1], "2.0": [MonthlyValueV2], } STORAGE = StorageType.TEXT # Reading while selecting a version without mutating the class file = VersionedFile.read("/path/to/file.txt", version="1.5") # resolve_version("1.5", VERSIONS) will return the components for "1.0" # Validating the read content result = file.validate(version="1.0") print(result.matched) # True if the content matches the 1.0 schema ``` -------------------------------- ### Define the file model Source: https://rjmalves.github.io/cfinterface/_sources/examples/plot_section_file.rst.txt Create a SectionFile subclass that specifies the order of sections and provides convenience properties to access them. ```Python class MyTabularFile(SectionFile): SECTIONS = [HeaderSection, TableSection] @property def header(self) -> HeaderSection | None: return self.data.get_sections_of_type(HeaderSection) @property def table(self) -> TableSection | None: return self.data.get_sections_of_type(TableSection) ``` -------------------------------- ### Run linting and type checking Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/contributing.rst.txt Manually triggers ruff for linting and formatting, and mypy for static type checking. ```bash uv run ruff check ./cfinterface uv run ruff format --check ./cfinterface uv run mypy ./cfinterface ``` -------------------------------- ### Sample output of the read operation Source: https://rjmalves.github.io/cfinterface/examples/plot_block_file.html The expected console output after reading and processing the sample data. ```text === Header === autor: joao.silva versao: 2.1 data: 15/03/2025 === Data Blocks (2) === Block 1 (3 records): id=001 nome=Produto A preco=49.90 id=002 nome=Produto B preco=129.00 id=003 nome=Produto C preco=15.50 Block 2 (2 records): id=101 nome=Servico X preco=299.00 id=102 nome=Servico Y preco=199.00 Round-trip: 14 lines -> 14 lines ``` -------------------------------- ### Implementing Block and BlockFile for Custom File Parsing Source: https://rjmalves.github.io/cfinterface/getting_started/tutorial.html Define custom block structures by inheriting from Block and aggregate them into a BlockFile. The framework handles the orchestration of reading and writing based on defined patterns. ```python 1 from typing import IO 2 import pandas as pd 3 4 from cfinterface import IntegerField, FloatField, DatetimeField, LiteralField 5 from cfinterface import Line, Block, BlockFile 6 7 class FirstBlock(Block): 8 9 __slots__ = [ 10 "__header_lines", 11 "__line_model", 12 ] 13 14 BEGIN_PATTERN = "MY_FIRST_BLOCK_BEGINNING_PATTERN" 15 END_PATTERN = "MY_FIRST_BLOCK_ENDING_PATTERN" 16 17 NUM_HEADER_LINES = 5 18 19 def __init__(self, previous=None, next=None, data=None) -> None: 20 super().__init__(previous, next, data) 21 self.__header_lines = [] 22 date_field = DatetimeField(size=7, starting_position=0, format="%y/%m") 23 index_field = IntegerField(size=4, starting_position=11) 24 value_field = FloatField(size=6, starting_position=19) 25 self.__line_model = Line([date_field, index_field, value_field]) 26 27 def __eq__(self, o: object) -> bool: 28 if not isinstance(o, FirstBlock): 29 return False 30 block: FirstBlock = o 31 if not all( 32 [ 33 isinstance(self.data, pd.DataFrame), 34 isinstance(block.data, pd.DataFrame), 35 ] 36 ): 37 return False 38 else: 39 return self.data.equals(block.data) 40 41 # Override 42 def read(self, file: IO, *args, **kwargs) -> bool: 43 44 # Discards the line with the beginning pattern 45 file.readline() 46 47 # Reads header lines for writing later 48 for _ in range(self.__class__.NUM_HEADER_LINES): 49 self.__header_lines.append(file.readline()) 50 51 # Reads the data content 52 dates = [] 53 indices = [] 54 values = [] 55 56 while True: 57 58 line = file.readline() 59 if FirstBlock.ends(line): 60 self.data = pd.DataFrame({"Date": dates, "Index": indices, "Value": values}) 61 break 62 63 date, index, value = self.__line_model.read(line) 64 dates.append(date) 65 indices.append(index) 66 values.append(value) 67 68 # Override 69 def write(self, file: IO, *args, **kwargs): 70 71 file.write(self.__class__.BEGIN_PATTERN + "\n") 72 73 # Writes header lines 74 for line in self.__header_lines: 75 file.write(line) 76 77 # Writes data lines 78 for _, line in self.data.iterrows(): 79 self.__line_model.write([line["Date"], line["Index"], line["Value"]]) 80 81 file.write(self.__class__.END_PATTERN + "\n") 82 83 84 class SecondBlock(Block): 85 # Implement in a similar way for the second block specifics 86 87 88 class MyBlockFile(BlockFile): 89 90 BLOCKS = [ 91 FirstBlock, 92 SecondBlock, 93 ] 94 95 # All the reading and writing logic is done by the framework, 96 # finding when each block begins and calling their implemetations. 97 # The user can implement some properties for better suiting its use cases: 98 99 @property 100 def first_block_data(self) -> pd.DataFrame: 101 block_dfs = [b.data for b in self.data.get_blocks_of_type(FirstBlock)] 102 return pd.concat(block_dfs, ignore_index=True) 103 104 file = MyBlockFile.read("/path/to/file_describe_above.txt") 105 assert type(file.first_block_data) == pd.DataFrame 106 file.write("/path/to/some_other_desired_file.txt") 107 # The content of the written file should be the same 108 # as the source file ``` -------------------------------- ### Specify Storage Type Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Configures a file model to handle either text or binary data. ```python from cfinterface import BlockFile from cfinterface.storage import StorageType class MyTextualFile(BlockFile): STORAGE = StorageType.TEXT class MyBinaryFile(BlockFile): STORAGE = StorageType.BINARY ``` -------------------------------- ### Run pre-commit hooks manually Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/contributing.rst.txt Executes all configured pre-commit hooks against all files in the repository. ```bash uv run pre-commit run --all-files ``` -------------------------------- ### Component hierarchy diagram Source: https://rjmalves.github.io/cfinterface/_sources/guides/architecture.rst.txt Visual representation of the framework's component dependency structure. ```text Field (FloatField, IntegerField, LiteralField, DatetimeField) | v Line (ordered sequence of Fields; delegates I/O to the adapter) | v Register / Block / Section (intermediate components; operate on file handles) | v RegisterFile / BlockFile / SectionFile (high-level file classes) ``` -------------------------------- ### Write Block File to Disk Source: https://rjmalves.github.io/cfinterface/_downloads/e7adef7c3e69b33c01aad59ee4550c16/plot_block_file.ipynb Demonstrates writing a parsed block file object directly to a specified file path on the filesystem. ```python with tempfile.TemporaryDirectory() as tmpdir: input_path = Path(tmpdir) / "input.txt" output_path = Path(tmpdir) / "output.txt" input_path.write_text(EXAMPLE_CONTENT, encoding="utf-8") file_obj2 = MyRecordFile.read(str(input_path)) file_obj2.write(str(output_path)) print(f"\nOutput file created: {output_path.exists()}") print(f"Size: {output_path.stat().st_size} bytes") ``` -------------------------------- ### Implement RegisterFile Class Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/tutorial.rst.txt Define a custom RegisterFile class with register types and accessors. ```python class MyRegisterFile(RegisterFile): REGISTERS = [ DataHigh, DataLow, ] # All the reading and writing logic is done by the framework, # finding which register is in each line and calling their implemetations. # The user can implement some properties for better suiting its use cases: @property def data_high(self) -> DataHigh | list[DataHigh] | None: return self.data.get_registers_of_type(DataHigh) @property def data_low(self) -> DataLow | list[DataLow] | None: return self.data.get_registers_of_type(DataLow) ``` -------------------------------- ### Repository Base and Implementations Source: https://rjmalves.github.io/cfinterface/_modules/cfinterface/adapters/writing/repository.html Defines the abstract base class Repository and its concrete subclasses for binary and text-based file writing. These classes manage file lifecycle through context manager methods. ```python from abc import ABC, abstractmethod from typing import ( IO, Any, BinaryIO, Literal, TextIO, Union, overload, ) from cfinterface.storage import StorageType [docs] class Repository(ABC): __slots__ = ["_to", "_wrap_io"] def __init__(self, to: str | IO[Any], *args: Any) -> None: self._to = to self._wrap_io = isinstance(to, str) def __enter__(self) -> "Repository": return self def __exit__(self, *args: Any) -> None: # noqa: B027 pass @abstractmethod def write(self, data: str | bytes) -> None: raise NotImplementedError @property @abstractmethod def file(self) -> IO[Any]: raise NotImplementedError [docs] class BinaryRepository(Repository): __slots__ = ["_filepointer"] def __init__(self, path: str | IO[Any], *args: Any) -> None: super().__init__(path) self._filepointer: BinaryIO = None # type: ignore[assignment] def __enter__(self) -> "BinaryRepository": self._filepointer = ( open(self._to, "wb") # type: ignore[arg-type] if self._wrap_io else self._to # type: ignore[assignment] ) super().__enter__() return self def __exit__(self, *args: Any) -> None: super().__exit__(*args) if self._wrap_io: self._filepointer.close() def write(self, data: str | bytes) -> None: if isinstance(data, bytes): self._filepointer.write(data) @property def file(self) -> BinaryIO: return self._filepointer [docs] class TextualRepository(Repository): __slots__ = ["_filepointer", "_encoding"] def __init__(self, path: str | IO[Any], encoding: str = "utf-8") -> None: super().__init__(path) self._filepointer: TextIO = None # type: ignore[assignment] self._encoding = encoding def __enter__(self) -> "TextualRepository": self._filepointer = ( open(self._to, "w", encoding=self._encoding) # type: ignore[arg-type] if self._wrap_io else self._to # type: ignore[assignment] ) super().__enter__() return self def __exit__(self, *args: Any) -> None: super().__exit__(*args) if self._wrap_io: self._filepointer.close() def write(self, data: str | bytes) -> None: if isinstance(data, str): self._filepointer.write(data) @property def file(self) -> TextIO: return self._filepointer @overload def factory(kind: Literal["TEXT"]) -> type[TextualRepository]: ... @overload def factory(kind: Literal["BINARY"]) -> type[BinaryRepository]: ... @overload def factory(kind: str | StorageType) -> type[Repository]: ... [docs] def factory(kind: Union[str, "StorageType"]) -> type[Repository]: mappings: dict[str | StorageType, type[Repository]] = { StorageType.TEXT: TextualRepository, StorageType.BINARY: BinaryRepository, } return mappings.get(kind, TextualRepository) ``` -------------------------------- ### Write Block File to Path Source: https://rjmalves.github.io/cfinterface/_sources/examples/plot_block_file.rst.txt Shows how to write a MyRecordFile object directly to a specified file path. ```Python with tempfile.TemporaryDirectory() as tmpdir: input_path = Path(tmpdir) / "input.txt" output_path = Path(tmpdir) / "output.txt" input_path.write_text(EXAMPLE_CONTENT, encoding="utf-8") file_obj2 = MyRecordFile.read(str(input_path)) file_obj2.write(str(output_path)) print(f"\nOutput file created: {output_path.exists()}") print(f"Size: {output_path.stat().st_size} bytes") ``` ```none Output file created: True Size: 299 bytes ``` -------------------------------- ### RegisterFile.read Source: https://rjmalves.github.io/cfinterface/reference/registers/files/registerfile.html Reads a file from a path or buffer and returns a RegisterFile instance. ```APIDOC ## RegisterFile.read(content, *args, version=None, **kwargs) ### Description Reads from a file path or buffer. The version parameter selects a component set from VERSIONS without mutating the class. ### Parameters - **content** (str | bytes) - Required - The file path or buffer content to read. - **version** (str | None) - Optional - Version key to select a component set. ### Returns - **RegisterFile** - An instance of the RegisterFile class. ``` -------------------------------- ### Run project tests Source: https://rjmalves.github.io/cfinterface/_sources/getting_started/contributing.rst.txt Executes the test suite using pytest. ```bash uv run pytest ./tests ``` -------------------------------- ### Iterate Over File Sections Source: https://rjmalves.github.io/cfinterface/_sources/examples/plot_section_file.rst.txt Demonstrates how to iterate through all sections contained within a file object. ```Python with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "report.txt" path.write_text(EXAMPLE_CONTENT, encoding="utf-8") file_obj3 = MyTabularFile.read(str(path)) print("\n=== All sections ===") for sec in file_obj3.data: print(f" {type(sec).__name__}") ``` ```none === All sections === DefaultSection HeaderSection TableSection ``` -------------------------------- ### Read, Validate, and Resolve Versioned Files Source: https://rjmalves.github.io/cfinterface/_downloads/29750da81eab0b9bf38b43fc5a450c6e/plot_versioned_file.ipynb Reads a versioned file from a temporary directory, validates its structure against a specific version, and demonstrates version resolution. ```python V2_CONTENT = ( "FILE_HEADER\n" "My Versioned File v2\n" "END_FILE_HEADER\n" "DATA_BLOCK\n" " 1 Alice 95.50\n" " 2 Bob 87.25\n" "END_DATA\n" ) with tempfile.TemporaryDirectory() as tmpdir: v2_path = Path(tmpdir) / "v2.txt" v2_path.write_text(V2_CONTENT) file_v2 = MyVersionedFile.read(str(v2_path), version="2.0") print("=== Version 2.0 Read ===") for block in file_v2.data: print(f" {type(block).__name__}: {block.data}") # %% # Validate version match # ----------------------- # After reading, ``validate()`` checks if the parsed data matches # the expected schema for a given version. result = file_v2.validate(version="2.0") print("\n=== Version Validation ===") print(f" Matched: {result.matched}") print(f" Expected: {[t.__name__ for t in result.expected_types]}") print(f" Found: {[t.__name__ for t in result.found_types]}") print(f" Default ratio: {result.default_ratio:.1%}") # %% # resolve_version utility # ------------------------ # Use ``resolve_version()`` to see which version key is selected # for a given request. components = resolve_version("1.5", MyVersionedFile.VERSIONS) resolved_names = [t.__name__ for t in components] if components else [] print("\n=== resolve_version('1.5') ===") print(f" Resolved to v1.0 components: {resolved_names}") ``` -------------------------------- ### Aggregate fields into a Line Source: https://rjmalves.github.io/cfinterface/_sources/guides/architecture.rst.txt Shows how to group multiple fields into a Line component and perform a read operation using a specified storage type. ```python from cfinterface import LiteralField, FloatField from cfinterface.components.line import Line from cfinterface.storage import StorageType fields = [ LiteralField(size=20, starting_position=0), FloatField(size=10, starting_position=20, decimal_digits=2), ] line = Line(fields, storage=StorageType.TEXT) values = line.read("Current Account -1234.56 ") # values == ["Current Account", -1234.56] ```