Skip to content

XetoCLI

XetoCLI

A client interface to Haxall's Xeto CLI tools for type checking and analysis of Haystack records.

This API is experimental and subject to change in future releases.

XetoCLI can be directly imported as follows:

from phable import XetoCLI
Source code in phable/xeto_cli.py
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class XetoCLI:
    """A client interface to Haxall's Xeto CLI tools for type checking and analysis
    of Haystack records.

    This API is experimental and subject to change in future releases.

    `XetoCLI` can be directly imported as follows:

    ```python
    from phable import XetoCLI
    ```
    """

    def __init__(
        self,
        *,
        local_haxall_path: str | Path | None = None,
        io_format: Literal["json", "zinc"] = "zinc",
    ):
        """Initialize a `XetoCLI` instance.

        Parameters:
            local_haxall_path:
                Optional path to a local Haxall installation directory. If provided,
                commands will execute against the local installation. If `None`, commands
                will execute via a Docker container called `phable_haxall_cli_run` that is assumed
                to be running. The `phable_haxall_cli_run` Docker container can be built and started
                by cloning [phable](https://github.com/rick-jennings/phable) and following the instructions
                [here](https://github.com/rick-jennings/phable/blob/main/tests/xeto_cli/README.md).
            io_format:
                Data serialization format for communication with Haxall. Either `json`
                or `zinc`. Defaults to `zinc`.
        """
        self._local_haxall_bin_path = (
            Path(local_haxall_path) / "bin" if local_haxall_path is not None else None
        )
        self._io_format = io_format
        self._encoder = PH_IO_FACTORY[io_format]["encoder"]
        self._decoder = PH_IO_FACTORY[io_format]["decoder"]

    def fits_explain(self, recs: list[dict[str, Any]], graph: bool = True) -> Grid:
        """Analyze records against Xeto type specifications and return detailed
        explanations.

        This method executes the Haxall `xeto fits` command to determine whether the
        provided records conform to their declared Xeto types. It returns a detailed
        explanation of type conformance, including any type mismatches or missing
        required tags.

        **Example:**

        ```python
        from phable import Marker, Number, Ref, XetoCLI

        cli = XetoCLI()
        recs = [
            {
                "dis": "Site 1",
                "site": Marker(),
                "area": Number(1_000, "square_foot"),
                "spec": Ref("ph::Site"),
            }
        ]
        result = cli.fits_explain(recs)
        ```

        Parameters:
            recs:
                List of Haystack record dictionaries to analyze. Each record should
                specify a Xeto type with a `spec` tag.
            graph:
                If `True`, includes a detailed graph output showing the type hierarchy and
                conformance details. Defaults to `True`.

        Returns:
            `Grid` explaining how each record fits (or fails to fit) its expected Xeto type specification.
        """

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=f".{self._io_format}", delete=False
        ) as temp_file:
            temp_file.write(self._encoder.to_str(recs))
            temp_file.flush()
            temp_file_path = temp_file.name

        try:
            if self._local_haxall_bin_path is None:
                cli_stdout = _exec_docker_cmd(self._io_format, graph, temp_file_path)
            else:
                cli_stdout = _exec_localhost_cmd(
                    self._io_format, graph, temp_file_path, self._local_haxall_bin_path
                )
            return self._decoder.from_str(cli_stdout)
        finally:
            try:
                os.unlink(temp_file_path)
            except OSError:
                pass

__init__

__init__(*, local_haxall_path=None, io_format='zinc')

Initialize a XetoCLI instance.

Parameters:

Name Type Description Default
local_haxall_path str | pathlib.Path | None

Optional path to a local Haxall installation directory. If provided, commands will execute against the local installation. If None, commands will execute via a Docker container called phable_haxall_cli_run that is assumed to be running. The phable_haxall_cli_run Docker container can be built and started by cloning phable and following the instructions here.

None
io_format typing.Literal['json', 'zinc']

Data serialization format for communication with Haxall. Either json or zinc. Defaults to zinc.

