Environment

Environment implementation.

Environment

Provides environment information.

Individual environment components can be registered separately to extend the global Environment object with additional information (see BaseEnvironmentComponent).

Source code in src/zenml/environment.py

class Environment(metaclass=SingletonMetaClass):
    """Provides environment information.

    Individual environment components can be registered separately to extend
    the global Environment object with additional information (see
    `BaseEnvironmentComponent`).
    """

    def __init__(self) -> None:
        """Initializes an Environment instance.

        Note: Environment is a singleton class, which means this method will
        only get called once. All following `Environment()` calls will return
        the previously initialized instance.
        """

    @staticmethod
    def get_system_info() -> Dict[str, str]:
        """Information about the operating system.

        Returns:
            A dictionary containing information about the operating system.
        """
        system = platform.system()

        if system == "Windows":
            release, version, csd, ptype = platform.win32_ver()

            return {
                "os": "windows",
                "windows_version_release": release,
                "windows_version": version,
                "windows_version_service_pack": csd,
                "windows_version_os_type": ptype,
            }

        if system == "Darwin":
            return {"os": "mac", "mac_version": platform.mac_ver()[0]}

        if system == "Linux":
            return {
                "os": "linux",
                "linux_distro": distro.id(),
                "linux_distro_like": distro.like(),
                "linux_distro_version": distro.version(),
            }

        # We don't collect data for any other system.
        return {"os": "unknown"}

    @staticmethod
    def python_version() -> str:
        """Returns the python version of the running interpreter.

        Returns:
            str: the python version
        """
        return platform.python_version()

    @staticmethod
    def in_container() -> bool:
        """If the current python process is running in a container.

        Returns:
            `True` if the current python process is running in a
            container, `False` otherwise.
        """
        # TODO [ENG-167]: Make this more reliable and add test.
        return INSIDE_ZENML_CONTAINER

    @staticmethod
    def in_docker() -> bool:
        """If the current python process is running in a docker container.

        Returns:
            `True` if the current python process is running in a docker
            container, `False` otherwise.
        """
        if os.path.exists("./dockerenv") or os.path.exists("/.dockerinit"):
            return True

        try:
            with open("/proc/1/cgroup", "rt") as ifh:
                info = ifh.read()
                return "docker" in info
        except (FileNotFoundError, Exception):
            return False

    @staticmethod
    def in_kubernetes() -> bool:
        """If the current python process is running in a kubernetes pod.

        Returns:
            `True` if the current python process is running in a kubernetes
            pod, `False` otherwise.
        """
        if "KUBERNETES_SERVICE_HOST" in os.environ:
            return True

        try:
            with open("/proc/1/cgroup", "rt") as ifh:
                info = ifh.read()
                return "kubepod" in info
        except (FileNotFoundError, Exception):
            return False

    @staticmethod
    def in_google_colab() -> bool:
        """If the current Python process is running in a Google Colab.

        Returns:
            `True` if the current Python process is running in a Google Colab,
            `False` otherwise.
        """
        try:
            import google.colab  # noqa

            return True

        except ModuleNotFoundError:
            return False

    @staticmethod
    def in_notebook() -> bool:
        """If the current Python process is running in a notebook.

        Returns:
            `True` if the current Python process is running in a notebook,
            `False` otherwise.
        """
        if Environment.in_google_colab():
            return True

        try:
            ipython = get_ipython()  # type: ignore[name-defined]
        except NameError:
            return False

        if ipython.__class__.__name__ in [
            "TerminalInteractiveShell",
            "ZMQInteractiveShell",
            "DatabricksShell",
        ]:
            return True
        return False

    @staticmethod
    def in_github_codespaces() -> bool:
        """If the current Python process is running in GitHub Codespaces.

        Returns:
            `True` if the current Python process is running in GitHub Codespaces,
            `False` otherwise.
        """
        return (
            "CODESPACES" in os.environ
            or "GITHUB_CODESPACE_TOKEN" in os.environ
            or "GITHUB_CODESPACES_PORT_FORWARDING_DOMAIN" in os.environ
        )

    @staticmethod
    def in_vscode_remote_container() -> bool:
        """If the current Python process is running in a VS Code Remote Container.

        Returns:
            `True` if the current Python process is running in a VS Code Remote Container,
            `False` otherwise.
        """
        return (
            "REMOTE_CONTAINERS" in os.environ
            or "VSCODE_REMOTE_CONTAINERS_SESSION" in os.environ
        )

    @staticmethod
    def in_paperspace_gradient() -> bool:
        """If the current Python process is running in Paperspace Gradient.

        Returns:
            `True` if the current Python process is running in Paperspace
            Gradient, `False` otherwise.
        """
        return "PAPERSPACE_NOTEBOOK_REPO_ID" in os.environ

    @staticmethod
    def in_github_actions() -> bool:
        """If the current Python process is running in GitHub Actions.

        Returns:
            `True` if the current Python process is running in GitHub
            Actions, `False` otherwise.
        """
        return "GITHUB_ACTIONS" in os.environ

    @staticmethod
    def in_gitlab_ci() -> bool:
        """If the current Python process is running in GitLab CI.

        Returns:
            `True` if the current Python process is running in GitLab
            CI, `False` otherwise.
        """
        return "GITLAB_CI" in os.environ

    @staticmethod
    def in_circle_ci() -> bool:
        """If the current Python process is running in Circle CI.

        Returns:
            `True` if the current Python process is running in Circle
            CI, `False` otherwise.
        """
        return "CIRCLECI" in os.environ

    @staticmethod
    def in_bitbucket_ci() -> bool:
        """If the current Python process is running in Bitbucket CI.

        Returns:
            `True` if the current Python process is running in Bitbucket
            CI, `False` otherwise.
        """
        return "BITBUCKET_BUILD_NUMBER" in os.environ

    @staticmethod
    def in_ci() -> bool:
        """If the current Python process is running in any CI.

        Returns:
            `True` if the current Python process is running in any
            CI, `False` otherwise.
        """
        return "CI" in os.environ

    @staticmethod
    def in_wsl() -> bool:
        """If the current process is running in Windows Subsystem for Linux.

        source: https://www.scivision.dev/python-detect-wsl/

        Returns:
            `True` if the current process is running in WSL, `False` otherwise.
        """
        return "microsoft-standard" in platform.uname().release

    @staticmethod
    def in_lightning_ai_studio() -> bool:
        """If the current Python process is running in Lightning.ai studios.

        Returns:
            `True` if the current Python process is running in Lightning.ai studios,
            `False` otherwise.
        """
        return (
            "LIGHTNING_CLOUD_URL" in os.environ
            and "LIGHTNING_CLOUDSPACE_HOST" in os.environ
        )

