Reference

Reference of the smos-walker project

smos_walker.api

Entrypoint to smos_walker

The module's facade does expose the following functions:

• index_datablock - Index a datablock (DBL) according to its schema.
• create_query - Query object over an indexed datablock.

smos_walker.api.facade

SmosWalker

Instantiate an object to help manipulating datablocks

See test_smoswalker_highlevel_api for a practical example of usage.

Attributes:

Name	Type	Description
xsd_path	str	Path pointing towards the XSD file, also known as "binx.xsd".
xml_schema_path	str	Path pointing towards an XML schema file.
datablock_folder_path	str | None	Path pointing towards the folder containing the data block.
statically_decorated_tree	BaseNode	A statically-decorated tree. Does not require a datablock. Corresponds to "Step 2"
indexed_datablock	IndexedDataBlock | None	The indexed datablock

Source code in smos_walker\api\facade.py

class SmosWalker:
    """Instantiate an object to help manipulating datablocks

    See `test_smoswalker_highlevel_api` for a practical example of usage.

    Attributes:
        xsd_path: Path pointing towards the XSD file, also known as "binx.xsd".
        xml_schema_path: Path pointing towards an XML schema file.
        datablock_folder_path: Path pointing towards the folder containing the data block.

        statically_decorated_tree: A statically-decorated tree. Does not require a datablock. Corresponds to "Step 2"
        indexed_datablock: The indexed datablock
    """

    def __init__(
        self,
        xsd_path: str,
        xml_schema_path: str,
        datablock_folder_path: str | None = None,
    ):
        """Constructor

        Args:
            xsd_path: Path pointing towards the XSD file, also known as "binx.xsd".
            xml_schema_path: Path pointing towards an XML schema file.
            datablock_folder_path: Path pointing towards the folder containing the data block.
        """
        self.xsd_path: str = xsd_path  # Store here the xsd content
        self.xml_schema_path: str = xml_schema_path  # Store here the binXschema content

        # Keep a plain statically decorated tree, as a reference
        # It is not reused when indexing datablock as dynamic decoration occurs in-place.
        # For now, the static analysis is just remade when needed.
        # It can be printed for information with `get_dynamically_decorated_tree_text_representation`
        self.statically_decorated_tree: BaseNode = load_statically_decorated_tree(
            xsd_path, xml_schema_path
        )

        # Annotate nodes with paths
        self.statically_decorated_tree.accept(DictIndexNodeVisitor())

        self.datablock_folder_path: str | None = None
        self.indexed_datablock: IndexedDataBlock | None = None

        # If the datablock is already provided, immediately index it
        if datablock_folder_path:
            self.set_datablock(datablock_folder_path)

    def set_datablock(self, datablock_folder_path: str):
        """Sets the datablock path and index it

        Args:
            datablock_folder_path: Path pointing towards the folder containing the data block.
        """
        self.datablock_folder_path = datablock_folder_path
        self._index_datablock()

    def query_timed(
        self, path: str, *coordinates: int
    ) -> NumpifiedDataBlockType | None:
        start = time.perf_counter()
        result = self.query(path, *coordinates)
        end = time.perf_counter()
        logging.info(f"Query execution time: {end - start:.2f} seconds")
        return result

    def query(self, path: str, *coordinates: int) -> NumpifiedDataBlockType | None:
        """Fetch data from the index datablock

        Args:
            path: Slash-separated path pointing to a node in the typesystem hierarchy.
                To see a list of available paths, use str_query(path).
            coordinates: Optional extra coordinates to uniquely identify a node
                (some type hierarchies have a high-dimensionality, usually due to the nested
                use of Array Variables in the schema description)
        Raises:
            SmosWalkerException: error message containing help to uniquely identify a node in the typesystem hierarchy.

        Returns:
            A Numpy's ndarray if the node could be located in the typesystem and the datablock.
        """
        self._require_indexed_datablock()

        # Allow user to quickly give extra dynamic coordinates after failure to query the datablock.
        return self.create_query(path).consume_query(*coordinates)

    @property
    def datablock_info(self):
        """Information about the indexed datablock"""
        return self.str_indexed_datablock()

    def str_indexed_datablock(self):
        return str(self.indexed_datablock)

    @property
    def paths(self):
        return list(
            render_path(path) for path in self.indexed_datablock.all_available_paths
        )

    @property
    def type_info(self):
        """Information about a path in the typesystem hierarchy"""
        return self.get_type_info()

    def get_type_info(self, path: str | None = None):
        """Information about a path in the typesystem hierarchy"""
        return self.str_query(path)

    def str_query(self, path: str | None) -> str:
        return str(self.create_query(path))

    @property
    def static_tree(self) -> str:
        return self.str_static_tree()

    def str_static_tree(self) -> str:
        """Print a user-friendly representation of the statically decorated tree (aka Step 2 in the XML schema parsing)

        Returns:
            text representation of the statically decorated tree
        """
        return self._get_statically_decorated_tree_text_representation()

    @property
    def info(self) -> str:
        return self.get_info()

    def _repr_pretty_(self, pretty_printer, _):
        paths = "\n".join(self.paths)
        static_tree_details = output_tree_to_txt(
            self.get_typesystem_node(None),
            write_dimension_name_flag=True,
            write_dynamic_attributes=True,
            write_primitive_type=True,
            write_path=True,
            write_static=True,
            write_block_byte_size=True,
            ignore_static_nodes=False,
            ignore_non_dimensional_nodes=False,
        )
        return pretty_printer.text(
            f"""# SmosWalker Information

## Indexed DataBlock

{self.str_indexed_datablock()}

## Static Tree

### Summary

{self.str_static_tree()}

### Details

{static_tree_details}

## Dynamic Tree

{self.str_dynamic_tree()}

## Paths

{paths}
"""
        )

    def get_info(self, path: str | None = None) -> str:
        """alias for`str_dynamic_tree`"""
        return self.str_dynamic_tree(path)

    def str_dynamic_tree(self, path: str | None = None) -> str:
        """Print a user-friendly representation of the dynamically decorated tree (aka Step 3 in the XML schema parsing)

        Requires an indexed datablock. Use `set_datablock` if no datablock is set yet.

        Args:
            path: Slash-separated path pointing to a node in the typesystem hierarchy.
                To see a list of available paths, use str_query().
                No path given means print the whole tree from the root.

        Returns:
            text representation of the dynamically decorated tree
        """
        return self._get_dynamically_decorated_tree_text_representation(
            path,
            write_numpifiable=True,
        )

    def get_typesystem_node(self, path: str | None = None) -> ConcreteNodeType | None:
        """Tries to retrieve the original typesystem node for the given path

        Applies to the dynamically decorated tree obtained from the indexed datablock.

        Args:
            path: Slash-separated path pointing to a node in the typesystem hierarchy.
                To see a list of available paths, use str_query().

        Returns:
            typesystem node if found.
        """
        return self.create_query(path).get_typesystem_node()

    def get_length(self, path: str, *coordinates: int) -> int | None:
        """Gives the max index of the list of dynamic offsets, and navigate through coordinates if needed.

        Args:
            path: Slash-separated path pointing to a node in the typesystem hierarchy.
                To see a list of available paths, use str_query().
            coordinates: Optional extra coordinates to uniquely identify a node
                (some type hierarchies have a high-dimensionality,
                usually due to the nested use of Array Variables in the schema description)

        Returns:
            max index if possible, else None (eg, if the dimension of the node is 0)
        """
        offset = self.get_offset(path, *coordinates)
        return len(offset) if offset else None

    def get_dynamic_size(self, path: str, *coordinates: int) -> int | None:
        return self.create_query(path, *coordinates).get_dynamic_size()

    def get_dimensionality(self, path: str) -> int | None:
        """Get the dimensionality of the node. It is the count of integer indexes to provide when using `query`.

        Args:
            path: Slash-separated path pointing to a node in the typesystem hierarchy.
                To see a list of available paths, use str_query().

        Returns:
            the dimensionality
        """
        return self.create_query(path).get_dimensionality()

    def create_query(self, path: str | None) -> IndexedDataBlockQuery:
        """Factory method to instantiate queries over the indexed datablock.

        Unless needed, use the `query` method directly.

        Args:
            path: Slash-separated path pointing to a node in the typesystem hierarchy.
                To see a list of available paths, use str_query().

        Returns:
            IndexedDataBlockQuery object
        """
        indexed_datablock = self._require_indexed_datablock()

        return create_query(indexed_datablock)[path]

    def _get_statically_decorated_tree_text_representation(self, /, **kwargs):
        return output_tree_to_txt(
            self.statically_decorated_tree,
            write_dynamic_attributes=False,
            write_path=False,
            **kwargs,
        )

    def _get_dynamically_decorated_tree_text_representation(self, path, /, **kwargs):
        self._require_indexed_datablock()
        node = self.get_typesystem_node(path)

        return output_tree_to_txt(
            node,
            write_dynamic_attributes=True,
            write_path=True,
            ignore_static_nodes=not node.static,
            **kwargs,
        )

    def _index_datablock(self):
        self.indexed_datablock = index_datablock(
            self.xsd_path, self.xml_schema_path, self.datablock_folder_path
        )

    def get_offset(self, path: str, *coordinates: int) -> list[int] | None:
        return self.create_query(path).get_dynamic_offset(*coordinates)

    def _require_indexed_datablock(self):
        if not self.indexed_datablock:
            raise SmosWalkerException(
                "No indexed_datablock available! Please set it before using this method via `set_datablock`"
            )
        return self.indexed_datablock

