o }oi@s ddlZddlZddlmZddlZddlZddlmZddl m Z ddl m Z dZ dZ  ddejejeejfd ejejeejffd d Zd ed efddZdejed ejfddZdejd ejefddZdejd ejejfddZddZdddZdS)N)Path)ImageTensor)TarPathzNEMO 2.0zNEMO 1.0overwrite_kwargsdefaultscs>ddfddp itfddt|DS)aConverts a type dictionary class into a tuple of PyTriton Tensor objects. This function takes a class with type hints and converts each typed field into a PyTriton Tensor specification, handling nested list types and mapping Python types to numpy dtypes. Args: typedict_class: A class with type hints that will be converted to Tensor specs overwrite_kwargs: Optional dictionary of kwargs to override default Tensor parameters defaults: Optional dictionary of default values (unused) Returns: tuple: A tuple of PyTriton Tensor objects, one for each typed field in the input class Raises: Exception: If an unsupported type is encountered during type mapping cSsD|turtjS|turtjS|turtjS|turtSt d|)Nz Unknown type ) intnpint32floatfloat32boolbool_strbytes Exception)type_rE/home/ubuntu/.local/lib/python3.10/site-packages/nemo/deploy/utils.py _map_type4sz"typedict2tensor.._map_typecs`d}t|turt|d}|d7}t|tus |d8}|dkr'd|nd}||dS)Nr))r)shapedtype)typing get_originlistget_args)rcountr)rrr_get_tensor_params@sz+typedict2tensor.._get_tensor_paramsc3s.|]\}}tdd|i|VqdS)nameNrr).0r!r)r rrr Js  z"typedict2tensor..)tuplerget_type_hintsitems)typedict_classrrr)r rrrtypedict2tensors    r(pathreturncCs>tj|r t|}nt|}|dr|drtStS)aDetermines the version of a NeMo checkpoint from its file structure. Examines the provided checkpoint path to determine if it follows the NeMo 2.0 or NeMo 1.0 format based on the presence of 'context' and 'weights' directories. Args: path (str): Path to the NeMo checkpoint file or directory Returns: str: Version string - either NEMO2 or NEMO1 constant indicating the checkpoint version contextweights)osr)isdirrrexistsNEMO2NEMO1)r)rrrnemo_checkpoint_versionPs r2str_listcCs"t|dtjf}tj|dS)aConverts a list of strings to a numpy array of UTF-8 encoded bytes. Takes a list of strings and converts it to a numpy array with an additional dimension, then encodes the strings as UTF-8 bytes. Args: str_list (List[str]): List of strings to convert Returns: np.ndarray: Numpy array of UTF-8 encoded bytes with shape (N, 1) where N is the length of the input list .utf-8)r arraynewaxischarencode)r3 str_ndarrayrrrstr_list2numpyhs r:r9cCs.|d}tjj|dd}|jdd}|S)aConverts a numpy array of UTF-8 encoded bytes back to a list of strings. Takes a numpy array of UTF-8 encoded bytes and decodes it back to strings, removing any extra dimensions, and returns the result as a Python list. Args: str_ndarray (np.ndarray): Numpy array of UTF-8 encoded bytes, typically with shape (N, 1) where N is the length of the resulting list Returns: List[str]: List of decoded strings rr4)encodingr)axis)astyper r7decodesqueezetolist)r9rrrstr_ndarray2listys  rA img_ndarraycCsdd|D}|S)aConverts a numpy array of images to a list of PIL Image objects. Takes a numpy array containing one or more images and converts each image to a PIL Image object using Image.fromarray(). Args: img_ndarray (np.ndarray): Numpy array of images, where each image is a 2D or 3D array representing pixel values Returns: List[Image.Image]: List of PIL Image objects created from the input array cSsg|]}t|qSr)r fromarray)r"irrr szndarray2img..r)rBimg_listrrr ndarray2imgsrGcCsxt|tjr |}nt|tjs+t|}|ttj t tj fv}|r+tj |d}|jdkr7|dtjf}||S)aCasts input data to a numpy array with the required dtype. Takes input data that may be a torch.Tensor, numpy array, or other sequence type and converts it to a numpy array with the specified dtype. For string dtypes, the data is encoded as UTF-8 bytes. The output array is ensured to have at least 2 dimensions. Args: data: Input data to cast. Can be a torch.Tensor, numpy array, or sequence type that can be converted to a numpy array. required_dtype: The desired numpy dtype for the output array. Returns: np.ndarray: A numpy array containing the input data cast to the required dtype, with at least 2 dimensions. r4.) isinstancetorchrcpunumpyr ndarrayr5objectobject_rbytes_r7r8ndimr6r=)datarequired_dtype data_is_strrrr cast_outputs     rUcCsFtjs tdtj|kr|gndg}tjj|||d|dS)a#Broadcasts a list of text data to all processes. Args: data (list): List of strings to broadcast. src (int, optional): Source rank. Defaults to 0. group (ProcessGroup, optional): The process group to work on. If None, the default process group will be used. z+Distributed environment is not initialized.N)srcgroupr)rJ distributedis_initialized RuntimeErrorget_rankbroadcast_object_list)rRrVrW object_listrrrbroadcast_lists r^)NN)rN)r-rpathlibrrLr rJPILrpytriton.model_configrnemo.export.tarutilsrr0r1OptionalDictrAnyr(r2ListrMr:rArGrUr^rrrrs.     2