__init__()

Initializes an Environment instance.

Note: Environment is a singleton class, which means this method will only get called once. All following Environment() calls will return the previously initialized instance.

Source code in src/zenml/environment.py

def __init__(self) -> None:
    """Initializes an Environment instance.

    Note: Environment is a singleton class, which means this method will
    only get called once. All following `Environment()` calls will return
    the previously initialized instance.
    """

get_system_info() staticmethod

Information about the operating system.

Returns:

Type	Description
Dict[str, str]	A dictionary containing information about the operating system.

Source code in src/zenml/environment.py

@staticmethod
def get_system_info() -> Dict[str, str]:
    """Information about the operating system.

    Returns:
        A dictionary containing information about the operating system.
    """
    system = platform.system()

    if system == "Windows":
        release, version, csd, ptype = platform.win32_ver()

        return {
            "os": "windows",
            "windows_version_release": release,
            "windows_version": version,
            "windows_version_service_pack": csd,
            "windows_version_os_type": ptype,
        }

    if system == "Darwin":
        return {"os": "mac", "mac_version": platform.mac_ver()[0]}

    if system == "Linux":
        return {
            "os": "linux",
            "linux_distro": distro.id(),
            "linux_distro_like": distro.like(),
            "linux_distro_version": distro.version(),
        }

    # We don't collect data for any other system.
    return {"os": "unknown"}

in_bitbucket_ci() staticmethod

If the current Python process is running in Bitbucket CI.

Returns:

Type	Description
bool	True if the current Python process is running in Bitbucket
bool	CI, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_bitbucket_ci() -> bool:
    """If the current Python process is running in Bitbucket CI.

    Returns:
        `True` if the current Python process is running in Bitbucket
        CI, `False` otherwise.
    """
    return "BITBUCKET_BUILD_NUMBER" in os.environ

in_ci() staticmethod

If the current Python process is running in any CI.

Returns:

Type	Description
bool	True if the current Python process is running in any
bool	CI, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_ci() -> bool:
    """If the current Python process is running in any CI.

    Returns:
        `True` if the current Python process is running in any
        CI, `False` otherwise.
    """
    return "CI" in os.environ

in_circle_ci() staticmethod

If the current Python process is running in Circle CI.

Returns:

Type	Description
bool	True if the current Python process is running in Circle
bool	CI, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_circle_ci() -> bool:
    """If the current Python process is running in Circle CI.

    Returns:
        `True` if the current Python process is running in Circle
        CI, `False` otherwise.
    """
    return "CIRCLECI" in os.environ

in_container() staticmethod

If the current python process is running in a container.

Returns:

Type	Description
bool	True if the current python process is running in a
bool	container, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_container() -> bool:
    """If the current python process is running in a container.

    Returns:
        `True` if the current python process is running in a
        container, `False` otherwise.
    """
    # TODO [ENG-167]: Make this more reliable and add test.
    return INSIDE_ZENML_CONTAINER