datablock_info property

Information about the indexed datablock

type_info property

Information about a path in the typesystem hierarchy

__init__(xsd_path, xml_schema_path, datablock_folder_path=None)

Constructor

Parameters:

Name	Type	Description	Default
xsd_path	str	Path pointing towards the XSD file, also known as "binx.xsd".	required
xml_schema_path	str	Path pointing towards an XML schema file.	required
datablock_folder_path	str | None	Path pointing towards the folder containing the data block.	None

Source code in smos_walker\api\facade.py

def __init__(
    self,
    xsd_path: str,
    xml_schema_path: str,
    datablock_folder_path: str | None = None,
):
    """Constructor

    Args:
        xsd_path: Path pointing towards the XSD file, also known as "binx.xsd".
        xml_schema_path: Path pointing towards an XML schema file.
        datablock_folder_path: Path pointing towards the folder containing the data block.
    """
    self.xsd_path: str = xsd_path  # Store here the xsd content
    self.xml_schema_path: str = xml_schema_path  # Store here the binXschema content

    # Keep a plain statically decorated tree, as a reference
    # It is not reused when indexing datablock as dynamic decoration occurs in-place.
    # For now, the static analysis is just remade when needed.
    # It can be printed for information with `get_dynamically_decorated_tree_text_representation`
    self.statically_decorated_tree: BaseNode = load_statically_decorated_tree(
        xsd_path, xml_schema_path
    )

    # Annotate nodes with paths
    self.statically_decorated_tree.accept(DictIndexNodeVisitor())

    self.datablock_folder_path: str | None = None
    self.indexed_datablock: IndexedDataBlock | None = None

    # If the datablock is already provided, immediately index it
    if datablock_folder_path:
        self.set_datablock(datablock_folder_path)

create_query(path)

Factory method to instantiate queries over the indexed datablock.

Unless needed, use the query method directly.

Parameters:

Name	Type	Description	Default
path	str | None	Slash-separated path pointing to a node in the typesystem hierarchy. To see a list of available paths, use str_query().	required

Returns:

Type	Description
IndexedDataBlockQuery	IndexedDataBlockQuery object

Source code in smos_walker\api\facade.py

def create_query(self, path: str | None) -> IndexedDataBlockQuery:
    """Factory method to instantiate queries over the indexed datablock.

    Unless needed, use the `query` method directly.

    Args:
        path: Slash-separated path pointing to a node in the typesystem hierarchy.
            To see a list of available paths, use str_query().

    Returns:
        IndexedDataBlockQuery object
    """
    indexed_datablock = self._require_indexed_datablock()

    return create_query(indexed_datablock)[path]

get_dimensionality(path)

Get the dimensionality of the node. It is the count of integer indexes to provide when using query.

Parameters:

Name	Type	Description	Default
path	str	Slash-separated path pointing to a node in the typesystem hierarchy. To see a list of available paths, use str_query().	required

Returns:

Type	Description
int | None	the dimensionality

Source code in smos_walker\api\facade.py

def get_dimensionality(self, path: str) -> int | None:
    """Get the dimensionality of the node. It is the count of integer indexes to provide when using `query`.

    Args:
        path: Slash-separated path pointing to a node in the typesystem hierarchy.
            To see a list of available paths, use str_query().

    Returns:
        the dimensionality
    """
    return self.create_query(path).get_dimensionality()

get_info(path=None)

alias forstr_dynamic_tree

Source code in smos_walker\api\facade.py

def get_info(self, path: str | None = None) -> str:
    """alias for`str_dynamic_tree`"""
    return self.str_dynamic_tree(path)

get_length(path, *coordinates)

Gives the max index of the list of dynamic offsets, and navigate through coordinates if needed.

Parameters:

Name	Type	Description	Default
path	str	Slash-separated path pointing to a node in the typesystem hierarchy. To see a list of available paths, use str_query().	required
coordinates	int	Optional extra coordinates to uniquely identify a node (some type hierarchies have a high-dimensionality, usually due to the nested use of Array Variables in the schema description)	()

Returns:

Type	Description
int | None	max index if possible, else None (eg, if the dimension of the node is 0)

Source code in smos_walker\api\facade.py

def get_length(self, path: str, *coordinates: int) -> int | None:
    """Gives the max index of the list of dynamic offsets, and navigate through coordinates if needed.

    Args:
        path: Slash-separated path pointing to a node in the typesystem hierarchy.
            To see a list of available paths, use str_query().
        coordinates: Optional extra coordinates to uniquely identify a node
            (some type hierarchies have a high-dimensionality,
            usually due to the nested use of Array Variables in the schema description)

    Returns:
        max index if possible, else None (eg, if the dimension of the node is 0)
    """
    offset = self.get_offset(path, *coordinates)
    return len(offset) if offset else None

get_type_info(path=None)

Information about a path in the typesystem hierarchy

Source code in smos_walker\api\facade.py

def get_type_info(self, path: str | None = None):
    """Information about a path in the typesystem hierarchy"""
    return self.str_query(path)

get_typesystem_node(path=None)

Tries to retrieve the original typesystem node for the given path

Applies to the dynamically decorated tree obtained from the indexed datablock.

Parameters:

Name	Type	Description	Default
path	str | None	Slash-separated path pointing to a node in the typesystem hierarchy. To see a list of available paths, use str_query().	None

Returns:

Type	Description
ConcreteNodeType | None	typesystem node if found.

Source code in smos_walker\api\facade.py

def get_typesystem_node(self, path: str | None = None) -> ConcreteNodeType | None:
    """Tries to retrieve the original typesystem node for the given path

    Applies to the dynamically decorated tree obtained from the indexed datablock.

    Args:
        path: Slash-separated path pointing to a node in the typesystem hierarchy.
            To see a list of available paths, use str_query().

    Returns:
        typesystem node if found.
    """
    return self.create_query(path).get_typesystem_node()

query(path, *coordinates)

Fetch data from the index datablock

Parameters:

Name	Type	Description	Default
path	str	Slash-separated path pointing to a node in the typesystem hierarchy. To see a list of available paths, use str_query(path).	required
coordinates	int	Optional extra coordinates to uniquely identify a node (some type hierarchies have a high-dimensionality, usually due to the nested use of Array Variables in the schema description)	()

Raises:

Type	Description
SmosWalkerException	error message containing help to uniquely identify a node in the typesystem hierarchy.

Returns:

Type	Description
NumpifiedDataBlockType | None	A Numpy's ndarray if the node could be located in the typesystem and the datablock.

Source code in smos_walker\api\facade.py

def query(self, path: str, *coordinates: int) -> NumpifiedDataBlockType | None:
    """Fetch data from the index datablock

    Args:
        path: Slash-separated path pointing to a node in the typesystem hierarchy.
            To see a list of available paths, use str_query(path).
        coordinates: Optional extra coordinates to uniquely identify a node
            (some type hierarchies have a high-dimensionality, usually due to the nested
            use of Array Variables in the schema description)
    Raises:
        SmosWalkerException: error message containing help to uniquely identify a node in the typesystem hierarchy.

    Returns:
        A Numpy's ndarray if the node could be located in the typesystem and the datablock.
    """
    self._require_indexed_datablock()

    # Allow user to quickly give extra dynamic coordinates after failure to query the datablock.
    return self.create_query(path).consume_query(*coordinates)

set_datablock(datablock_folder_path)

Sets the datablock path and index it

Parameters:

Name	Type	Description	Default
datablock_folder_path	str	Path pointing towards the folder containing the data block.	required

Source code in smos_walker\api\facade.py

def set_datablock(self, datablock_folder_path: str):
    """Sets the datablock path and index it

    Args:
        datablock_folder_path: Path pointing towards the folder containing the data block.
    """
    self.datablock_folder_path = datablock_folder_path
    self._index_datablock()

str_dynamic_tree(path=None)

Print a user-friendly representation of the dynamically decorated tree (aka Step 3 in the XML schema parsing)

Requires an indexed datablock. Use set_datablock if no datablock is set yet.

Parameters:

Name	Type	Description	Default
path	str | None	Slash-separated path pointing to a node in the typesystem hierarchy. To see a list of available paths, use str_query(). No path given means print the whole tree from the root.	None

Returns:

Type	Description
str	text representation of the dynamically decorated tree

Source code in smos_walker\api\facade.py

def str_dynamic_tree(self, path: str | None = None) -> str:
    """Print a user-friendly representation of the dynamically decorated tree (aka Step 3 in the XML schema parsing)

    Requires an indexed datablock. Use `set_datablock` if no datablock is set yet.

    Args:
        path: Slash-separated path pointing to a node in the typesystem hierarchy.
            To see a list of available paths, use str_query().
            No path given means print the whole tree from the root.

    Returns:
        text representation of the dynamically decorated tree
    """
    return self._get_dynamically_decorated_tree_text_representation(
        path,
        write_numpifiable=True,
    )

str_static_tree()

Print a user-friendly representation of the statically decorated tree (aka Step 2 in the XML schema parsing)

Returns:

