teflow.utils module

class teflow.utils.AttrDict

基类:OrderedDict

A lightweight extended dictionary class, designed to maintain key order and enable attribute-style access via dot notation for a more intuitive interaction.

示例

>>> d = AttrDict({'key1': 'value1'})
>>> d['key1']
'value1'
>>> d.key1
'value1'

备注

  • Dot access mode only supports accessing existing keys. Setting values using dot notation is not supported.

  • This class may not handle nested dictionaries in dot access mode.

retain(keys, match_order=False)

Retain specified keys in the dictionary, optionally matching order.

参数:
  • keys (Iterable or Sequence) – The keys to be retained. When ‘match_order’ is False, ‘keys’ can be any Iterable (like set, list, or string). When ‘match_order’ is True, ‘keys’ must be a Sequence (like list or tuple) to ensure the order of elements.

  • match_order (bool, optional) – If True, the order of keys in the resulting dictionary will match the order of keys in the ‘keys’ parameter. Default is False.

返回:

A dictionary of the keys that were removed and their values.

返回类型:

dict

抛出:

TypeError – If ‘keys’ is not an Iterable or not a Sequence when ‘match_order’ is True.

示例

>>> d = AttrDict(b=2, g=3, note='somewhat', c=2)
>>> popped = d.retain('abcdefg', match_order=True)
>>> # Equivalent to the above:
>>> # popped = d.retain(['a','b','c','d','e','f','g'], match_order=True)
>>> print(d)
AttrDict([('b', 2), ('c', 2), ('g', 3)])
>>> print(popped)
{'note': 'somewhat'}
class teflow.utils.CfgParser(allow_section_whitespace=True, case_sensitive=True, **kwargs)

基类:ConfigParser

A custom configuration parser class derived from ConfigParser, with enhanced features:

  • Allows for case sensitivity in configuration options.

  • Allows whitespace in section names.

  • Predefined converters for array, and list[_XXX] types.

  • Delimiters: Only the ‘=’ character is accepted for separating keys and values in the configuration file, ensuring a consistent format.

  • Comment Prefixes: The ‘#’ character is used to indicate comments, allowing for clear and concise configuration files.

  • Inline Comment Prefixes: Inline comments are also indicated with the ‘#’ character.

  • Interpolation: Uses ExtendedInterpolation() to provide advanced string formatting within the configuration file, allowing values to reference other values or variables.

  • Prefix Match: Introduce pmatch() method to identify [SectionName.SectionType] type sections.

All these settings are initialized to ensure a consistent and clear configuration file format, while still allowing for advanced features and flexibility. Users can override these settings as per their requirements.

Initialize the configuration parser with custom settings.

参数:
  • case_sensitive (bool, optional) – Whether the parsing should be case sensitive. Default is True.

  • allow_section_whitespace (bool, optional) – Whether to allow whitespace in section names. Default is True.

  • **kwargs – Additional keyword arguments to pass to the ConfigParser.

pmatch(section)

Matches sections by prefix SectionName and retrieves content from [SectionName.SectionType] or [SectionName] formatted sections.

class teflow.utils.ExecWrapper(obj, args=(), opts=())

基类:object

A utility class for dynamically managing the arguments of a callable or a class constructor, allowing for dynamic argument updates and execution.

UNSET

The unset flag for an argument.

参数:
  • obj (Any) – A callable object or class constructor to manage.

  • args (list, optional) – The names of required arguments for the managed object.

  • opts (list, optional) – The names of optional arguments for the managed object.

execute(**kwargs)

Executes the callable or instantiates the class. The arguments provided will override the corresponding stored values temporarily, but this applies only to the current execution. To make permanent changes to the arguments, use the update() method.

参数:

**kwargs (any, optional) – Keyword arguments to be passed to the callable or constructor.

返回:

The result of the callable execution or class instantiation.

返回类型:

Any

抛出:
  • ValueError – If any required argument is missing.

  • RuntimeError – If the execution of the callable or the class instantiation fails.

update(**kwargs)

Updates the values of both required and optional arguments.

参数:

