File size: 3,369 Bytes
bcdf9fa |
1 2 3 4 5 6 7 8 9 10 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 |
# Copyright 2024 Bytedance Ltd. and/or its affiliates
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Utilities to check if packages are available.
We assume package availability won't change during runtime.
"""
import importlib.util
from functools import cache
from typing import List, Optional
@cache
def is_megatron_core_available():
try:
mcore_spec = importlib.util.find_spec("megatron.core")
except ModuleNotFoundError:
mcore_spec = None
return mcore_spec is not None
@cache
def is_vllm_available():
try:
vllm_spec = importlib.util.find_spec("vllm")
except ModuleNotFoundError:
vllm_spec = None
return vllm_spec is not None
@cache
def is_sglang_available():
try:
sglang_spec = importlib.util.find_spec("sglang")
except ModuleNotFoundError:
sglang_spec = None
return sglang_spec is not None
def import_external_libs(external_libs=None):
if external_libs is None:
return
if not isinstance(external_libs, List):
external_libs = [external_libs]
import importlib
for external_lib in external_libs:
importlib.import_module(external_lib)
def load_extern_type(file_path: Optional[str], type_name: Optional[str]):
"""Load a external data type based on the file path and type name"""
import importlib.util
import os
if not file_path:
return None
if not os.path.exists(file_path):
raise FileNotFoundError(f"Custom type file '{file_path}' not found.")
spec = importlib.util.spec_from_file_location("custom_module", file_path)
module = importlib.util.module_from_spec(spec)
try:
spec.loader.exec_module(module)
except Exception as e:
raise RuntimeError(f"Error loading module from '{file_path}'") from e
if not hasattr(module, type_name):
raise AttributeError(f"Custom type '{type_name}' not found in '{file_path}'.")
return getattr(module, type_name)
def _get_qualified_name(func):
"""Get full qualified name including module and class (if any)."""
module = func.__module__
qualname = func.__qualname__
return f"{module}.{qualname}"
def deprecated(replacement: str = ""):
"""Decorator to mark APIs as deprecated."""
import warnings
import functools
def decorator(func):
qualified_name = _get_qualified_name(func)
@functools.wraps(func)
def wrapped(*args, **kwargs):
msg = f"Warning: API '{qualified_name}' is deprecated."
if replacement:
msg += f" Please use '{replacement}' instead."
warnings.warn(msg, category=DeprecationWarning, stacklevel=2)
return func(*args, **kwargs)
return wrapped
return decorator |