Type	Description
str	text representation of the statically decorated tree

Source code in smos_walker\api\facade.py

def str_static_tree(self) -> str:
    """Print a user-friendly representation of the statically decorated tree (aka Step 2 in the XML schema parsing)

    Returns:
        text representation of the statically decorated tree
    """
    return self._get_statically_decorated_tree_text_representation()

create_query(indexed_datablock)

Creates a query object for a given data block.

Parameters:

Name	Type	Description	Default
indexed_datablock	IndexedDataBlock	The indexed data block.	required

Returns:

Type	Description
IndexedDataBlockQuery	The query object.

Source code in smos_walker\api\facade.py

def create_query(indexed_datablock: IndexedDataBlock) -> IndexedDataBlockQuery:
    """Creates a query object for a given data block.

    Args:
        indexed_datablock: The indexed data block.

    Returns:
        The query object.
    """
    return IndexedDataBlockQuery(indexed_datablock)

index_datablock(xsd_path, xml_schema_path, datablock_folder_path)

Computes an index for a given datablock, with the help of its schema definition.

Parameters:

Name	Type	Description	Default
xsd_path	str	Path pointing towards the XSD file, also known as "binx.xsd".	required
xml_schema_path	str	Path pointing towards an XML schema file.	required
datablock_folder_path	str	Path pointing towards the folder containing the data block.	required

Returns:

Type	Description
IndexedDataBlock	The resulting datablock-specific index.

Source code in smos_walker\api\facade.py

def index_datablock(
    xsd_path: str, xml_schema_path: str, datablock_folder_path: str
) -> IndexedDataBlock:
    """Computes an index for a given datablock, with the help of its schema definition.

    Args:
        xsd_path: Path pointing towards the XSD file, also known as "binx.xsd".
        xml_schema_path: Path pointing towards an XML schema file.
        datablock_folder_path: Path pointing towards the folder containing the data block.

    Returns:
        The resulting datablock-specific index.
    """
    decorated_tree: BaseNode = load_statically_decorated_tree(xsd_path, xml_schema_path)

    datablock: DataBlockType | None = read_dbl(datablock_folder_path)

    if datablock is None:
        raise SmosWalkerException("datablock is None")

    dynamically_decorate_tree(decorated_tree, datablock)

    return IndexedDataBlock(decorated_tree, datablock)

smos_walker.api.tools

Useful high-level tools that are too specific to be part of the API's facade

ProblematicSchema

Bases: TypedDict

Description of schema that cannot be successfully statically decorated

Source code in smos_walker\api\tools.py

class ProblematicSchema(TypedDict):
    """Description of schema that cannot be successfully statically decorated"""

    path: str
    exception: str

generate_human_readable_statically_decorated_schemas(schemas_base_path)

Generates human-readable text files of statically decorated XML data schemas.

Parameters:

Name	Type	Description	Default
schemas_base_path	Path	Home path containing all schemas as well as the XSD file.	required

Returns:

Name	Type	Description
list	ConsumerResultsType	All files that did pose a problem during text generation (empty if full success)

Source code in smos_walker\api\tools.py

def generate_human_readable_statically_decorated_schemas(
    schemas_base_path: Path,
) -> ConsumerResultsType:
    """Generates human-readable text files of statically decorated XML data schemas.

    Args:
        schemas_base_path: Home path containing all schemas as well as the XSD file.

    Returns:
        list: All files that did pose a problem during text generation (empty if full success)
    """
    logging.info("generate_human_readable_statically_decorated_schemas")

    return provide_statically_decorated_trees(
        schemas_base_path, [dump_decoration_tree_consumer]
    )

output_tree_to_json(tree)

Outputs a tree to JSON

Parameters:

Name	Type	Description	Default
tree	BaseNode	Statically- or dynamically-decorated type system tree	required

Returns:

Type	Description
str	JSON-representation of the tree

Source code in smos_walker\api\tools.py

def output_tree_to_json(tree: BaseNode) -> str:
    """Outputs a tree to JSON

    Args:
        tree: Statically- or dynamically-decorated type system tree

    Returns:
        JSON-representation of the tree
    """
    dict_export_visitor = DictExportNodeVisitor()
    dictified_tree = tree.accept(dict_export_visitor)
    return json.dumps(dictified_tree, indent=2, sort_keys=True)

output_tree_to_txt(tree, **kwargs)

Outputs a tree to TXT

Parameters:

Name	Type	Description	Default
tree	BaseNode	Statically- or dynamically-decorated type system tree	required

Returns:

Type	Description
str	TXT-representation of the tree

Source code in smos_walker\api\tools.py

def output_tree_to_txt(tree: BaseNode, **kwargs) -> str:
    """Outputs a tree to TXT

    Args:
        tree: Statically- or dynamically-decorated type system tree

    Returns:
        TXT-representation of the tree
    """
    human_readable_visitor = HumanReadableNodeVisitor(**kwargs)
    tree.accept(human_readable_visitor)
    return str(human_readable_visitor)

provide_statically_decorated_trees(schemas_base_path, consumers, *, verbose=True)

Provide statically decorated trees, by iterating over all binXschema files contained in the given folder.

Parameters:

Name	Type	Description	Default
schemas_base_path	Path	Home path containing all schemas as well as the XSD file.	required
consumers	list[DecoratedTreeConsumer]	List of decorated tree consumers. The current path is also provided.	required
verbose	bool	Log progress	True

Returns:

Type	Description
ConsumerResultsType	list[ProblematicSchema]: description

Source code in smos_walker\api\tools.py

def provide_statically_decorated_trees(
    schemas_base_path: Path,
    consumers: list[DecoratedTreeConsumer],
    *,
    verbose: bool = True,
) -> ConsumerResultsType:
    """Provide statically decorated trees, by iterating over all binXschema files contained in the given folder.

    Args:
        schemas_base_path:  Home path containing all schemas as well as the XSD file.
        consumers: List of decorated tree consumers. The current path is also provided.
        verbose: Log progress

    Returns:
        list[ProblematicSchema]: _description_
    """

    if verbose:
        logging.info("provide_statically_decorated_trees")

    problematic_schemas: list[ProblematicSchema] = []

    results: dict[str, Any] = {consumer.__name__: {} for consumer in consumers}

    index = 0
    for index, schema_path in enumerate(schemas_base_path.glob("**/*.binXschema.xml")):

        try:
            decorated_tree: BaseNode = load_statically_decorated_tree(
                str(schemas_base_path / "binx/binx.xsd"), str(schema_path)
            )

            if verbose:
                logging.info(f"✅ [{index:04d}] {schema_path.name=}")

            for consumer in consumers:
                result = consumer(decorated_tree, schema_path)
                results[consumer.__name__][schema_path.stem] = result

        except SmosWalkerException as exception:
            if verbose:
                logging.info(
                    f"🟡 [{index:04d}] {schema_path.name=}. Exception caught: {exception}"
                )
            problematic_schemas.append(
                {"path": str(schema_path.stem), "exception": str(exception)}
            )

    if verbose:
        logging.info(
            (
                f"{index} schemas parsed. "
                f"Exceptions encountered with {len(problematic_schemas)} files: {problematic_schemas=}"
            )
        )

    return results, problematic_schemas

smos_walker.cli

Command-line interface for smos_walker

The module's facade does expose an entrypoint.

smos_walker.cli.facade

entrypoint(xsd_path, xml_schema_path, *, datablock_folder_path=None, output_format='json', step=1)

summary

Parameters:

Name	Type	Description	Default
xsd_path	str	Path pointing towards the XSD file, also known as "binx.xsd".	required
xml_schema_path	str	Path pointing towards an XML schema file.	required
datablock_folder_path	str | None	Required by step 3 (dynamic indexing). Defaults to None.	None
output_format	Literal['txt', 'json']	Output format.	'json'
step	Literal[1, 2, 3]	Step	1

Returns:

Type	Description
str	String-representation of the working tree

Source code in smos_walker\cli\facade.py

def entrypoint(
    xsd_path: str,
    xml_schema_path: str,
    *,
    datablock_folder_path: str | None = None,
    output_format: Literal["txt", "json"] = "json",
    step: Literal[1, 2, 3] = 1
) -> str:
    """_summary_

    Args:
        xsd_path: Path pointing towards the XSD file, also known as "binx.xsd".
        xml_schema_path: Path pointing towards an XML schema file.
        datablock_folder_path: Required by step 3 (dynamic indexing). Defaults to None.
        output_format: Output format.
        step: Step

    Returns:
        String-representation of the working tree
    """
    parsed_typesystem_tree = render_typesystem_parsed_tree_from_xml(
        xsd_path, xml_schema_path
    )

    if step == 1:
        converter = json_dumps  # This is not a BaseNode, so Visitors cannot be applied
        return converter(parsed_typesystem_tree)

    decorated_tree = get_statically_decorated_tree(parsed_typesystem_tree)

    if step == 2:
        converter = CLI_OUTPUT_FORMATS[output_format]
        return converter(decorated_tree)

    if datablock_folder_path is None:
        logging.warning(
            "The binary datablock is required to dynamically decorate the type tree."
        )
        return None

    dynamically_decorate_tree(decorated_tree, datablock_folder_path)

    if step == 3:
        converter = CLI_OUTPUT_FORMATS[output_format]
        return converter(decorated_tree)

    logging.warning("No step after step 3.")
    return None