**kwargs (any, optional) – Keyword arguments to update the values of arguments. Note that any invalid arguments passed will be ignored without warning.

class teflow.utils.ListDict

基类:AttrDict

Extends AttrDict to include methods for manipulating list values within the dictionary, such as append(), extend(), merge(), and sum().

append(obj)

Append values from another dictionary to the current instance.

参数:

obj (dict) – Dictionary from which values are to be appended.

抛出:

ValueError – If the object to append is not a dictionary.

extend(obj)

Extend values from another dictionary to the current instance.

参数:

obj (dict) – Dictionary from which values are to be extended.

抛出:

ValueError – If the object to extend is not a dictionary.

classmethod merge(objs, keys=None, toarray=False)

Merge a list of dictionaries to a new instance.

参数:
  • objs (list of dicts) – List of dictionaries to merge.

  • keys (list of str, optional) – Keys to consider for merging. If not provided, uses keys from the first dictionary.

  • toarray (bool, optional) – If True, converts the merged result into a numpy.ndarray.

返回:

Merged dictionary.

返回类型:

ListDict

抛出:

ValueError – If any object in objs is not a dictionary.

示例

>>> d1 = {'a': 1, 'b': 2}
>>> d2 = {'a': 3, 'b': 4}
>>> merged = ListDict.merge([d1, d2])
>>> print(merged)
ListDict([('a', [1, 3]), ('b', [2, 4])])
classmethod sum(objs, start=0, keys=None)

Sum values from a list of dictionaries to a new instance.

参数:
  • objs (list of dicts) – List of dictionaries to sum.

  • start (int or float, optional) – Starting value for sum. Default is 0.

  • keys (list of str, optional) – Keys to consider for summing. If not provided, uses keys from the first dictionary.

返回:

Dictionary with summed values.

返回类型:

ListDict

抛出:

ValueError – If any object in objs is not a dictionary.

示例

>>> d1 = {'a': 1, 'b': 2}
>>> d2 = {'a': 3, 'b': 4}
>>> summed = ListDict.sum([d1, d2])
>>> print(summed)
ListDict([('a', 4), ('b', 6)])
class teflow.utils.Metric(kind='MSE')

基类:object

Metric class provides functionalities for computing various error metrics. This class can be initialized with a specific kind of metric (such as ‘MSE’, ‘RMSE’) and later be called directly to compute the error based on the initialized metric kind. Additionally, each metric kind can also be used directly as a static method.

示例

>>> metric = Metric('MSE')
>>> y = [1, 2, 3]
>>> y2 = [2, 4, 5]
>>> metric(y, y2)
3.0
>>> Metric.MSE(y, y2)
3.0
参数:

kind (str, optional) – Kind of metric to be used. Default is ‘MSE’.

static MAE(y, y2, axis=-1)

Mean Absolute Error:

\[\frac{1}{n} \sum_{i=1}^{n} |y_2[i] - y[i]|\]
static MAPE(y, y2, axis=-1)

Mean Absolute Percentage Error:

\[\frac{1}{n} \sum_{i=1}^{n} \left| \frac{y_2[i] - y[i]}{y[i]} \right|\]
static MSE(y, y2, axis=-1)

Mean-Square Error:

\[\frac{1}{n} \sum_{i=1}^{n} (y_2[i] - y[i])^2\]
static RMSE(y, y2, axis=-1)

Root-Mean-Square error:

\[\sqrt{\frac{1}{n} \sum_{i=1}^{n} (y_2[i] - y[i])^2}\]
static SMAPE(y, y2, axis=-1)

Symmetric Mean Absolute Percentage Error:

\[\frac{2}{n} \sum_{i=1}^{n} \frac{|y_2[i] - y[i]|}{|y_2[i]| + |y[i]|}\]
teflow.utils.get_root_logger(stdout=True, filename=None, mode='a', level=None, fmt='[%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S', file_fmt='%(asctime)s [%(levelname)s @ %(name)s] %(message)s', file_datafmt='%Y-%m-%d %H:%M:%S', *, pkgname=None)

Get root logger object