Source code for towhee.dag.utils.callstack

# Copyright 2021 Zilliz. All rights reserved.
# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import inspect
from typing import List, Union
import hashlib

[docs]class Callstack: """ A Callstack object contains the working frames at the moment. Args: ignore (`int`): The number of frames to ignore on the top of the callstack. """
[docs] def __init__(self, ignore: int = 0): self.frames = inspect.stack() # ignore the frame of Callstack.__init__ ignore += 1 if ignore > len(self.frames): raise ValueError(f"ignore = {ignore-1} is out of frame range") del self.frames[0:ignore] self.size = len(self.frames)
[docs] def num_frames(self) -> int: """ Get the number of frames. Returns: (`int`) The size of current stack. """ return self.size
[docs] def find_func(self, func_name: str) -> Union[int, None]: """ Given a function name, find the first-matched and outermost frame from current stack. Args: func_name (`str`): The function name to find. Returns: (`Union[int, None]`) If at least one matching frame exits, return the first-matched frame index. Else, return None. """ for i in range(self.size - 1, -1, -1): if self.frames[i].function == func_name: return i return None
[docs] def hash(self, start: int = None, end: int = None, items: List[str] = None) -> str: """ Get the hash value of the attributes contained in `items` between index `start` and `end` (includes `start`, excludes `end`). Args: start (`int`): The index of the start frame. end (`int`): The index of the end frame. items (`List[str]`): The items to be hashed. Supported items are {filename, lineno, function, code_context, position, lasti}, where code_context denotes the current line of code of the context, position denotes the frame's index of the callstack, lasti denotes the index of last attempted instruction in bytecode. Returns: (`str`) The hash value. Raises: (`IndexError`) If the args [`start`, `end`) is out of the frame range or `end` less than `start`. (`ValueError`) If an item in `items` is not supported, i.e. not one of {filename, lineno, function, code_context, position, lasti}. """ start = start or 0 end = end or self.size if end > self.size or end <= 0 or start >= self.size or start < 0: raise IndexError(f"index range [{start}, {end}) out of frame range" f"[0, {self.size})") if start >= end: raise IndexError(f"end = {end} is less than or equal to start = {start}") full_item = {"filename", "lineno", "function", "code_context", "position", "lasti"} if not set(items).issubset(set(full_item)): invalid_item = set(items) - (set(items) & full_item) raise ValueError(f"{invalid_item} not supported") md5 = hashlib.md5() for i, frame in enumerate(self.frames[start:end]): frame_dict = frame._asdict() frame_dict["position"] = i + start frame_dict["lasti"] = frame_dict["frame"].f_lasti frame_dict["code_context"] = ("".join(frame_dict["code_context"]) if frame_dict["code_context"] else "") for item in items: md5.update(str(frame_dict[item]).encode("utf-8")) return md5.hexdigest()