smos_walker.core

Contains core smos_walker's logic

smos_walker.core.node

ArrayFixedNode

Bases: BaseNode

Array Fixed Node: the nicest array node.

It is analoguous to static arrays in programming languages such as C.

It satisfies the following properties:

• It has one or more dimensions. Their naming is optional.
• It is static when its unique child is static.
• When static, it can be mapped to a Numpy's dtype
• It has one child, that can be a leaf or a struct

Property known to be wrong - If the child is a struct, it has to be static. => It can contain other arrayvariables

To prove these, the PredicateNodeVisitor could be used.

Attributes:

Name	Type	Description
dimensions	list[NodeDimension]	The name of the type containing the structure.
cardinality	int	Total number of children accross all dimensions.
child	LeafNode | StructNode	The type of data contained in the array.

Source code in smos_walker\core\node.py

class ArrayFixedNode(BaseNode):
    """Array Fixed Node: the nicest array node.

    It is analoguous to static arrays in programming languages such as C.

    It satisfies the following properties:

    - It has one or more dimensions. Their naming is optional.
    - It is static when its unique child is static.
    - When static, it can be mapped to a Numpy's `dtype`
    - It has one child, that can be a leaf or a struct

    Property known to be wrong
    - If the child is a struct, it has to be static. => It can contain other arrayvariables

    To prove these, the `PredicateNodeVisitor` could be used.

    Attributes:
        dimensions: The name of the type containing the structure.
        cardinality: Total number of children accross all dimensions.
        child: The type of data contained in the array.
    """

    def __init__(
        self,
        name: str,
        dimensions: list[NodeDimension],  # multiple dimensions allowed [label, dim]
        child: LeafNode | StructNode,
    ) -> None:
        super().__init__(name)
        self.dimensions: list[NodeDimension] = dimensions
        self.cardinality: int = math.prod(
            dimension["size"] for dimension in self.dimensions
        )
        self.child: LeafNode | StructNode = child

    @cached_property
    def block_byte_size(self):
        if self.child.static:
            return self.cardinality * (self.child.block_byte_size)
        return None

    @property
    def quasi_static(self):
        if not self.child:
            return None  # information not available yet
        return self.child.static

    def to_dict(self):
        return merge_dicts(
            shallow_render_base_node(self), shallow_render_array_fixed_node(self)
        )

    def accept(self, visitor: NodeVisitor):
        return visitor.visit_array_fixed(self)

ArrayVariableNode

Bases: BaseNode

Array Variable Node: the one that did create the need for this project in the first place.

It is analoguous to dynamically allocated arrays in programming languages such as C.

It satisfies the following properties:

• It has only one variable dimension
• It is never static, as its total size cannot be determined without an instance of a datablock providing its length
• When quasi-static, it can be mapped to a Numpy's dtype
• The only dimension is always named
• Children are always Struct (uncertain, as ArrayFixed node can reference LeafNodes directly)
• size_ref is always a Leaf (very likely, as it stores a scalar describing the dimension)

Properties to be proved formally:

• The most nested ArrayVariable in a schema hierarchy is quasi-static.

Warning

Currently, the child attribute can only be a StructNode, but this is likely to be a StructNode | LeafNode, though not encountered yet.

To prove these, the PredicateNodeVisitor could be used.

Attributes:

Name	Type	Description
size_ref	LeafNode	A leaf node describing the array length. It can only be determined as runtime. The size_ref is the header of an array variable. Related to the dimensions attribute of an ArrayFixed.
dynamic_size	RecursiveListOfIntegers | None	Like BaseNode's dynamic_coverage and dynamic_offset, this attribute can be multi-dimensional and can only be read at runtime. For example, for an arrayvariable 2 in an arrayvariable 1, multiple arrayvariables 2 will be present at runtime and this variable will contain a list of integers.
child	StructNode	The type of data contained in the array.

Source code in smos_walker\core\node.py

class ArrayVariableNode(BaseNode):
    """Array Variable Node: the one that did create the need for this project in the first place.

    It is analoguous to dynamically allocated arrays in programming languages such as C.

    It satisfies the following properties:

    - It has only one variable dimension
    - It is never static, as its total size cannot be determined without an instance of a datablock providing its length
    - When quasi-static, it can be mapped to a Numpy's `dtype`
    - The only dimension is always named
    - Children are always Struct (uncertain, as ArrayFixed node can reference LeafNodes directly)
    - `size_ref` is always a Leaf (very likely, as it stores a scalar describing the dimension)

    Properties to be proved formally:

    - The most nested ArrayVariable in a schema hierarchy is quasi-static.

    Warning:
        Currently, the `child` attribute can only be a `StructNode`,
        but this is likely to be a `StructNode | LeafNode`, though not encountered yet.

    To prove these, the `PredicateNodeVisitor` could be used.

    Attributes:
        size_ref: A leaf node describing the array length. It can only be determined as runtime.
            The size_ref is the _header_ of an array variable. Related to the `dimensions` attribute of an `ArrayFixed`.
        dynamic_size: Like `BaseNode`'s `dynamic_coverage` and `dynamic_offset`,
            this attribute can be multi-dimensional and can only be read at runtime.
                For example, for an arrayvariable 2 in an arrayvariable 1,
                multiple arrayvariables 2 will be present at runtime and this variable will contain a list of integers.
        child: The type of data contained in the array.
    """

    def __init__(
        self,
        name: str,
        child: StructNode,
        size_ref: LeafNode,
    ) -> None:
        super().__init__(name)
        self.size_ref: LeafNode = size_ref
        self.child: StructNode = child

        # Actual length, read dynamically
        self.dynamic_size: RecursiveListOfIntegers | None = None

    @property
    def quasi_static(self):
        if not self.child:
            return None  # information not available yet
        return self.child.static

    def to_dict(self):
        return merge_dicts(
            shallow_render_base_node(self),
            shallow_render_array_variable_node(self),
        )

    def accept(self, visitor: NodeVisitor):
        return visitor.visit_array_variable(self)

BaseNode

Base Node: The abstract base node that all others inherit from.

A convention followed in this class hierarchy is to prefix by dynamic attributes that can only determined with a concrete instance of a datablock. It means that they cannot be determined solely from the type system XML schema.

Attributes:

Name	Type	Description
name	str	Mandatory node's name (static)
label	str | None	Optional label over the node (static)
block_byte_size	int | None	Computed at compile time, if possible (static)
dynamic_dimensionality	int | None	Dimensionality: number of integers required to uniquely identify the node
dynamic_coverage	int | None	Accumulated size obtained during each of the node's visit when reading the binary data block
dynamic_offset	RecursiveListOfIntegers | None	Dynamic offset while reading. The array is of dimension dynamic_dimensionality

Source code in smos_walker\core\node.py

class BaseNode:
    """Base Node: The abstract base node that all others inherit from.

    A convention followed in this class hierarchy is to prefix by `dynamic` attributes that can only
    determined with a concrete instance of a datablock.
    It means that they cannot be determined solely from the type system XML schema.

    Attributes:
        name: Mandatory node's name (static)
        label: Optional label over the node (static)
        block_byte_size: Computed at compile time, if possible (static)

        dynamic_dimensionality: Dimensionality: number of integers required to uniquely identify the node
        dynamic_coverage: Accumulated size obtained during each of the node's visit when reading the binary data block
        dynamic_offset: Dynamic offset while reading. The array is of dimension `dynamic_dimensionality`
    """

    def __init__(self, name: str, label: str | None = None) -> None:
        """
        Args:
            name: Mandatory node's name
            label: Optional label over the node
        """

        # Static attributes
        # -----------------
        self.name: str = name
        self.label: str | None = label

        # Dynamic attributes
        # ------------------
        self.dynamic_dimensionality: int | None = None
        self.dynamic_coverage: int | None = None
        self.dynamic_offset: RecursiveListOfIntegers | None = None

        # Optional attributes
        # -------------------
        self.path: str | None = None

    def __repr__(self):
        return json.dumps(self.to_dict())

    def __str__(self):
        return self.__repr__()

    @cached_property
    def static(self) -> bool:
        """True if the node's children block size is known in advance (static)

        To be overriden by subclasses
        """
        return self.block_byte_size is not None

    @cached_property
    def block_byte_size(self) -> int | None:
        return None

    @property
    def numpifiable(self) -> bool:
        return (
            self.static
            # pylint: disable=no-member
            or (isinstance(self, ArrayVariableNode) and self.quasi_static)
            # If children are numpifiable, the node is numpifiable
            # or isinstance(self, StructNode)
            # and all(child.numpifiable for child in self.children)
        )

    def to_dict(self):
        return shallow_render_base_node(self)

    def accept(self, visitor: NodeVisitor):
        pass  # To be implemented by subclasses

static: bool property cached

True if the node's children block size is known in advance (static)

To be overriden by subclasses

__init__(name, label=None)

Parameters:

Name	Type	Description	Default
name	str	Mandatory node's name	required
label	str | None	Optional label over the node	None

Source code in smos_walker\core\node.py

