def startswith(self, prefix, *args, **kwargs):
return self.lower().startswith(prefix.lower(), *args, **kwargs)
- def endswith(self, prefix, *args, **kwargs):
- return self.lower().endswith(prefix.lower(), *args, **kwargs)
+ def endswith(self, suffix, *args, **kwargs):
+ return self.lower().endswith(suffix.lower(), *args, **kwargs)
+
+ def replace(self, old, new, *args, **kwargs):
+ return self.lower().replace(old.lower(), new.lower(), *args, **kwargs)
###############################################################################
def __add__(self, other):
return CnfList(super().__add__(other))
+ def add(self, *args, **kwargs):
+ """
+ Add a CNF variable to the list.
+
+ Arguments can either be a single instance of the :py:class:`Cnf`
+ class or a list of arguments to be passed to the constructor of
+ that class. Similar to the :py:func:`add_child` method for a `Cnf`.
+
+ :returns: the instance that was created
+ :rtype: :py:class:`Cnf`
+ """
+ # support passing a Cnf instance
+ if len(args) == 1 and not kwargs:
+ cnf = args[0]
+ assert isinstance(cnf, Cnf), "A Cnf instance is mandatory with one argument"
+ else:
+ cnf = Cnf(*args, **kwargs)
+
+ self.append(cnf)
+ return cnf
+
class BaseCnf:
"""Base class representing a CNF variable with minimal functionality."""
_CHILD_TEMPLATE = "{lineno} {indent}({parent}) {name},{instance}: \"{value}\""
_NEST_INDENT = " "
- def __init__(self, name, value, instance=-1, parent=None,
+ def __init__(self, name, value, instance=0, parent=None,
lineno=None, comment=None):
"""
Create this instance.
# support passing a Cnf instance
if len(args) == 1 and not kwargs:
cnf = args[0]
- assert isinstance(cnf, Cnf), \
- "With one argument, a Cnf instance is mandatory"
+ assert isinstance(cnf, Cnf), "A Cnf instance is mandatory with one argument"
else:
cnf = Cnf(*args, **kwargs)
"""
if renumber:
self.renumber()
- return {"cnf": [x.to_cnfvar_dict() for x in self]}
+ return {"cnf": [x.to_cnf_structure() for x in self]}
def to_cnf_file(self, path, renumber=True, encoding=ENCODING):
"""
class CnfSerializationMixin(BaseCnf):
"""Add serialization support to BaseCnf."""
- def to_cnfvar_dict(self):
+ def to_cnf_structure(self):
"""
Convert this instance to dictionary from the :py:mod:`cnfvar` module.
if self.comment is not None:
d["comment"] = self.comment
if len(self.children) > 0:
- d["children"] = [c.to_cnfvar_dict() for c in self.children]
+ d["children"] = [c.to_cnf_structure() for c in self.children]
return d
def to_json_string(self, renumber=True):
"""Shortcut method for getting the first item with a given instance."""
return self.with_instance(instance).first(default=default)
+ def highest_instance(self):
+ """Shortcut method for getting the next instance in a list of items."""
+ return max([c.instance for c in self]) if len(self) > 0 else -1
+
###############################################################################
# PUBLIC CLASSES
:type: :py:class:`arnied_api.Arnied`
"""
self._driver = backend_driver
- log.debug("Initialized CnfStore with driver `%s`", type(backend_driver))
+ # TODO: implement `self._wait_for_arnied()` which should busy-loop with
+ # the arnied varlink socket and handle "Disconnected" errors, then perhaps
+ # drop the old binary cnf store method from the arnied wrapper
+ log.debug(f"Initialized cnf store with driver `{backend_driver.__name__}`")
def query(self, name=None, instance=None):
"""
This method can be used in child classes to use an alternative scheme,
however for performance reasons the base API class uses the default and
relies on the cnfvar backend to do this job.
+
+ ..todo:: This method compensates for limitations in production code that
+ might end up fixed up there deprecating our patching here.
"""
def _do_commit(self, original_cnfs, arnied_cnfs, fix_problems=False):
:type: :py:class:`CnfBinary`
"""
super().__init__(backend_driver=backend_driver)
- log.debug("Initialized BinaryCnfStore with driver `%s`", type(backend_driver))
+ # We assume that any external events happening to arnied require reinitializing
+ # the cnf store which is bound to the lifespan of a single arnied process.
+ self._call_arnied(arnied_wrapper.verify_running, timeout=self.ARNIED_TIMEOUT)
+ log.debug(f"Initialized binary cnf store with driver `{backend_driver.__name__}`")
def query(self, name=None, instance=None):
"""
cnf.renumber()
log.debug("Committing variables via binaries:\n%s", cnf)
- self._call_arnied(arnied_wrapper.verify_running, timeout=self.ARNIED_TIMEOUT)
try:
self._driver.set_cnf(input_str=str(cnf), fix_problems=fix_problems)
except subprocess.CalledProcessError as ex:
cnf.renumber()
log.debug("Deleting variables via binaries:\n%s", cnf)
- self._call_arnied(arnied_wrapper.verify_running, timeout=self.ARNIED_TIMEOUT)
try:
self._driver.set_cnf(input_str=str(cnf), delete=True, fix_problems=fix_problems)
except subprocess.CalledProcessError as ex:
git://developer.intra2net.com / pyi2ncommon / commitdiff