"""
Created on Apr 2, 2012
"""
import collections.abc
import copy
from datetime import datetime, timezone
from inspect import isgenerator
import json
import logging
import os
import platform
import re
import sys
import time
import traceback
from os import PathLike
from typing import Any, Protocol, TypeVar
from xml.etree import ElementTree
import nornir_pools
import nornir_shared.misc
import nornir_shared.prettyoutput as prettyoutput
import nornir_buildmanager.no_delete as _no_delete_mod
import nornir_shared.reflection
from nornir_shared.mqtt_telemetry import publish_run_event
from nornir_shared.tasktimer import TaskTimer
from . import argparsexml
from .pipeline_exceptions import *
from nornir_buildmanager.exceptions import (
NornirMissingDependencyException,
NornirRethrownException,
)
from nornir_buildmanager.pipelinemanager_iterate_filters import (
resolve_iterate_candidates,
sort_iterate_candidates,
)
from nornir_buildmanager.volumemanager import (
XElementWrapper,
XResourceElementWrapper,
XContainerElementWrapper,
VolumeManager,
)
T = TypeVar('T', covariant=True)
# Nested Iterate→PythonCall stages that finish faster than this do not emit
# dashboard/console "Stage … completed" status (skip-heavy Mapping walks).
_NESTED_STAGE_STATUS_MIN_SEC = 1.0
# Per-iterate node dumps (``Iterate: Mapping``, ``MappingNodeObj = …``). Off
# unless ``-verbose`` or this env is set so ``-debug`` logs stay readable.
_PIPELINE_PROGRESS_ENV = "NORNIR_LOG_PIPELINE_PROGRESS"
def _env_flag_enabled(name: str) -> bool:
"""True when *name* is a conventional truthy environment flag."""
return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
def _should_log_pipeline_progress(argset: "ArgumentSet | None" = None) -> bool:
"""True when per-iterate pipeline progress should be written to the log."""
if _env_flag_enabled(_PIPELINE_PROGRESS_ENV):
return True
if argset is None:
return False
return bool(argset.Arguments.get("verbose"))
[docs]
class SupportsRead(Protocol[T]):
[docs]
def read(self, size: int = -1) -> bytes: ...
[docs]
def read_data(obj: SupportsRead) -> bytes:
return obj.read()
# import xml.etree
[docs]
class ArgumentSet:
"""Collection of arguments from each source"""
@property
def Arguments(self):
return self._Arguments
@property
def Attribs(self):
return self._Attribs
@property
def Parameters(self):
return self._Parameters
@property
def Variables(self):
return self._Variables
def __init__(self, PipelineName: str | None = None):
self._Arguments = {}
self._Attribs = {}
self._Parameters = {}
self._Variables = {}
self.PipelineName = PipelineName
[docs]
def SubstituteStringVariables(self, xpath: str) -> str:
"""Replace all instances of # in a string with the variable names"""
iStart = xpath.find("#")
while iStart >= 0:
xpath = self.ReplaceVariable(xpath, iStart)
iStart = xpath.find("#")
return xpath
[docs]
def TryGetValueForKey(self, key):
if key in self.Arguments:
return self.Arguments[key]
if key in self.Variables:
return self.Variables[key]
raise KeyError(str(key) + " not found")
[docs]
def ReplaceVariable(self, xpath: str, iStart: int):
"""Replace # variable names in an xpath with variable string values"""
# Find the next # if it exists
iStart += 1 # Skip the # symbol
iEndVar = xpath[iStart + 1:].find('#')
if iEndVar < 0:
# OK, just check to the end of the string
iEndVar = len(xpath)
else:
iEndVar = iStart + iEndVar + 1
while iEndVar > iStart:
keyCheck = xpath[iStart:iEndVar]
try:
value = self.TryGetValueForKey(keyCheck)
xpath = xpath[:iStart - 1] + str(value) + xpath[iEndVar:]
return xpath
except KeyError:
pass
iEndVar -= 1
# logger = logging.getLogger(__name__ + ".ReplaceVariable")
# logger.error("nornir_buildmanager XPath variable not defined.\nXPath: " + xpath)
prettyoutput.LogErr("nornir_buildmanager XPath variable not defined.\nXPath: " + xpath)
sys.exit()
[docs]
def TryGetSubstituteObject(self, val: str):
"""Place an object directly into a dictionary.
:param str val: The key to lookup
:return: Tuple (bool, value) Returns true on success with the value or None for failure. Not if the value is found and the value is None then (true, None) is returned."""
if val is None:
return False, None
if len(val) == 0:
return False, None
if val[0] == '#':
if val[1:].find('#') >= 0:
# If there are two '#' signs in the string it is a string not an object
return False, None
# Find the existing entry in the dargs
key = val[1:]
try:
# If no object found then return None
return True, self.TryGetValueForKey(key)
except KeyError as e:
return False, None
return False, None
[docs]
def AddArguments(self, args):
"""Add arguments from the command line"""
if isinstance(args, dict):
self._Arguments.update(args)
else:
self._Arguments.update(args.__dict__)
[docs]
def KeyWordArgs(self):
kwargs = {}
kwargs.update(self.Variables)
kwargs.update(self.Attribs)
kwargs['Parameters'] = self.Parameters
return kwargs
[docs]
def AddAttributes(self, Node: ElementTree.Element):
"""Add attributes from an element node to the dargs dictionary.
Parameter values preceeded by a '#' are keys to existing
entries in the dictionary whose values are copied to the
entry for the attribute"""
for key in Node.attrib:
if key in self.Attribs:
raise PipelineError(PipelineNode=Node,
message="%s attribute already present in arguments. Remove duplicate use from pipelines.xml" % key)
val = Node.attrib[key]
if len(val) == 0:
self.Attribs[key] = val
continue
(found, subObj) = self.TryGetSubstituteObject(val)
if found:
self.Attribs[key] = subObj
continue
val = self.SubstituteStringVariables(val)
self.Attribs[key] = val
try:
self.Attribs[key] = int(val)
continue
except ValueError:
pass
try:
self.Attribs[key] = float(val)
continue
except ValueError:
pass
[docs]
def RemoveAttributes(self, Node: ElementTree.Element):
"""Remove attributes present in the node from the attrib dictionary"""
for key in Node.attrib:
if key in self.Attribs:
val = Node.attrib[key]
# If we assign a variable to an attribute with the same name to ourselves do not overwrite
if val[0] == '#':
if val[1:] == key:
continue
del self.Attribs[key]
[docs]
def ClearAttributes(self):
self.Attribs.clear()
[docs]
def AddParameters(self, Node: ElementTree.Element, dargsKeyname=None):
"""Add entries from a dictionary node to the dargs dictionary.
Parameter values preceeded by a '#' are keys to existing
entries in the dictionary whose values are copied to the
entry for the attribute.
If dargsKeyname is none all parameters are added directly
to dargs dictionary. Otherwise they are added as a dictionary
under a key=dargsKeyname"""
ParamNodes = Node.findall('Parameters')
NewParameters = {}
for PN in ParamNodes:
for entryNode in PN:
if entryNode.tag == 'Entry':
name = entryNode.attrib['Name']
val = entryNode.attrib.get('Value', '')
(found, subObj) = self.TryGetSubstituteObject(val)
if found:
NewParameters[name] = subObj
continue
val = self.SubstituteStringVariables(val)
if len(val) == 0:
NewParameters[name] = val
continue
NewParameters[name] = val
try:
NewParameters[name] = int(val)
continue
except ValueError:
pass
try:
NewParameters[name] = float(val)
continue
except ValueError:
pass
if dargsKeyname is None:
self.Parameters.update(NewParameters)
else:
self.Parameters[dargsKeyname] = NewParameters
[docs]
def RemoveParameters(self, Node: ElementTree.Element):
"""
:param Node:
:return:
Remove entries from a dargs dictionary.
"""
# print "Remove Parameters"
ParamNodes = Node.findall('Parameters')
for PN in ParamNodes:
for entryNode in PN:
if entryNode.tag == 'Entry':
name = entryNode.attrib['Name']
if 'Value' not in entryNode.attrib:
continue
val = entryNode.attrib['Value']
if name in list(self.Parameters.keys()):
if val[0] == '#':
if val[1:] == name:
continue
# print "Removing " + name
del self.Parameters[name]
[docs]
def ClearParameters(self):
self.Parameters.clear()
[docs]
def AddVariable(self, key, value):
self.Variables[key] = value
[docs]
def RemoveVariable(self, key):
del self.Variables[key]
[docs]
class PipelineManager:
"""Responsible for the execution of a pipeline specified in an XML file following the buildscript.xsd specification"""
logger = logging.getLogger('PipelineManager')
_description: str | None = None
_help: str | None = None
_epilog: str | None = None
def __init__(self, pipelinesRoot: ElementTree.ElementTree, pipelineData):
self.VolumeTree = None
self.PipelineData = pipelineData
self.defaultArgs = dict()
self.PipelineRoot = pipelinesRoot
self._StageTimer: TaskTimer | None = None
self._PipelineName: str | None = None
self._VolumePath: str | None = None
self._iterate_depth: int = 0
self._description = pipelineData.attrib['Description'] if 'Description' in pipelineData.attrib else None
self._help = pipelineData.attrib['Help'] if 'Help' in pipelineData.attrib else None
self._epilog = pipelineData.attrib['Epilog'] if 'Epilog' in pipelineData.attrib else None
@property
def Description(self) -> str | None:
if hasattr(self, '_description'):
return self._description
return None
@property
def Help(self) -> str | None:
if hasattr(self, '_help'):
return self._help
return None
@property
def Epilog(self) -> str | None:
if hasattr(self, '_epilog'):
return self._epilog
return None
[docs]
@classmethod
def ToElementString(cls, element: ElementTree.Element) -> str:
if element.tag == 'Iterate':
return "Iterate: " + element.attrib['XPath']
outStr = ""
strList = [s.decode('Utf-8') for s in ElementTree.tostringlist(element)]
for s in strList:
outStr = outStr + " " + s
if s == '>':
break
return outStr
@staticmethod
def _StageVolumeLabel(volume_elem: XElementWrapper) -> str:
"""Return a filesystem path or descriptive label for stage timing keys."""
if isinstance(volume_elem, XResourceElementWrapper):
return volume_elem.FullPath
return str(volume_elem)
@staticmethod
def _ElementTelemetryFields(
element: XElementWrapper,
pipeline_node: ElementTree.Element | None = None) -> dict[str, Any]:
"""Return MQTT event fields describing *element* for dashboard progress.
``pipeline_node`` is accepted for call-site symmetry but does not override
the descriptive label derived from the volume element type.
"""
del pipeline_node # Label comes from the element, not VariableName.
fields: dict[str, Any] = {}
cls_name = type(element).__name__
name = getattr(element, "Name", None)
number = getattr(element, "Number", None)
if cls_name == "SectionNode" and number is not None:
fields["section"] = number
padded = name if name is not None else f"{int(number):04d}"
fields["label"] = f"section_node - {padded}"
elif cls_name == "ChannelNode":
if name is not None:
fields["element"] = name
fields["label"] = f"ChannelNode - {name}"
elif cls_name == "FilterNode":
if name is not None:
fields["element"] = name
fields["label"] = f"filter node - {name}"
elif cls_name == "MappingNode":
mapping_label = str(element)
fields["element"] = mapping_label
control = getattr(element, "Control", None)
if control is not None:
fields["section"] = control
fields["label"] = f"MappingNode - {mapping_label}"
else:
if name is not None:
fields["element"] = name
if number is not None:
fields["section"] = number
fields["label"] = f"{cls_name} - {name if name is not None else element}"
return fields
[docs]
@classmethod
def PrintPipelineEnumeration(cls, PipelineXML: str | ElementTree.ElementTree):
PipelineXML = cls.LoadPipelineXML(PipelineXML)
cls.logger.info("Enumerating available pipelines")
prettyoutput.Log("Enumerating available pipelines")
for pipeline in PipelineXML.getroot() or []: # type: ignore[union-attr]
cls.logger.info(' ' + pipeline.attrib.get('Name', ""))
prettyoutput.Log(' ' + pipeline.attrib.get('Name', ""))
prettyoutput.Log(' ' + pipeline.attrib.get('Description', "") + '\n')
[docs]
@classmethod
def LoadPipelineXML(cls, PipelineXML: str | ElementTree.ElementTree | bytes | PathLike[str] | PathLike[bytes] |
SupportsRead[bytes] | SupportsRead[str]) -> ElementTree.ElementTree:
# Python 3 switched to unicode always so the encoding should not be necessary for non-english character sets
if int(platform.python_version_tuple()[0]) < 3:
if isinstance(PipelineXML, str):
PipelineXML = PipelineXML.encode(sys.getdefaultencoding())
if isinstance(PipelineXML, ElementTree.ElementTree):
return PipelineXML
elif isinstance(PipelineXML, ElementTree.Element):
return ElementTree.ElementTree(PipelineXML)
elif isinstance(PipelineXML, str):
try:
return ElementTree.parse(PipelineXML) # type: ignore[return-value]
except FileNotFoundError:
PipelineManager.logger.critical("Provided pipeline filename does not exist: " + PipelineXML)
prettyoutput.LogErr("Provided pipeline filename does not exist: " + PipelineXML)
sys.exit()
elif isinstance(PipelineXML, bytes):
str_xml = PipelineXML.decode('utf-8')
return ElementTree.ElementTree(ElementTree.fromstring(str_xml)) # type: ignore[return-value]
raise Exception("Invalid argument: " + str(PipelineXML))
[docs]
@classmethod
def ListPipelines(cls, pipeline_tree: ElementTree.ElementTree | ElementTree.Element) -> list[str]:
assert (isinstance(pipeline_tree, ElementTree.ElementTree) or isinstance(pipeline_tree, ElementTree.Element))
PipelineNodes = pipeline_tree.findall("Pipeline")
PipelineNames = [p.attrib['Name'] for p in PipelineNodes]
return sorted(PipelineNames)
[docs]
@classmethod
def Load(cls, PipelineXml: str | ElementTree.ElementTree, PipelineName: str | None = None):
# PipelineData = Pipelines.CreateFromDOM(XMLDoc)
# SelectedPipeline = None
XMLDoc = cls.LoadPipelineXML(PipelineXml)
if PipelineName is None:
PipelineManager.logger.warning("No pipeline name specified.")
prettyoutput.Log("No pipeline name specified")
cls.PrintPipelineEnumeration(XMLDoc)
return None
else:
SelectedPipeline = XMLDoc.find("Pipeline[@Name='" + PipelineName + "']")
if SelectedPipeline is None:
PipelineManager.logger.critical("No pipeline found named " + PipelineName)
prettyoutput.LogErr("No pipeline found named " + PipelineName)
cls.PrintPipelineEnumeration(XMLDoc)
return None
return PipelineManager(pipelinesRoot=XMLDoc.getroot(), pipelineData=SelectedPipeline) # type: ignore[arg-type]
[docs]
@classmethod
def RunPipeline(cls, PipelineXmlFile: str | ElementTree.ElementTree, PipelineName: str, args,
volume_tree=None):
# PipelineData = Pipelines.CreateFromDOM(XMLDoc)
Pipeline = cls.Load(PipelineXmlFile, PipelineName)
return Pipeline.Execute(args, volume_tree=volume_tree) # type: ignore[union-attr]
[docs]
def GetArgParser(self, parser=None, IncludeGlobals: bool = True):
"""Create the complete argument parser for the pipeline
:param parser:
:param bool IncludeGlobals: Arguments common to all pipelines are included if this flag is set to True. True by default. False is used to create documentation
"""
if IncludeGlobals:
parser = argparsexml.CreateOrExtendParserForArguments(self.PipelineRoot.findall('Arguments/Argument'),
parser)
parser = argparsexml.CreateOrExtendParserForArguments(self.PipelineData.findall('Arguments/Argument'), parser)
PipelineManager._AddParserDescription(self.PipelineData, parser)
return parser
@classmethod
def _AddParserDescription(cls, PipelineNode, parser):
if PipelineNode is None:
return
# print str(PipelineNode.attrib)
# parser.prog = PipelineNode.attrib['Name'];
if 'Help' in PipelineNode.attrib:
parser.help = PipelineNode.attrib['Help']
if 'Description' in PipelineNode.attrib:
parser.description = PipelineNode.attrib['Description']
if 'Epilog' in PipelineNode.attrib:
parser.epilog = PipelineNode.attrib['Epilog']
@classmethod
def __extractXPathFromNode(cls, PipelineNode, ArgSet):
xpath = PipelineNode.attrib['XPath']
xpath = ArgSet.SubstituteStringVariables(xpath)
return xpath
[docs]
@classmethod
def GetSearchRoot(cls, VolumeElem: XElementWrapper, PipelineNode, ArgSet):
RootIterNodeName = PipelineNode.get('Root', None)
RootForSearch = VolumeElem
if RootIterNodeName is not None:
if RootIterNodeName not in ArgSet.Variables:
raise PipelineSearchRootNotFound(argname=RootIterNodeName, PipelineNode=PipelineNode,
VolumeElem=VolumeElem)
RootForSearch = ArgSet.Variables[RootIterNodeName]
return RootForSearch
IndentLevel = 0
[docs]
def Execute(self, args, volume_tree=None):
"""This executes the loaded pipeline on the specified volume of data.
parser is an instance of the argparser class which should be
extended with any pipeline specific arguments args are the parameters from the command line"""
# DOM = self.PipelineData.toDOM()
# PipelineElement = DOM.firstChild
ArgSet = ArgumentSet()
PipelineElement = self.PipelineData
prettyoutput.Log("Adding pipeline arguments")
# parser = self.GetArgParser(parser)
# (args, unused) = parser.parse_known_args(passedArgs)
ArgSet.AddArguments(args)
ArgSet.AddParameters(PipelineElement)
# Load the Volume.XML file in the output directory
volume_xml_existed = True
if volume_tree is not None:
self.VolumeTree = volume_tree
else:
volume_data_xml = os.path.join(args.volumepath, "VolumeData.xml")
legacy_volume_xml = os.path.join(args.volumepath, "Volume.xml")
volume_xml_existed = os.path.exists(volume_data_xml) or os.path.exists(legacy_volume_xml)
self.VolumeTree = VolumeManager.Load(args.volumepath, Create=True)
if self.VolumeTree is None:
PipelineManager.logger.critical("Could not load or create volume.xml " + args.outputpath)
prettyoutput.LogErr("Could not load or create volume.xml " + args.outputpath)
sys.exit()
self._StageTimer = TaskTimer()
self._PipelineName = getattr(args, 'PipelineName', None) or self.PipelineData.get('Name', 'unknown')
self._VolumePath = args.volumepath
# Fail fast when Create=True invented an empty volume for a non-import pipeline.
# Alignment/assemble stages select Block/Section and otherwise skip silently.
if volume_tree is None:
block_count = len(list(self.VolumeTree.findall('Block')))
pipeline_name = self._PipelineName or ""
is_import_pipeline = pipeline_name.startswith("Import") or pipeline_name.startswith("Adopt")
if (not volume_xml_existed or block_count == 0) and not is_import_pipeline:
err = (
f"Volume at {args.volumepath} has no usable data "
f"(VolumeData.xml existed={volume_xml_existed}, Block count={block_count}). "
f"Pipeline '{pipeline_name}' cannot run — every Block/Section select will skip. "
f"Set launch input nornirVolumesRoot to a host path that contains a real "
f"TEM volume with VolumeData.xml after Import/Prune/Mosaic, then set nornirVolumeName."
)
PipelineManager.logger.critical(err)
prettyoutput.LogErr(err)
if getattr(args, 'debug', False):
raise RuntimeError(err)
sys.exit(2)
# dargs = copy.deepcopy(defaultDargs)
no_delete = getattr(args, 'no_delete', False)
_no_delete_mod.set_no_delete(bool(no_delete))
try:
self.ExecuteChildPipelines(ArgSet, self.VolumeTree, PipelineElement)
finally:
_no_delete_mod.set_no_delete(False)
self._WriteStageTimings()
nornir_pools.ReleaseStagePools()
return self.VolumeTree
def _WriteStageTimings(self) -> None:
"""Append per-stage timing records for this pipeline execute to StageTimings.json."""
if self._StageTimer is None or self._VolumePath is None:
return
stages = [
{"stage": stage_name, "seconds": elapsed}
for stage_name, elapsed in self._StageTimer.ElapsedTimes.items()
]
if not stages:
return
record = {
"pipeline": self._PipelineName or "unknown",
"utc_timestamp": datetime.now(timezone.utc).isoformat(),
"stages": stages,
"total_seconds": sum(stage["seconds"] for stage in stages),
}
output_path = os.path.join(self._VolumePath, "StageTimings.json")
existing: list[dict] = []
if os.path.exists(output_path):
try:
with open(output_path, "r", encoding="utf-8") as input_file:
loaded = json.load(input_file)
if isinstance(loaded, list):
existing = loaded
else:
PipelineManager.logger.warning(
"StageTimings.json is not a list; starting a fresh timing log at %s",
output_path)
except (OSError, UnicodeError, json.JSONDecodeError) as e:
# Called from Execute's finally; a truncated file from a killed run
# must not mask the original exception (#139).
PipelineManager.logger.warning(
"Ignoring unreadable StageTimings.json at %s (%s); starting a fresh timing log",
output_path, e)
existing.append(record)
with open(output_path, "w", encoding="utf-8") as output_file:
json.dump(existing, output_file, indent=2)
[docs]
def ExecuteChildPipelines(self, ArgSet, VolumeElem: XElementWrapper,
PipelineNode):
"""Run all of the child pipeline elements on the volume element"""
if PipelineNode.tag != "Iterate" or _should_log_pipeline_progress(ArgSet):
PipelineManager.logger.info(PipelineManager.ToElementString(PipelineNode))
PipelinesRun = 0
try:
self.AddPipelineNodeVariable(PipelineNode, VolumeElem, ArgSet)
for ChildNode in PipelineNode:
try:
prettyoutput.IncreaseIndent()
self.ProcessStageElement(VolumeElem, ChildNode, ArgSet)
PipelinesRun += 1
except NornirMissingDependencyException as e:
# This means builds cannot succeed. We should clearly warn the user and stop so the cause of failure is clear
PipelineManager.logger.error(e.message)
prettyoutput.LogErr(e.message)
raise NornirRethrownException() from e
except PipelineSelectFailed as e:
if ArgSet.Arguments["debug"]:
PipelineManager.logger.info(str(e))
PipelineManager.logger.info("Select statement did not match. Skipping to next iteration\n")
break
except PipelineSearchFailed as e:
PipelineManager.logger.debug(str(e))
PipelineManager.logger.info("Search statement did not match. Skipping to next iteration\n")
break
except PipelineListIntersectionFailed as e:
PipelineManager.logger.info(
"Node attribute was not in the list of desired values. Skipping to next iteration.\n" + str(e.message)) # type: ignore[operator]
break
except PipelineRegExSearchFailed as e:
PipelineManager.logger.info(
f"Regular expression did not match. regex {e.regex} != {e.attribValue}. Skipping to next iteration.\n" + str(
e.attribValue))
break
except PipelineError as e:
errStr = "Unexpected error, exiting pipeline\n" + str(e.message)
PipelineManager.logger.error(errStr)
prettyoutput.LogErr(errStr)
raise NornirRethrownException() from e
finally:
prettyoutput.DecreaseIndent()
finally:
self.RemovePipelineNodeVariable(ArgSet, PipelineNode)
# To prevent later calls from being able to access variables from earlier steps be sure to remove the variable from the dargs
return PipelinesRun
[docs]
def ProcessStageElement(self, VolumeElem: XElementWrapper, PipelineNode,
ArgSet=None):
# outStr = PipelineManager.ToElementString(PipelineNode)
# prettyoutput.CurseString('Section', outStr)
# prettyoutput.Log("Processing Stage Element: " + outStr)
# Copy dargs so we do not modify what the parent passed us
# dargs = copy.copy(dargs)
if isinstance(VolumeElem, XResourceElementWrapper):
prettyoutput.CurseString("Meta", VolumeElem.FullPath)
if PipelineNode.tag == 'Select':
self.ProcessSelectNode(ArgSet, VolumeElem, PipelineNode)
elif PipelineNode.tag == 'Iterate':
self.ProcessIterateNode(ArgSet, VolumeElem, PipelineNode)
elif PipelineNode.tag == 'RequireSetMembership':
self.RequireSetMembership(ArgSet, VolumeElem, PipelineNode)
elif PipelineNode.tag == 'RequireMatch':
self.ProcessRequireMatchNode(ArgSet, VolumeElem, PipelineNode)
elif PipelineNode.tag == 'PythonCall':
self.ProcessPythonCall(ArgSet, VolumeElem, PipelineNode)
elif PipelineNode.tag == 'Arguments':
pass
else:
raise Exception("Unexpected element name in Pipeline.XML: " + PipelineNode.tag)
[docs]
@staticmethod
def RequireSetMembership(ArgSet, VolumeElem: XElementWrapper, PipelineNode):
"""If the attribute value is not present in the provided list the element is skipped.
If the provided list is none we do not skip."""
RootForMatch = PipelineManager.GetSearchRoot(VolumeElem, PipelineNode, ArgSet)
AttribName = PipelineNode.attrib.get("Attribute", "Name")
AttribName = ArgSet.SubstituteStringVariables(AttribName)
listVariable = PipelineNode.attrib.get("List", None)
if listVariable is None:
raise PipelineError(VolumeElem=VolumeElem,
PipelineNode=PipelineNode,
message="List attribute missing on <RequireSetMembership> node")
(found, listOfValid) = ArgSet.TryGetSubstituteObject(listVariable)
if not found:
# No set to compare with. We allow it.
return
elif listOfValid is None:
# No set to compare with. We allow it.
return
Attrib = getattr(RootForMatch, AttribName, None)
if Attrib is None:
raise PipelineArgumentNotFound(VolumeElem=VolumeElem,
PipelineNode=PipelineNode,
argname=AttribName)
if Attrib not in listOfValid:
raise PipelineListIntersectionFailed(VolumeElem=VolumeElem, PipelineNode=PipelineNode,
listOfValid=listOfValid, attribValue=Attrib)
return
[docs]
@staticmethod
def ProcessRequireMatchNode(ArgSet, VolumeElem: XElementWrapper, PipelineNode):
"""If the regular expression does not match the attribute an exception is raised.
This skips the current iteration of an enclosing <iterate> element"""
RootForMatch = PipelineManager.GetSearchRoot(VolumeElem, PipelineNode, ArgSet)
AttribName = PipelineNode.attrib.get("Attribute", "Name")
AttribName = ArgSet.SubstituteStringVariables(AttribName)
RegExStr = PipelineNode.attrib.get("RegEx", None)
RegExStr = ArgSet.SubstituteStringVariables(RegExStr)
if RegExStr is None:
raise PipelineArgumentNotFound(VolumeElem=VolumeElem,
PipelineNode=PipelineNode,
argname="RegEx",
message="Match node missing RegEx attribute")
Attrib = RootForMatch.attrib.get(AttribName, None)
if Attrib is None:
raise PipelineArgumentNotFound(VolumeElem=VolumeElem,
PipelineNode=PipelineNode,
argname=AttribName)
if RegExStr == '*':
return
match = re.match(RegExStr, Attrib)
if match is None:
raise PipelineRegExSearchFailed(VolumeElem=VolumeElem, PipelineNode=PipelineNode, regex=RegExStr,
attribValue=Attrib)
return
@staticmethod
def _ElementNeedsValidation(element: XElementWrapper) -> bool:
"""Return whether an element supports and requests validation."""
try:
return bool(element.NeedsValidation)
except (NotImplementedError, AttributeError):
return False
[docs]
def ProcessSelectNode(self, ArgSet, VolumeElem: XElementWrapper, PipelineNode):
xpath = PipelineManager.__extractXPathFromNode(PipelineNode, ArgSet)
RootForSearch = PipelineManager.GetSearchRoot(VolumeElem, PipelineNode, ArgSet)
SelectedVolumeElem = None
while SelectedVolumeElem is None:
SelectedVolumeElem = RootForSearch.find(xpath)
if SelectedVolumeElem is None:
raise PipelineSelectFailed(PipelineNode=PipelineNode, VolumeElem=RootForSearch, xpath=xpath)
# Containers will be tested at load time. The load linked element code checks containers
if not isinstance(SelectedVolumeElem,
XContainerElementWrapper):
if PipelineManager._ElementNeedsValidation(SelectedVolumeElem):
(IsValid, Reason) = SelectedVolumeElem.IsValid()
if not IsValid:
# Check if the node is locked, otherwise clean it and look for another node
if 'Locked' in SelectedVolumeElem.attrib:
if SelectedVolumeElem.Locked:
PipelineManager.logger.info(
"Did not clean locked element {0}\n".format(SelectedVolumeElem.FullPath))
break
if _no_delete_mod.is_no_delete():
# Under no-delete, keep the invalid node bound so the
# pipeline can still run against it. Nulling and
# re-searching would loop forever because Clean() is
# suppressed and the invalid element stays in the tree.
PipelineManager.logger.info(
"NO-DELETE: Would clean invalid element %s (%s); keeping for pipeline.",
getattr(SelectedVolumeElem, 'FullPath', str(SelectedVolumeElem)),
Reason,
)
break
SelectedVolumeElem.Clean(Reason)
PipelineManager._SaveNodes(SelectedVolumeElem.Parent)
SelectedVolumeElem = None
if SelectedVolumeElem is not None:
self.AddPipelineNodeVariable(PipelineNode, SelectedVolumeElem, ArgSet)
[docs]
def ProcessIterateNode(self, ArgSet, VolumeElem: XElementWrapper, PipelineNode):
xpath = PipelineManager.__extractXPathFromNode(PipelineNode, ArgSet)
RootForSearch = PipelineManager.GetSearchRoot(VolumeElem, PipelineNode, ArgSet)
candidates = resolve_iterate_candidates(
RootForSearch, xpath, PipelineNode, ArgSet, VolumeElem,
PipelineManager.GetSearchRoot)
sort_attribute = PipelineNode.attrib.get("SortAttribute")
if sort_attribute:
candidates = sort_iterate_candidates(candidates, sort_attribute)
validate = True
if 'Validate' in PipelineNode.attrib:
validate_value = PipelineNode.attrib.get('Validate', 'True').lower()
validate = validate_value == 'true' or validate_value == '1' or validate_value == 'y' or validate_value == 'yes'
# Make sure downstream activities do not corrupt the dictionary for the caller
CopiedArgSet = copy.copy(ArgSet)
variable_name = PipelineNode.attrib.get('VariableName', 'Iterate')
track_id = f"iterate:{variable_name}"
# Do not materialize candidates for len(); MQTT omits None totals
# (publish_run_event) so streaming can start before the full set is known (#140).
depth = self._iterate_depth
self._iterate_depth += 1
NumProcessed = 0
save_parent = set()
progress_current = 0
try:
publish_run_event(
"iterate_progress",
current=0,
total=None,
depth=depth,
track_id=track_id,
label=variable_name)
for VolumeElemChild in candidates:
if validate and PipelineManager._ElementNeedsValidation(VolumeElemChild):
(cleaned, reason) = VolumeElemChild.CleanIfInvalid()
if cleaned:
prettyoutput.Log(f"Cleaned invalid element during search: {VolumeElemChild}\nReason: {reason}")
save_parent.add(VolumeElemChild.Parent)
continue
NumProcessed += self.ExecuteChildPipelines(CopiedArgSet, VolumeElemChild, PipelineNode)
progress_current += 1
tele = PipelineManager._ElementTelemetryFields(VolumeElemChild, PipelineNode)
tele.pop("label", None)
publish_run_event(
"iterate_progress",
current=progress_current,
total=None,
depth=depth,
track_id=track_id,
label=variable_name,
**tele)
finally:
self._iterate_depth -= 1
publish_run_event(
"iterate_progress_complete",
track_id=track_id,
total=progress_current)
for parent in save_parent:
PipelineManager._SaveNodes(parent)
if NumProcessed == 0:
raise PipelineSearchFailed(PipelineNode=PipelineNode, VolumeElem=RootForSearch, xpath=xpath)
@classmethod
def _SaveNodes(cls, NodesToSave):
if NodesToSave is None:
return
# Stage functions may return True/False to indicate work without naming a
# node to persist (#135). Bool must be rejected before the Element /
# Iterable branches — otherwise VolumeManager.Save raises ValueError.
if isinstance(NodesToSave, bool):
return
# ElementTree.Element (and XElementWrapper) are Iterable over children.
# Saving must target the returned node itself, not walk its children.
if isinstance(NodesToSave, ElementTree.Element):
VolumeManager.Save(NodesToSave)
return
if isinstance(NodesToSave, collections.abc.Iterable) or isgenerator(NodesToSave):
for node in NodesToSave:
if node is None:
continue
VolumeManager.Save(node)
return
VolumeManager.Save(NodesToSave)
[docs]
def ProcessPythonCall(self, ArgSet, VolumeElem: XElementWrapper, PipelineNode):
# Try to find a stage for the element we encounter in the pipeline.
# PipelineModule = 'nornir_buildmanager.operations' # This should match the default in the xsd file, but pyxb doesn't seem to emit the default valuef
PipelineModule = PipelineNode.get("Module", "nornir_buildmanager.operations")
PipelineFunction = PipelineNode.get('Function', PipelineNode.tag)
stageFunc = nornir_shared.reflection.get_module_class(str(PipelineModule), str(PipelineFunction))
if ArgSet.Arguments['verbose']:
prettyoutput.Log("CALL " + str(PipelineModule) + "." + str(PipelineFunction))
# PipelineManager.logger.info("CALL " + str(PipelineModule) + "." + str(PipelineFunction))
if stageFunc is None:
errorStr = "Stage implementation not found: " + str(PipelineModule) + "." + str(PipelineFunction)
PipelineManager.logger.error(errorStr + ElementTree.tostring(PipelineNode, encoding='utf-8'))
raise PipelineError(VolumeElem=VolumeElem, PipelineNode=PipelineNode, message=errorStr)
else:
stage_label = f"{PipelineModule}.{PipelineFunction}"
# Nested Iterate PythonCalls (e.g. StosGridRefine per Mapping) often
# skip in milliseconds; keep stage_start/end events for current_stage
# but do not flood dashboard/console Stage status lines for those.
nested_stage = self._iterate_depth > 0
if not nested_stage:
prettyoutput.CurseString('Stage', stage_label)
# TODO: Update args from the element
# Update dargs with the attributes
ArgSet.AddAttributes(PipelineNode)
ArgSet.AddParameters(PipelineNode)
stage_key = f"{PipelineModule}.{PipelineFunction} @ {PipelineManager._StageVolumeLabel(VolumeElem)}"
tele = PipelineManager._ElementTelemetryFields(VolumeElem, PipelineNode)
publish_run_event(
"stage_start",
module=str(PipelineModule),
function=str(PipelineFunction),
**tele)
stage_started = time.perf_counter()
try:
# PipelineManager.AddAttributes(dargs, PipelineNode)
# Check for parameters under the function node and load them into the dictionary
# PipelineManager.AddParameters(dargs, PipelineNode, dargsKeyname='Parameters')
kwargs = ArgSet.KeyWordArgs()
kwargs["Logger"] = PipelineManager.logger.getChild(PipelineFunction)
# kwargs["CallElement"] = PipelineNode
kwargs["VolumeElement"] = VolumeElem
kwargs["VolumeNode"] = self.VolumeTree
# Add an empty dictionary if no parameters set
if 'Parameters' not in kwargs:
kwargs['Parameters'] = {}
if self._StageTimer is not None:
self._StageTimer.Start(stage_key)
# NodesToSave = None
if not ArgSet.Arguments["debug"]:
try:
NodesToSave = stageFunc(**kwargs)
except Exception as e:
errorStr = '\n' + '-' * 60 + '\n'
errorStr = errorStr + str(PipelineModule) + '.' + str(PipelineFunction) + " Exception\n"
errorStr = errorStr + '-' * 60 + '\n'
errorStr += traceback.format_exc()
errorStr = errorStr + '-' * 60 + '\n'
PipelineManager.logger.error(errorStr)
publish_run_event(
"stage_failed",
module=str(PipelineModule),
function=str(PipelineFunction),
**tele)
raise PipelineError(VolumeElem=VolumeElem,
PipelineNode=PipelineNode,
message=errorStr) from e
else:
# In debug mode we do not want to catch any exceptions
# stage functions can return None,True, or False to indicate they did work.
# if they return false we do not need to run the expensive save operation
print(str(PipelineModule) + '.' + str(PipelineFunction))
NodesToSave = stageFunc(**kwargs)
PipelineManager._SaveNodes(NodesToSave)
publish_run_event(
"stage_end",
module=str(PipelineModule),
function=str(PipelineFunction),
**tele)
finally:
if self._StageTimer is not None:
self._StageTimer.End(stage_key, print_elapsed=False)
ArgSet.ClearAttributes()
ArgSet.ClearParameters()
elapsed = time.perf_counter() - stage_started
if not nested_stage:
prettyoutput.CurseString('Stage', stage_label + " completed")
elif elapsed >= _NESTED_STAGE_STATUS_MIN_SEC:
prettyoutput.CurseString('Stage', stage_label + " completed")
# PipelineManager.RemoveParameters(dargs, PipelineNode)
# PipelineManager.RemoveAttributes(dargs, PipelineNode)
[docs]
@staticmethod
def AddPipelineNodeVariable(PipelineNode, VolumeElem: XElementWrapper, ArgSet):
"""Adds a variable to our dictionary passed to functions"""
if 'VariableName' in PipelineNode.attrib:
key = PipelineNode.attrib['VariableName']
# if key in ArgSet.Variables:
# raise PipelineError(PipelineNode=PipelineNode, VolumeElem=VolumeElem, message=str(key) + " is a duplicate variable name")
ArgSet.AddVariable(key, VolumeElem)
if _should_log_pipeline_progress(ArgSet):
outStr = VolumeElem.ToElementString()
PipelineManager.logger.info(PipelineNode.attrib['VariableName'] + " = " + outStr)
elif PipelineNode.tag == "Select":
raise PipelineError(PipelineNode=PipelineNode, message="VariableName attribute required on Select Element")
[docs]
@staticmethod
def RemovePipelineNodeVariable(ArgSet, PipelineNode):
"""Adds a variable to our dictionary passed to functions"""
if 'VariableName' in PipelineNode.attrib:
del ArgSet.Variables[PipelineNode.attrib['VariableName']]
def _GetVariableName(PipelineNode):
if 'VariableName' in PipelineNode.attrib:
return PipelineNode.attrib['VariableName']
# self.Variables[PipelineNode.attrib['VariableName']] = VolumeElem
# outStr = VolumeElem.ToElementString()
# if(self.Parameters['verbose']):
# prettyoutput.Log(PipelineNode.attrib['VariableName'] + " = " + outStr)
# PipelineManager.logger.info(PipelineNode.attrib['VariableName'] + " = " + outStr)
elif PipelineNode.tag == "Select":
raise PipelineError(PipelineNode=PipelineNode, message="Variable name attribute required on Select Element")
if __name__ == "__main__":
XmlFilename = 'D:\\Buildscript\\Pipelines.xml'
PipelineManager.Load(XmlFilename)