def __init__(self, name: str, label: str | None = None) -> None:
    """
    Args:
        name: Mandatory node's name
        label: Optional label over the node
    """

    # Static attributes
    # -----------------
    self.name: str = name
    self.label: str | None = label

    # Dynamic attributes
    # ------------------
    self.dynamic_dimensionality: int | None = None
    self.dynamic_coverage: int | None = None
    self.dynamic_offset: RecursiveListOfIntegers | None = None

    # Optional attributes
    # -------------------
    self.path: str | None = None

LeafNode

Bases: BaseNode

Leaf Node: the final concrete data described by the type system.

It is analoguous to primitive types in programming languages.

It satisfies the following properties:

• it is always static
• it has no children (as its name indicated, like a leaf on a tree)

Attributes:

Name	Type	Description
primitive_type	str	A string representing the primitive of the node. No type checking is performed ; an unexpected value will engender a runtime error.

Source code in smos_walker\core\node.py

class LeafNode(BaseNode):
    """Leaf Node: the final concrete data described by the type system.

    It is analoguous to primitive types in programming languages.

    It satisfies the following properties:

    - it is always `static`
    - it has no children (as its name indicated, like a leaf on a tree)

    Attributes:
        primitive_type: A string representing the primitive of the node.
            No type checking is performed ; an unexpected value will engender a runtime error.
    """

    def __init__(
        self,
        name: str,
        primitive_type: str,
    ) -> None:
        super().__init__(name)
        self.primitive_type: str = primitive_type

    @cached_property
    def block_byte_size(self):
        return PrimitiveTypeTagNameMapping[self.primitive_type]["byte_count"]

    def to_dict(self):
        return merge_dicts(
            shallow_render_base_node(self),
            {
                "_class": self.__class__.__name__,
                "primitive_type": self.primitive_type,
            },
        )

    def accept(self, visitor: NodeVisitor):
        return visitor.visit_leaf(self)

NodeDimension

Bases: TypedDict

A Node's dimension is composed of an optional name and a mandatory size.

Source code in smos_walker\core\node.py

class NodeDimension(TypedDict):
    """A Node's dimension is composed of an optional name and a mandatory size."""

    name: str | None
    size: int

NodeVisitor

Bases: Visitor

Generic interface for the visitor pattern, used for tree exploration

Source code in smos_walker\core\node.py

class NodeVisitor(Visitor):  # Interface
    """Generic interface for the visitor pattern, used for tree exploration"""

    def visit_leaf(self, node: "LeafNode"):
        pass

    def visit_struct(self, node: "StructNode"):
        pass

    def visit_array_fixed(self, node: "ArrayFixedNode"):
        pass

    def visit_array_variable(self, node: "ArrayVariableNode"):
        pass

StructNode

Bases: BaseNode

Struct Node: the abstract data structure containing the others.

It is analoguous to structures in programming languages such as C.

More specifically, structures are closely related to the type system. Indeed, the "structure" and "type" concepts are merged into one in this parser. In the original XML schema files, useType tags often have an only struct tag child. However, it can also contains one ArrayFixed or one ArrayVariable. If that is the case, the array node child is wrapped into a list containing itself, to simulate a structure node having it as a child.

It satisfies the following properties:

• It has a list of children. They can be of all type deriving BaseNode.
• It is static when all its children are static
• When static, it can be mapped to a Numpy's dtype

Attributes:

Name	Type	Description
type_name	str	The name of the type containing the structure.
children	list[BaseNode]	Members of the structure.

Source code in smos_walker\core\node.py

class StructNode(BaseNode):
    """Struct Node: the abstract data structure containing the others.

    It is analoguous to structures in programming languages such as C.

    More specifically, structures are closely related to the type system.
    Indeed, the "structure" and "type" concepts are merged into one in this parser.
    In the original XML schema files, `useType` tags often have an only `struct` tag child.
    However, it can also contains one  `ArrayFixed` or one `ArrayVariable`.
    If that is the case, the array node child is wrapped into a list containing itself,
    to simulate a structure node having it as a child.

    It satisfies the following properties:

    - It has a list of children. They can be of all type deriving `BaseNode`.
    - It is static when all its children are static
    - When static, it can be mapped to a Numpy's `dtype`

    Attributes:
        type_name: The name of the type containing the structure.
        children: Members of the structure.
    """

    def __init__(
        self,
        name: str,
        type_name: str,
        children: list[BaseNode],
    ) -> None:
        super().__init__(name)
        self.type_name: str = type_name
        self.children: list[BaseNode] = children

        if isinstance(self.children, (ArrayFixedNode, ArrayVariableNode)):
            # During initial conception, it has been - wrongly - assumed that struct tags
            # always were a direct child of useType tags, hence the merging between the
            # two concepts. A StructNode is actually more a "UseTypeNode".
            # Since Arrays can also be children of useType tags, the fix was to wrap them
            # into a struct node.
            self.children = [self.children]

    @cached_property
    def block_byte_size(self):
        if all(child.static for child in self.children):
            return sum(child.block_byte_size for child in self.children)
        return None

    def to_dict(self):
        return merge_dicts(
            shallow_render_base_node(self), shallow_render_struct_node(self)
        )

    def accept(self, visitor: NodeVisitor):
        return visitor.visit_struct(self)

    def yield_children(self):
        for child in self.children:
            yield child

smos_walker.core.visitors

Catalog of Visitors that do visit the decorated typesystem tree.

AbstractPredicateNodeVisitor

Bases: NodeVisitor

Base predicate node visitor. Needs to be extended.

Source code in smos_walker\core\visitors.py

class AbstractPredicateNodeVisitor(NodeVisitor):
    """Base predicate node visitor. Needs to be extended."""

    def __init__(self, predicate: Callable[[BaseNode], bool]):
        self.predicate = predicate

    def visit_leaf(self, node: LeafNode):
        return self.predicate(node)

DictExportNodeVisitor

Bases: NodeVisitor

Export all known information about a node (both static and dynamic)

Source code in smos_walker\core\visitors.py

class DictExportNodeVisitor(NodeVisitor):
    """Export all known information about a node (both static and dynamic)"""

    def visit_leaf(self, node: LeafNode):
        return merge_dicts(render_base_node_dynamic_offset(node), node.to_dict())

    def visit_struct(self, node: StructNode):
        return merge_dicts(
            render_base_node_dynamic_offset(node),
            node.to_dict(),
            {
                "~children": [child.accept(self) for child in node.children],
            },
            # {
            #     "~children": [child.to_dict() for child in node.children],
            # },
        )

    def visit_array_fixed(self, node: ArrayFixedNode):
        return merge_dicts(
            render_base_node_dynamic_offset(node),
            node.to_dict(),
            {
                "~child": node.child.accept(self),
            },
            # {
            #     "~child": node.child.to_dict(),
            # },
        )

    def visit_array_variable(self, node: ArrayVariableNode):
        return merge_dicts(
            render_base_node_dynamic_offset(node),
            {
                "dynamic_size": node.dynamic_size
            },  # Like dynamic_offset, this list can be very long.
            node.to_dict(),
            {
                "~child": node.child.accept(self),
            },
            # {
            #     "~child": node.child.to_dict(),
            # },
        )

DictIndexNodeVisitor

Bases: NodeVisitor

Index all nodes by their name for easy index, and fill the path member of nodes IN PLACE

Note

Modifies IN PLACE the hierarchy.

Warning

• names should be unique for a given parent node.
• the last visited node will have precedence over the other synonymous nodes in case of conflicts.

Source code in smos_walker\core\visitors.py

class DictIndexNodeVisitor(NodeVisitor):
    """Index all nodes by their name for easy index, and fill the `path` member of nodes IN PLACE

    Note:
        Modifies IN PLACE the hierarchy.

    Warning:
        - names should be unique for a given parent node.
        - the last visited node will have precedence over the other synonymous nodes in case of conflicts.
    """

    def __init__(self):
        self.path: list[str] = []
        self.mapping: dict[PathType, ConcreteNodeType] = {}

    def add_to_mapping(self, node: ConcreteNodeType):
        tuple_path = tuple(self.path)
        self.mapping[tuple_path] = node
        node.path = tuple_path

    def visit_leaf(self, node: LeafNode):

        self.path.append(node.name)
        self.add_to_mapping(node)
        self.path.pop()

    def visit_struct(self, node: StructNode):

        if node.name:
            self.path.append(
                node.name if node.name else SmosWalkerConfig.ANONYMOUS_STRUCT_IDENTIFIER
            )
            self.add_to_mapping(node)
            for child in node.children:
                child.accept(self)
            self.path.pop()
        else:
            for child in node.children:
                child.accept(self)

    def visit_array_fixed(self, node: ArrayFixedNode):

        self.path.append(node.name)

        self.add_to_mapping(node)

        node.child.accept(self)
        self.path.pop()

    def visit_array_variable(self, node: ArrayVariableNode):

        self.path.append(node.name)
        self.add_to_mapping(node)

        node.child.accept(self)
        self.path.pop()

ExistentialPredicateNodeVisitor

Bases: AbstractPredicateNodeVisitor

Verify existence of a node in the tree that do match a particular condition

Source code in smos_walker\core\visitors.py

