Source code for lavaflow.utils.config

"""Utility functions for config.
"""

import logging
import re

import yaml

from dotmap import DotMap

logger = logging.getLogger('lavaflow')


# -----------------------------------------------------------------------------

# Custom yaml with inline comments

def remove_inline_comments(config_yaml, prefix='#'):
  """Remove inline comments from config yaml based on comment prefix.

  Args:
    config_yaml (str): config yaml with inline comments
    prefix (str): comment prefix (default: #)

  Returns:
    config_yaml (str): config with inline comments removed
  """

  return re.sub(f' *{prefix}.*\n', '\n', config_yaml)


# Comparisons

def compare_config(a, b, aname='a', bname='b', keys=True, types=True, values=True):
  """Function to compare the key structure and value types of two configs.

  Args:
    a (dict): config
    b (dict): config
    aname (str): config name
    bname (str): config name
    keys (bool): check if keys are not the same at each level of the dictionary
    types (bool): check if types are not the same
    values (bool): check if values are not the same

  Returns:
    d (dict): comparison summary

  Examples:

    >>> a = { 'a': 1, 'b': { 'c': 3, 'd': 4, 'e': 'x' }}
    >>> b = { 'a': 3, 'b': { 'c': 3, 'd': 'y', 'g': 7 }}
    >>> compare_config(a, b)
    ['a.a = 1 != b.a = 3', "a.b.keys() - b.b.keys() = ['e']", "b.b.keys() - a.b.keys() = ['g']", "type(a.b.d) = <class 'int'> != type(b.b.d) = <class 'str'>"]

  """

  def compare(x, y, kx, ky):

    if types and x is not None and y is not None and type(x) != type(y):
      return [f'type({kx}) = {type(x)} != type({ky}) = {type(y)}']

    if isinstance(x, dict):
      issues = []
      xkeys = set(x.keys())
      ykeys = set(y.keys())
      if keys:
        if xkeys - ykeys:
          issues.append(f'{kx}.keys() - {ky}.keys() = {list(xkeys - ykeys)}')
        if ykeys - xkeys:
          issues.append(f'{ky}.keys() - {kx}.keys() = {list(ykeys - xkeys)}')
      for k in xkeys.intersection(ykeys):
        issues.extend(compare(x[k], y[k], f'{kx}.{k}', f'{ky}.{k}'))
      return issues

    if values and x != y:
      return [f'{kx} = {x} != {ky} = {y}']

    return []

  return compare(a, b, aname, bname)