in_docker() staticmethod

If the current python process is running in a docker container.

Returns:

Type	Description
bool	True if the current python process is running in a docker
bool	container, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_docker() -> bool:
    """If the current python process is running in a docker container.

    Returns:
        `True` if the current python process is running in a docker
        container, `False` otherwise.
    """
    if os.path.exists("./dockerenv") or os.path.exists("/.dockerinit"):
        return True

    try:
        with open("/proc/1/cgroup", "rt") as ifh:
            info = ifh.read()
            return "docker" in info
    except (FileNotFoundError, Exception):
        return False

in_github_actions() staticmethod

If the current Python process is running in GitHub Actions.

Returns:

Type	Description
bool	True if the current Python process is running in GitHub
bool	Actions, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_github_actions() -> bool:
    """If the current Python process is running in GitHub Actions.

    Returns:
        `True` if the current Python process is running in GitHub
        Actions, `False` otherwise.
    """
    return "GITHUB_ACTIONS" in os.environ

in_github_codespaces() staticmethod

If the current Python process is running in GitHub Codespaces.

Returns:

Type	Description
bool	True if the current Python process is running in GitHub Codespaces,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_github_codespaces() -> bool:
    """If the current Python process is running in GitHub Codespaces.

    Returns:
        `True` if the current Python process is running in GitHub Codespaces,
        `False` otherwise.
    """
    return (
        "CODESPACES" in os.environ
        or "GITHUB_CODESPACE_TOKEN" in os.environ
        or "GITHUB_CODESPACES_PORT_FORWARDING_DOMAIN" in os.environ
    )

in_gitlab_ci() staticmethod

If the current Python process is running in GitLab CI.

Returns:

Type	Description
bool	True if the current Python process is running in GitLab
bool	CI, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_gitlab_ci() -> bool:
    """If the current Python process is running in GitLab CI.

    Returns:
        `True` if the current Python process is running in GitLab
        CI, `False` otherwise.
    """
    return "GITLAB_CI" in os.environ

in_google_colab() staticmethod

If the current Python process is running in a Google Colab.

Returns:

Type	Description
bool	True if the current Python process is running in a Google Colab,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_google_colab() -> bool:
    """If the current Python process is running in a Google Colab.

    Returns:
        `True` if the current Python process is running in a Google Colab,
        `False` otherwise.
    """
    try:
        import google.colab  # noqa

        return True

    except ModuleNotFoundError:
        return False

in_kubernetes() staticmethod

If the current python process is running in a kubernetes pod.

Returns:

Type	Description
bool	True if the current python process is running in a kubernetes
bool	pod, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_kubernetes() -> bool:
    """If the current python process is running in a kubernetes pod.

    Returns:
        `True` if the current python process is running in a kubernetes
        pod, `False` otherwise.
    """
    if "KUBERNETES_SERVICE_HOST" in os.environ:
        return True

    try:
        with open("/proc/1/cgroup", "rt") as ifh:
            info = ifh.read()
            return "kubepod" in info
    except (FileNotFoundError, Exception):
        return False

in_lightning_ai_studio() staticmethod

If the current Python process is running in Lightning.ai studios.

Returns:

Type	Description
bool	True if the current Python process is running in Lightning.ai studios,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_lightning_ai_studio() -> bool:
    """If the current Python process is running in Lightning.ai studios.

    Returns:
        `True` if the current Python process is running in Lightning.ai studios,
        `False` otherwise.
    """
    return (
        "LIGHTNING_CLOUD_URL" in os.environ
        and "LIGHTNING_CLOUDSPACE_HOST" in os.environ
    )

in_notebook() staticmethod

If the current Python process is running in a notebook.

Returns:

Type	Description
bool	True if the current Python process is running in a notebook,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_notebook() -> bool:
    """If the current Python process is running in a notebook.

    Returns:
        `True` if the current Python process is running in a notebook,
        `False` otherwise.
    """
    if Environment.in_google_colab():
        return True

    try:
        ipython = get_ipython()  # type: ignore[name-defined]
    except NameError:
        return False

    if ipython.__class__.__name__ in [
        "TerminalInteractiveShell",
        "ZMQInteractiveShell",
        "DatabricksShell",
    ]:
        return True
    return False

in_paperspace_gradient() staticmethod

If the current Python process is running in Paperspace Gradient.

Returns:

Type	Description
bool	True if the current Python process is running in Paperspace
bool	Gradient, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_paperspace_gradient() -> bool:
    """If the current Python process is running in Paperspace Gradient.

    Returns:
        `True` if the current Python process is running in Paperspace
        Gradient, `False` otherwise.
    """
    return "PAPERSPACE_NOTEBOOK_REPO_ID" in os.environ