class ExistentialPredicateNodeVisitor(AbstractPredicateNodeVisitor):
    """Verify existence of a node in the tree that do match a particular condition"""

    def visit_struct(self, node: StructNode):
        return self.predicate(node) or any(
            child.accept(self) for child in node.children
        )

    def visit_array_fixed(self, node: ArrayFixedNode):
        return self.predicate(node) or node.child.accept(self)

    def visit_array_variable(self, node: ArrayVariableNode):
        return self.predicate(node) or node.child.accept(self)

HumanReadableNodeVisitor

Bases: NodeVisitor

Renders a human-readable text-representation of a node.

Various flags to control displayed data

Source code in smos_walker\core\visitors.py

class HumanReadableNodeVisitor(NodeVisitor):
    """Renders a human-readable text-representation of a node.

    Various flags to control displayed data
    """

    def __init__(
        self,
        *,
        indent_pattern: str = "    ",
        write_dimension_name_flag: bool = False,
        write_dynamic_attributes: bool = False,
        write_primitive_type: bool = False,
        write_path: bool = True,
        write_static: bool = False,
        write_numpifiable: bool = False,
        write_block_byte_size=False,
        ignore_static_nodes: bool = False,
        ignore_non_dimensional_nodes: bool = False,
    ):
        """Constructor

        Args:
            indent_pattern: Indentation string.
            write_dimension_name_flag: Write the names of dimensions for Array Fixed.
            write_dynamic_attributes: Write dynamic attributes. Relevant only for dynamically decorated trees.
                See `render_dynamic_attributes`.
            write_primitive_type: Write the primitive type names for Leaf Nodes (eg `unsignedInteger-32`).
                Disabling it declutters the final output.
            write_path: Write the typesystem path pointing to the node.
                Useful for copypasting node reference for later querying.
            write_static: Write information about the staticity / quasi-staticity of a node.
            write_numpifiable: Write the node's `numpifiable` attribute.
                It means the data represented by the node can be viewed with numpy,
                given that enough coordinates are provided (matching the node's dimensionality)
            ignore_static_nodes: Ignore static nodes. Recommended for printing highest-level
                recommended numpifiable nodes, and decluttering the final output.
            ignore_non_dimensional_nodes: Ignore nodes with `dimensionality = None`.
                These nodes were bypassed during the dynamic decoration (phase 3)
        """
        self.lines: list[str] = []
        self.indentation = 0
        self.indent_pattern = indent_pattern
        self.write_dimension_name_flag = write_dimension_name_flag
        self.write_dynamic_attributes = write_dynamic_attributes
        self.write_primitive_type = write_primitive_type
        self.write_block_byte_size = write_block_byte_size

        self.write_path = write_path
        self.write_static = write_static
        self.write_numpifiable = write_numpifiable

        self.ignore_static_nodes = ignore_static_nodes
        self.ignore_non_dimensional_nodes = ignore_non_dimensional_nodes

    def write(self, line):
        self.lines.append(
            indent_string(self.indentation, line, indent_pattern=self.indent_pattern)
        )

    def write_node(self, node, string):
        self.write(
            f"{string} {self.render_numpifiable(node)} {self.render_static(node)} {self.render_block_byte_size(node)} "
            f"{self.render_dynamic_attributes(node)} {self.render_path(node)}"
        )

    def __str__(self):
        return "\n".join(self.lines)

    def nest(self):
        self.indentation += 1

    def unnest(self):
        self.indentation -= 1

    def guard(self, node: BaseNode):
        return (
            self.ignore_non_dimensional_nodes and node.dynamic_dimensionality is None
        ) or (self.ignore_static_nodes and node.static)

    def visit_leaf(self, node: LeafNode):
        if self.guard(node):
            return
        self.write_node(node, f"{render_leaf_node(node, self.write_primitive_type)}")

    def visit_struct(self, node: StructNode):
        if self.guard(node):
            return
        if node.name:
            self.write_node(node, f"{node.name}")
            self.nest()

        for child in node.children:
            child.accept(self)

        if node.name:
            self.unnest()

    def visit_array_fixed(self, node: ArrayFixedNode):
        if self.guard(node):
            return
        rendered_sizes = render_dimensions(
            node.dimensions, write_dimension=self.write_dimension_name_flag
        )
        self.write_node(node, f"{node.name}[{rendered_sizes}]")
        self.nest()
        node.child.accept(self)
        self.unnest()

    def visit_array_variable(self, node: ArrayVariableNode):
        if self.guard(node):
            return
        self.write_node(node, f"{node.name}[{render_leaf_node(node.size_ref)}]")
        self.nest()
        node.child.accept(self)
        self.unnest()

    def render_dynamic_attributes(self, node: BaseNode):
        if not self.write_dynamic_attributes:
            return ""
        return render_dynamic_attributes(node)

    def render_path(self, node: BaseNode):
        if not self.write_path or node.path is None:
            return ""
        return render_path(node.path)

    def render_block_byte_size(self, node: BaseNode):
        if not self.write_block_byte_size or node.path is None:
            return ""
        return f"<size={node.block_byte_size}>"

    def render_static(self, node: BaseNode):
        if not self.write_static:
            return ""

        string = "<static>" if node.static else "<non-static>"
        if isinstance(node, ArrayVariableNode):
            string += " "
            string += "<quasistatic>" if node.quasi_static else "<non-quasistatic>"
        return string

    def render_numpifiable(self, node: BaseNode):
        if not self.write_numpifiable:
            return ""
        return "<numpifiable>" if node.numpifiable else "<not-numpifiable>"

__init__(*, indent_pattern=' ', write_dimension_name_flag=False, write_dynamic_attributes=False, write_primitive_type=False, write_path=True, write_static=False, write_numpifiable=False, write_block_byte_size=False, ignore_static_nodes=False, ignore_non_dimensional_nodes=False)

Constructor

Parameters:

Name	Type	Description	Default
indent_pattern	str	Indentation string.	' '
write_dimension_name_flag	bool	Write the names of dimensions for Array Fixed.	False
write_dynamic_attributes	bool	Write dynamic attributes. Relevant only for dynamically decorated trees. See render_dynamic_attributes.	False
write_primitive_type	bool	Write the primitive type names for Leaf Nodes (eg unsignedInteger-32). Disabling it declutters the final output.	False
write_path	bool	Write the typesystem path pointing to the node. Useful for copypasting node reference for later querying.	True
write_static	bool	Write information about the staticity / quasi-staticity of a node.	False
write_numpifiable	bool	Write the node's numpifiable attribute. It means the data represented by the node can be viewed with numpy, given that enough coordinates are provided (matching the node's dimensionality)	False
ignore_static_nodes	bool	Ignore static nodes. Recommended for printing highest-level recommended numpifiable nodes, and decluttering the final output.	False
ignore_non_dimensional_nodes	bool	Ignore nodes with dimensionality = None. These nodes were bypassed during the dynamic decoration (phase 3)	False

Source code in smos_walker\core\visitors.py

def __init__(
    self,
    *,
    indent_pattern: str = "    ",
    write_dimension_name_flag: bool = False,
    write_dynamic_attributes: bool = False,
    write_primitive_type: bool = False,
    write_path: bool = True,
    write_static: bool = False,
    write_numpifiable: bool = False,
    write_block_byte_size=False,
    ignore_static_nodes: bool = False,
    ignore_non_dimensional_nodes: bool = False,
):
    """Constructor

    Args:
        indent_pattern: Indentation string.
        write_dimension_name_flag: Write the names of dimensions for Array Fixed.
        write_dynamic_attributes: Write dynamic attributes. Relevant only for dynamically decorated trees.
            See `render_dynamic_attributes`.
        write_primitive_type: Write the primitive type names for Leaf Nodes (eg `unsignedInteger-32`).
            Disabling it declutters the final output.
        write_path: Write the typesystem path pointing to the node.
            Useful for copypasting node reference for later querying.
        write_static: Write information about the staticity / quasi-staticity of a node.
        write_numpifiable: Write the node's `numpifiable` attribute.
            It means the data represented by the node can be viewed with numpy,
            given that enough coordinates are provided (matching the node's dimensionality)
        ignore_static_nodes: Ignore static nodes. Recommended for printing highest-level
            recommended numpifiable nodes, and decluttering the final output.
        ignore_non_dimensional_nodes: Ignore nodes with `dimensionality = None`.
            These nodes were bypassed during the dynamic decoration (phase 3)
    """
    self.lines: list[str] = []
    self.indentation = 0
    self.indent_pattern = indent_pattern
    self.write_dimension_name_flag = write_dimension_name_flag
    self.write_dynamic_attributes = write_dynamic_attributes
    self.write_primitive_type = write_primitive_type
    self.write_block_byte_size = write_block_byte_size

    self.write_path = write_path
    self.write_static = write_static
    self.write_numpifiable = write_numpifiable

    self.ignore_static_nodes = ignore_static_nodes
    self.ignore_non_dimensional_nodes = ignore_non_dimensional_nodes

ShallowDictExportNodeVisitor

Bases: NodeVisitor

Plain Visitor using the native dict representation methods of Nodes.

Source code in smos_walker\core\visitors.py

class ShallowDictExportNodeVisitor(NodeVisitor):
    """Plain Visitor using the native dict representation methods of Nodes."""

    def visit_leaf(self, node: LeafNode):
        return node.to_dict()

    def visit_struct(self, node: StructNode):
        return node.to_dict()

    def visit_array_fixed(self, node: ArrayFixedNode):
        return node.to_dict()

    def visit_array_variable(self, node: ArrayVariableNode):
        return node.to_dict()

