taps.run.env

taps/run/env.py

Hardware dataclass

Hardware(
    architecture: str,
    physical_cores: int,
    logical_cores: int,
    memory_capacity: float,
)

Hardware information.

Parameters:

• architecture (str) –

  CPU architecture.

• physical_cores (str) –

  CPU physical core count.

• logical_cores (str) –

  CPU logical core count.

• memory_capacity (str) –

  Memory capacity in GB.

collect classmethod

collect() -> Self

Collect hardware information.

Source code in taps/run/env.py

@classmethod
def collect(cls) -> Self:
    """Collect hardware information."""
    return cls(
        architecture=platform.machine(),
        physical_cores=psutil.cpu_count(logical=False),
        logical_cores=psutil.cpu_count(logical=True),
        memory_capacity=round(psutil.virtual_memory().total / 1e9, 2),
    )

Packages dataclass

Packages(
    dask: str,
    globus_compute: str,
    numpy: str,
    parsl: str,
    proxystore: str,
    pydantic: str,
    ray: str,
    taps: str,
)

Python package versions.

Parameters:

• dask (str) –

  Dask/Distributed version.

• globus_compute (str) –

  Globus Compute version.

• numpy (str) –

  Numpy version.

• parsl (str) –

  Parsl version.

• proxystore (str) –

  ProxyStore version.

• pydantic (str) –

  Pydantic version.

• ray (str) –

  Ray version.

• taps (str) –

  TaPS version.

collect classmethod

collect() -> Self

Collect package version information.

Source code in taps/run/env.py

@classmethod
def collect(cls) -> Self:
    """Collect package version information."""
    return cls(
        dask=_get_version('dask'),
        globus_compute=_get_version('globus_compute_sdk'),
        numpy=_get_version('numpy'),
        parsl=_get_version('parsl'),
        proxystore=_get_version('proxystore'),
        pydantic=_get_version('pydantic'),
        ray=_get_version('ray'),
        taps=_get_version('taps'),
    )

Python dataclass

Python(
    version: str,
    implementation: str,
    compiler: str,
    bit_length: int,
)

Python interpreter information.

Parameters:

• version (str) –

  Python version.

• implementation (str) –

  Python implementation.

• compiler (str) –

  Compiler used to compile this interpreter

• bit_length (str) –

  Bit-length of the interpreter (e.g., 32 vs 64-bit).

collect classmethod

collect() -> Self

Collect Python interpreter information.

Source code in taps/run/env.py

@classmethod
def collect(cls) -> Self:
    """Collect Python interpreter information."""
    return cls(
        version=platform.python_version(),
        implementation=platform.python_implementation(),
        compiler=platform.python_compiler(),
        bit_length=sys.maxsize.bit_length() + 1,
    )

System dataclass

System(hostname: str, platform: str, platform_ext: str)

System information.

Parameters:

• hostname (str) –

  Network name of node.

• platform (str) –

  Platform identifier.

• platform_ext (str) –

  Extended platform information.

collect classmethod

collect() -> Self

Collect system information.

Source code in taps/run/env.py

@classmethod
def collect(cls) -> Self:
    """Collect system information."""
    return cls(
        hostname=platform.node(),
        platform=sys.platform,
        platform_ext=platform.platform(),
    )

Environment dataclass

Environment(
    hardware: Hardware,
    packages: Packages,
    python: Python,
    system: System,
)

Environment information.

To view the current environment:

$ python -m taps.run.env
system:
  hostname: ...
  os: linux (...)
  cpu: x86_64 (8 cores / 16 logical)
  memory: 16 GB
python:
  version: 3.11.9
  build: CPython (64-bit runtime) [GCC 11.4.0]
packages:
  ...
  taps: 0.2.1

Parameters:

• hardware (str) –

  Hardware information.

• packages (str) –

  Python package versions.

• python (str) –

  Python interpreter information.

• system (str) –

  System information.

collect classmethod

collect() -> Self

Collect information on the current environment.

Source code in taps/run/env.py

@classmethod
def collect(cls) -> Self:
    """Collect information on the current environment."""
    return cls(
        hardware=Hardware.collect(),
        packages=Packages.collect(),
        python=Python.collect(),
        system=System.collect(),
    )

format

format() -> str

Format environment as a human-readable string.

Source code in taps/run/env.py

    def format(self) -> str:
        """Format environment as a human-readable string."""
        pcores = self.hardware.physical_cores
        lcores = self.hardware.logical_cores
        impl = self.python.implementation
        compiler = self.python.compiler
        bit_length = self.python.bit_length

        packages = [
            f'{name}: {version}\n'
            for name, version in dataclasses.asdict(self.packages).items()
        ]

        return f"""\
system:
  hostname: {self.system.hostname}
  os: {self.system.platform} ({self.system.platform_ext})
  cpu: {self.hardware.architecture} ({pcores} cores / {lcores} logical)
  memory: {self.hardware.memory_capacity} GB
python:
  version: {self.python.version}
  build: {impl} ({bit_length}-bit runtime) [{compiler}]
packages:
  {"  ".join(packages)}
""".strip()

json

json() -> dict[str, Any]

Get environment as JSON-compatible dictionary.

Source code in taps/run/env.py

def json(self) -> dict[str, Any]:
    """Get environment as JSON-compatible dictionary."""
    return dataclasses.asdict(self)

write_json

write_json(filepath: str | Path) -> None

Write environment to JSON file.

Parameters:

• filepath (str | Path) –

  JSON filepath.

Source code in taps/run/env.py

def write_json(self, filepath: str | pathlib.Path) -> None:
    """Write environment to JSON file.

    Args:
        filepath: JSON filepath.
    """
    filepath = pathlib.Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)
    with open(filepath, 'w') as f:
        json.dump(self.json(), f, indent=4, sort_keys=True)