'zinc'
Source code in phable/xeto_cli.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(
    self,
    *,
    local_haxall_path: str | Path | None = None,
    io_format: Literal["json", "zinc"] = "zinc",
):
    """Initialize a `XetoCLI` instance.

    Parameters:
        local_haxall_path:
            Optional path to a local Haxall installation directory. If provided,
            commands will execute against the local installation. If `None`, commands
            will execute via a Docker container called `phable_haxall_cli_run` that is assumed
            to be running. The `phable_haxall_cli_run` Docker container can be built and started
            by cloning [phable](https://github.com/rick-jennings/phable) and following the instructions
            [here](https://github.com/rick-jennings/phable/blob/main/tests/xeto_cli/README.md).
        io_format:
            Data serialization format for communication with Haxall. Either `json`
            or `zinc`. Defaults to `zinc`.
    """
    self._local_haxall_bin_path = (
        Path(local_haxall_path) / "bin" if local_haxall_path is not None else None
    )
    self._io_format = io_format
    self._encoder = PH_IO_FACTORY[io_format]["encoder"]
    self._decoder = PH_IO_FACTORY[io_format]["decoder"]

fits_explain

fits_explain(recs, graph=True)

Analyze records against Xeto type specifications and return detailed explanations.

This method executes the Haxall xeto fits command to determine whether the provided records conform to their declared Xeto types. It returns a detailed explanation of type conformance, including any type mismatches or missing required tags.

Example:

from phable import Marker, Number, Ref, XetoCLI

cli = XetoCLI()
recs = [
    {
        "dis": "Site 1",
        "site": Marker(),
        "area": Number(1_000, "square_foot"),
        "spec": Ref("ph::Site"),
    }
]
result = cli.fits_explain(recs)

Parameters:

Name Type Description Default
recs list[dict[str, typing.Any]]

List of Haystack record dictionaries to analyze. Each record should specify a Xeto type with a spec tag.

required
graph bool

If True, includes a detailed graph output showing the type hierarchy and conformance details. Defaults to True.

True

Returns:

Type Description
phable.Grid

Grid explaining how each record fits (or fails to fit) its expected Xeto type specification.

Source code in phable/xeto_cli.py
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def fits_explain(self, recs: list[dict[str, Any]], graph: bool = True) -> Grid:
    """Analyze records against Xeto type specifications and return detailed
    explanations.

    This method executes the Haxall `xeto fits` command to determine whether the
    provided records conform to their declared Xeto types. It returns a detailed
    explanation of type conformance, including any type mismatches or missing
    required tags.

    **Example:**

    ```python
    from phable import Marker, Number, Ref, XetoCLI

    cli = XetoCLI()
    recs = [
        {
            "dis": "Site 1",
            "site": Marker(),
            "area": Number(1_000, "square_foot"),
            "spec": Ref("ph::Site"),
        }
    ]
    result = cli.fits_explain(recs)
    ```

    Parameters:
        recs:
            List of Haystack record dictionaries to analyze. Each record should
            specify a Xeto type with a `spec` tag.
        graph:
            If `True`, includes a detailed graph output showing the type hierarchy and
            conformance details. Defaults to `True`.

    Returns:
        `Grid` explaining how each record fits (or fails to fit) its expected Xeto type specification.
    """

    with tempfile.NamedTemporaryFile(
        mode="w", suffix=f".{self._io_format}", delete=False
    ) as temp_file:
        temp_file.write(self._encoder.to_str(recs))
        temp_file.flush()
        temp_file_path = temp_file.name

    try:
        if self._local_haxall_bin_path is None:
            cli_stdout = _exec_docker_cmd(self._io_format, graph, temp_file_path)
        else:
            cli_stdout = _exec_localhost_cmd(
                self._io_format, graph, temp_file_path, self._local_haxall_bin_path
            )
        return self._decoder.from_str(cli_stdout)
    finally:
        try:
            os.unlink(temp_file_path)
        except OSError:
            pass