SimpleQueryNodeVisitor

Bases: NodeVisitor

Query node by attribute. Tries to find at least one matching attribute in the node's dict

Return the first found node. name should be unique ideally.

Source code in smos_walker\core\visitors.py

class SimpleQueryNodeVisitor(NodeVisitor):
    """Query node by attribute. Tries to find at least one matching attribute in the node's dict

    Return the first found node. name should be unique ideally.
    """

    def __str__(self):
        return super().__str__() + "|" + f"{self.query_dict=}"

    def __init__(self, query_dict: dict[str, str]):
        self.query_dict = query_dict

    def node_match(self, node):
        for attribute_name, value in self.query_dict.items():
            # See https://stackoverflow.com/questions/61517/python-dictionary-from-an-objects-fields
            if attribute_name in vars(node) and vars(node)[attribute_name] == value:
                return True
        return False

    def visit_leaf(self, node: LeafNode):
        return node if self.node_match(node) else None

    def visit_struct(self, node: StructNode):
        if self.node_match(node):
            return node
        for child in node.yield_children():
            found = child.accept(self)
            if found:
                return found
        return None

    def visit_array_fixed(self, node: ArrayFixedNode):
        return self.visit_array_common(node)

    def visit_array_variable(self, node: ArrayVariableNode):
        return self.visit_array_common(node)

    def visit_array_common(self, node: ArrayFixedNode | ArrayVariableNode):
        if self.node_match(node):
            return node
        found = node.child.accept(self)
        if found:
            return found
        return None

UniversalPredicateNodeVisitor

Bases: AbstractPredicateNodeVisitor

Verify that all nodes in the tree do match a particular condition

Source code in smos_walker\core\visitors.py

class UniversalPredicateNodeVisitor(AbstractPredicateNodeVisitor):
    """Verify that all nodes in the tree do match a particular condition"""

    def visit_struct(self, node: StructNode):
        return self.predicate(node) and all(
            child.accept(self) for child in node.children
        )

    def visit_array_fixed(self, node: ArrayFixedNode):
        return self.predicate(node) and node.child.accept(self)

    def visit_array_variable(self, node: ArrayVariableNode):
        return self.predicate(node) and node.child.accept(self)

render_dimensions(dimensions, write_dimension=True, separator=', ')

Renders a string-representation of an ArrayFixed's dimensions.

You can test this function using doctest:

poetry run python -m doctest smos_walker/core/visitors.py

Examples:

>>> render_dimensions([{'name': 'model', 'size': 3}, {'name': 'pol', 'size': 8}])
'model=3, pol=8'
>>> render_dimensions([{'name': None, 'size': 2}])
'2'

Parameters:

Name	Type	Description	Default
dimensions	list[NodeDimension]	List of dimensions	required
write_dimension	bool	When enabled, writes the dimension's name into the rendered string. Defaults to True.	True
separator	str	Separator between the rendered dimensions. Defaults to ", ".	', '

Returns:

Type	Description
str	Rendered dimensions

Source code in smos_walker\core\visitors.py

def render_dimensions(
    dimensions: list[NodeDimension], write_dimension: bool = True, separator: str = ", "
) -> str:
    """Renders a string-representation of an ArrayFixed's dimensions.

    You can test this function using doctest:

    ```bash
    poetry run python -m doctest smos_walker/core/visitors.py
    ```

    Examples:
        >>> render_dimensions([{'name': 'model', 'size': 3}, {'name': 'pol', 'size': 8}])
        'model=3, pol=8'
        >>> render_dimensions([{'name': None, 'size': 2}])
        '2'

    Args:
        dimensions: List of dimensions
        write_dimension: When enabled, writes the dimension's name into the rendered string. Defaults to True.
        separator: Separator between the rendered dimensions. Defaults to ", ".

    Returns:
        Rendered dimensions
    """
    return separator.join(
        str(
            f"{dimension['name']}={dimension['size']}"
            if write_dimension and "name" in dimension and dimension["name"] is not None
            else dimension["size"]
        )
        for dimension in dimensions
    )

smos_walker.core.exception

SmosWalkerException

Bases: Exception

Generic exception related to the project.

Source code in smos_walker\core\exception.py

class SmosWalkerException(Exception):
    """Generic exception related to the project."""

smos_walker.data_reader

Reads datablocks.

smos_walker.data_reader.facade

read_dbl(datablock_folder_path)

Reads a Data Block file. A Data Block file (*.DBL) is a binary file respecting a XML binx schema.

Parameters:

Name	Type	Description	Default
datablock_folder_path	str	Path pointing towards the folder containing the datablock file (*.DBL)	required

Returns:

Type	Description
DataBlockType | None	A data block (numpy array of uint8) if read was successful, None otherwise.

Source code in smos_walker\data_reader\facade.py

def read_dbl(datablock_folder_path: str) -> DataBlockType | None:
    """Reads a Data Block file. A Data Block file (`*.DBL`) is a binary file respecting a XML _binx_ schema.

    Args:
        datablock_folder_path: Path pointing towards the **folder** containing the datablock file (`*.DBL`)

    Returns:
        A data block (numpy array of `uint8`) if read was successful, `None` otherwise.
    """
    return get_earth_explorer_dbl_raw_content(datablock_folder_path)

smos_walker.typesystem_parser

Step 1: Typesystem parser

smos_walker.typesystem_parser.facade

render_typesystem_parsed_tree_from_xml(xsd_path, xml_schema_path)

Generates the most basic tree from the XML schema, with the only addition being the developed typesystem.

Parameters:

Name	Type	Description	Default
xsd_path	str	Path pointing towards the XSD file, also known as "binx.xsd".	required
xml_schema_path	str	Path pointing towards an XML schema file.	required

Returns:

Type	Description
RawNode	A basic tree constituted of dicts, with the explicited typesystem.

Source code in smos_walker\typesystem_parser\facade.py

def render_typesystem_parsed_tree_from_xml(
    xsd_path: str, xml_schema_path: str
) -> RawNode:
    """Generates the most basic tree from the XML schema, with the only addition being the developed typesystem.

    Args:
        xsd_path: Path pointing towards the XSD file, also known as "binx.xsd".
        xml_schema_path: Path pointing towards an XML schema file.

    Returns:
        A basic tree constituted of dicts, with the explicited typesystem.
    """
    data = read_xml(xsd_path, xml_schema_path)
    tree = render_typesystem_parsed_tree(data, data)
    return tree

smos_walker.typesystem_static_decorator

Step 2: Typesystem static decorator

smos_walker.typesystem_static_decorator.facade

get_statically_decorated_tree(typesystem_parsed_tree)

Statically decorates a typesystem-developed raw tree. Uses the dedicated BaseNode data structure.

Takes the resulting tree from the typesystem parsing component and statically decorate it.

Note

The return tree is a new data structure. It should be leaving the passed argument unaltered.

Parameters:

Name	Type	Description	Default
typesystem_parsed_tree	any	The typesystem-parsed tree resulting from the typeystem_parser component.	required

Returns:

Type	Description
BaseNode	A statically decorated tree

Source code in smos_walker\typesystem_static_decorator\facade.py

def get_statically_decorated_tree(typesystem_parsed_tree: any) -> BaseNode:
    """Statically decorates a typesystem-developed raw tree. Uses the dedicated `BaseNode` data structure.

    Takes the resulting tree from the typesystem parsing component and statically decorate it.

    Note:
        The return tree is a new data structure. It should be leaving the passed argument unaltered.

    Args:
        typesystem_parsed_tree: The typesystem-parsed tree resulting from the `typeystem_parser` component.

    Returns:
        A statically decorated tree
    """
    return statically_decorate_tree(typesystem_parsed_tree)

load_statically_decorated_tree(xsd_path, xml_schema_path)

User-friendly way to load statically-decorated tree. It includes the usage of the typesystem parser.

Parameters:

Name	Type	Description	Default
xsd_path	str	Path pointing towards the XSD file, also known as "binx.xsd".	required
xml_schema_path	str	Path pointing towards an XML schema file.	required

Returns:

Type	Description
BaseNode	A statically decorated tree

Source code in smos_walker\typesystem_static_decorator\facade.py

def load_statically_decorated_tree(xsd_path: str, xml_schema_path: str) -> BaseNode:
    """User-friendly way to load statically-decorated tree. It includes the usage of the typesystem parser.

    Args:
        xsd_path: Path pointing towards the XSD file, also known as "binx.xsd".
        xml_schema_path: Path pointing towards an XML schema file.

    Returns:
        A statically decorated tree
    """
    typesystem_parsed_tree = render_typesystem_parsed_tree_from_xml(
        xsd_path, xml_schema_path
    )
    return get_statically_decorated_tree(typesystem_parsed_tree)

smos_walker.typesystem_dynamic_decorator

Step 3: Typesystem dynamic decorator

smos_walker.typesystem_dynamic_decorator.facade

dynamically_decorate_tree(statically_decorated_tree, datablock)

Dynamically decorates a statically-decorated typesystem tree IN-PLACE

Takes the resulting tree from the static decoration part and dynamically decorate it with the help of a concrete datablock complying to the typesystem described in the XML schema.

Warning