in_vscode_remote_container() staticmethod

If the current Python process is running in a VS Code Remote Container.

Returns:

Type	Description
bool	True if the current Python process is running in a VS Code Remote Container,
bool	False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_vscode_remote_container() -> bool:
    """If the current Python process is running in a VS Code Remote Container.

    Returns:
        `True` if the current Python process is running in a VS Code Remote Container,
        `False` otherwise.
    """
    return (
        "REMOTE_CONTAINERS" in os.environ
        or "VSCODE_REMOTE_CONTAINERS_SESSION" in os.environ
    )

in_wsl() staticmethod

If the current process is running in Windows Subsystem for Linux.

source: https://www.scivision.dev/python-detect-wsl/

Returns:

Type	Description
bool	True if the current process is running in WSL, False otherwise.

Source code in src/zenml/environment.py

@staticmethod
def in_wsl() -> bool:
    """If the current process is running in Windows Subsystem for Linux.

    source: https://www.scivision.dev/python-detect-wsl/

    Returns:
        `True` if the current process is running in WSL, `False` otherwise.
    """
    return "microsoft-standard" in platform.uname().release

python_version() staticmethod

Returns the python version of the running interpreter.

Returns:

Name	Type	Description
str	str	the python version

Source code in src/zenml/environment.py

@staticmethod
def python_version() -> str:
    """Returns the python version of the running interpreter.

    Returns:
        str: the python version
    """
    return platform.python_version()

get_environment()

Returns a string representing the execution environment of the pipeline.

Returns:

Name	Type	Description
str	str	the execution environment

Source code in src/zenml/environment.py

def get_environment() -> str:
    """Returns a string representing the execution environment of the pipeline.

    Returns:
        str: the execution environment
    """
    # Order is important here
    if Environment.in_kubernetes():
        return EnvironmentType.KUBERNETES
    elif Environment.in_github_actions():
        return EnvironmentType.GITHUB_ACTION
    elif Environment.in_gitlab_ci():
        return EnvironmentType.GITLAB_CI
    elif Environment.in_circle_ci():
        return EnvironmentType.CIRCLE_CI
    elif Environment.in_bitbucket_ci():
        return EnvironmentType.BITBUCKET_CI
    elif Environment.in_ci():
        return EnvironmentType.GENERIC_CI
    elif Environment.in_github_codespaces():
        return EnvironmentType.GITHUB_CODESPACES
    elif Environment.in_vscode_remote_container():
        return EnvironmentType.VSCODE_REMOTE_CONTAINER
    elif Environment.in_lightning_ai_studio():
        return EnvironmentType.LIGHTNING_AI_STUDIO
    elif Environment.in_docker():
        return EnvironmentType.DOCKER
    elif Environment.in_container():
        return EnvironmentType.CONTAINER
    elif Environment.in_google_colab():
        return EnvironmentType.COLAB
    elif Environment.in_paperspace_gradient():
        return EnvironmentType.PAPERSPACE
    elif Environment.in_notebook():
        return EnvironmentType.NOTEBOOK
    elif Environment.in_wsl():
        return EnvironmentType.WSL
    else:
        return EnvironmentType.NATIVE

get_run_environment_dict()

Returns a dictionary of the current run environment.

Everything that is returned here will be saved in the DB as pipeline_run.client_environment and pipeline_run.orchestrator_environment for client and orchestrator respectively.

Returns:

Type	Description
Dict[str, str]	A dictionary of the current run environment.

Source code in src/zenml/environment.py

def get_run_environment_dict() -> Dict[str, str]:
    """Returns a dictionary of the current run environment.

    Everything that is returned here will be saved in the DB as
    `pipeline_run.client_environment` and
    `pipeline_run.orchestrator_environment` for client and orchestrator
    respectively.

    Returns:
        A dictionary of the current run environment.
    """
    return {
        "environment": get_environment(),
        **Environment.get_system_info(),
        "python_version": Environment.python_version(),
    }

get_system_details()

Returns OS, python and ZenML information.

Returns:

Name	Type	Description
str	str	OS, python and ZenML information

Source code in src/zenml/environment.py

def get_system_details() -> str:
    """Returns OS, python and ZenML information.

    Returns:
        str: OS, python and ZenML information
    """
    from zenml.integrations.registry import integration_registry

    info = {
        "ZenML version": __version__,
        "Install path": Path(__file__).resolve().parent,
        "Python version": Environment.python_version(),
        "Platform information": Environment.get_system_info(),
        "Environment": get_environment(),
        "Integrations": integration_registry.get_installed_integrations(),
    }
    return "\n".join(
        "{:>10} {}".format(k + ":", str(v).replace("\n", " "))
        for k, v in info.items()
    )