The tree is mutated IN-PLACE, this is NOT a pure function.

Parameters:

Name	Type	Description	Default
statically_decorated_tree	BaseNode	The statically-decorated tree resulting from the typesystem_static_decorator component.	required
datablock	DataBlockType	A whole datablock (.DBL file) that complies with the same XML schema used to generate the typesystem tree.	required

Raises:

Type	Description
SmosWalkerException	When the provided datablock is None.

Returns:

Type	Description
DynamicIndexerNodeVisitor	The visitor used to dynamically analyze the tree.

Source code in smos_walker\typesystem_dynamic_decorator\facade.py

def dynamically_decorate_tree(
    statically_decorated_tree: BaseNode, datablock: DataBlockType
) -> DynamicIndexerNodeVisitor:
    """Dynamically decorates a statically-decorated typesystem tree **IN-PLACE**

    Takes the resulting tree from the static decoration part and dynamically decorate it
    with the help of a concrete datablock complying to the typesystem described in the XML schema.

    Warning:
        The tree is mutated **IN-PLACE**, this is **NOT** a pure function.

    Args:
        statically_decorated_tree: The statically-decorated tree resulting from
            the `typesystem_static_decorator` component.
        datablock: A whole datablock (`.DBL` file) that complies with the same XML schema
            used to generate the typesystem tree.

    Raises:
        SmosWalkerException: When the provided `datablock` is `None`.

    Returns:
        The visitor used to dynamically analyze the tree.
    """
    if datablock is None:
        raise SmosWalkerException("DataBlock could not be read.")

    visitor = DynamicIndexerNodeVisitor(datablock)
    statically_decorated_tree.accept(visitor)

    return visitor

smos_walker.node_to_ndarray

Logic to convert nodes to numpy arrays (after dyanamic decoration)

smos_walker.node_to_ndarray.facade

mask_clip(tipping_point, size)

Clip values superior to a given tipping_point.

No check is performed to verify that tipping_point < size.

Examples:

>>> mask_values(5, 8)
array([False, False, False, False, False,  True,  True,  True])

Parameters:

Name	Type	Description	Default
tipping_point	int	Before this index, values are not masked, after, they do.	required
size	int	Size of desired mask	required

Returns:

Type	Description
BooleanNumpyArrayType	the mask

Source code in smos_walker\node_to_ndarray\facade.py

def mask_clip(tipping_point: int, size: int) -> BooleanNumpyArrayType:
    """Clip values superior to a given tipping_point.

    No check is performed to verify that tipping_point < size.

    Examples:
        >>> mask_values(5, 8)
        array([False, False, False, False, False,  True,  True,  True])

    Args:
        tipping_point: Before this index, values are not masked, after, they do.
        size: Size of desired mask

    Returns:
        the mask
    """
    mask = np.concatenate(
        (
            np.zeros(tipping_point, dtype=np.bool_),
            np.ones(size - tipping_point, dtype=np.bool_),
        )
    )
    return mask

node_to_ndarray_quasi_static(node, datablock, dynamic_size)

Numpy-typed view of a datablock's slice using a quasi-static array variable node

Parameters:

Name	Type	Description	Default
node	ArrayVariableNode	Dynamically-decorated node	required
datablock	DataBlockType	A datablock's slice	required
dynamic_size	int	Dynamic-size of the array variable. Note that this has to be an int, not a recursive list of int, as it is the criteria for quasi-staticity. The caller is responsible to provide this scalar value.	required

Raises:

Type	Description
SmosWalkerException	When the provided node is not a quasi-static array variable (guard).
SmosWalkerException	When the provided datablock's size does not match the one described by the provided node and provided dynamic_size (guard).

Returns:

Type	Description
NumpifiedDataBlockType	A numpy-typed view of the datablock's slice

Source code in smos_walker\node_to_ndarray\facade.py

def node_to_ndarray_quasi_static(
    node: ArrayVariableNode,
    datablock: DataBlockType,
    dynamic_size: int,
) -> NumpifiedDataBlockType:
    """Numpy-typed view of a datablock's slice using a quasi-static array variable node

    Args:
        node: Dynamically-decorated node
        datablock: A datablock's slice
        dynamic_size: Dynamic-size of the array variable.
            Note that this has to be an `int`, not a recursive list of `int`, as it is the criteria for quasi-staticity.
            The caller is responsible to provide this scalar value.

    Raises:
        SmosWalkerException: When the provided node is **not** a quasi-static array variable (guard).
        SmosWalkerException: When the provided datablock's size does not match the one described by
            the provided node and provided `dynamic_size` (guard).

    Returns:
        A numpy-typed view of the datablock's slice
    """
    if not (isinstance(node, ArrayVariableNode) and node.quasi_static):
        raise SmosWalkerException("The node is not a quasi-static ArrayVariable.")

    if dynamic_size * node.child.block_byte_size != len(datablock):
        raise SmosWalkerException(
            "The node's child's expected block byte size times the dynamic size "
            "of the node does not match with the binary slice"
        )

    type_info = node_to_type_info(node)
    return datablock.view(type_info["type"])

node_to_ndarray_static(node, datablock)

Numpy-typed view of a datablock's slice using a static node (Leaf, Struct, ArrayFixed)

Parameters:

Name	Type	Description	Default
node	LeafNode | StructNode | ArrayFixedNode	Dynamically-decorated static node	required
datablock	DataBlockType	A datablock's slice	required

Raises:

Type	Description
SmosWalkerException	When the provided node is not static (guard)
SmosWalkerException	When the provided datablock's size does not match the one described by the provided node (guard).

Returns:

Type	Description
NumpifiedDataBlockType	A numpy-typed view of the datablock's slice

Source code in smos_walker\node_to_ndarray\facade.py

def node_to_ndarray_static(
    node: LeafNode | StructNode | ArrayFixedNode,
    datablock: DataBlockType,
) -> NumpifiedDataBlockType:
    """Numpy-typed view of a datablock's slice using a static node (Leaf, Struct, ArrayFixed)

    Args:
        node: Dynamically-decorated static node
        datablock: A datablock's slice

    Raises:
        SmosWalkerException: When the provided node is **not** static (guard)
        SmosWalkerException: When the provided datablock's size does not match the one described
            by the provided node (guard).

    Returns:
        A numpy-typed view of the datablock's slice
    """
    if not node.static:
        raise SmosWalkerException(
            "The node [" + node.name + "] is not static (node_to_ndarray_static)"
        )
    if node.block_byte_size != len(datablock):
        raise SmosWalkerException(
            f"The node's expected block byte size ({node.block_byte_size=}) does not match with the datablock's slice "
            f"({len(datablock)=}) (delta: {node.block_byte_size - len(datablock)=})"
        )

    type_info = node_to_type_info(node)
    return datablock.view(type_info["type"])

node_to_ndarray_static_one_dynamic_dimensional(node, datablock)

summary

Parameters:

Name	Type	Description	Default
node	LeafNode | StructNode | ArrayFixedNode	Dynamically-decorated static node	required
datablock	DataBlockType	A datablock's slice	required

Raises:

Type	Description
SmosWalkerException	When the provided node is not static (guard)
SmosWalkerException	When the provided node's dynamic_dimensionality is not 1.
SmosWalkerException	When the provided datablock's detected contiguous bytes size does not match with the expected one described by the provided node (guard).

Returns:

Type	Description
NumpifiedDataBlockType	A numpy-typed view of the datablock's slice

Source code in smos_walker\node_to_ndarray\facade.py

def node_to_ndarray_static_one_dynamic_dimensional(
    node: LeafNode | StructNode | ArrayFixedNode, datablock: DataBlockType
) -> NumpifiedDataBlockType:
    """_summary_

    Args:
        node: Dynamically-decorated static node
        datablock: A datablock's slice

    Raises:
        SmosWalkerException: When the provided node is **not** static (guard)
        SmosWalkerException: When the provided node's `dynamic_dimensionality` is not 1.
        SmosWalkerException: When the provided datablock's detected contiguous bytes size does not match with
            the expected one described by the provided node (guard).

    Returns:
        A numpy-typed view of the datablock's slice
    """
    if not node.static:
        raise SmosWalkerException(
            "The node ["
            + node.name
            + "] is not static (node_to_ndarray_static_one_dynamic_dimensional)"
        )

    if not node.dynamic_dimensionality == 1:
        raise SmosWalkerException(
            "This method is adapted to numpify static nodes with dynamic_dimensionality = 1 "
            "(given dynamic_dimensionality = " + str(node.dynamic_dimensionality) + ")"
        )

    contiguous_bytes = datablock[
        (
            np.array(node.dynamic_offset).reshape(-1, 1)
            + np.arange(node.block_byte_size).reshape(1, -1)
        ).reshape(-1)
    ]

    if not (
        node.dynamic_coverage
        == len(node.dynamic_offset) * node.block_byte_size
        == len(contiguous_bytes)
    ):
        raise SmosWalkerException(
            f"The node's expected block byte size does not match with the binary slice "
            f"{len(node.dynamic_offset) * node.block_byte_size=}{node.dynamic_coverage=} "
            f"{node.block_byte_size=} {len(node.dynamic_offset)=} {len(contiguous_bytes)=}"
        )

    type_info = node_to_type_info(node)
    return contiguous_bytes.view(type_info["type"])