Source code for edelweissfe.utils.fieldoutput

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#  ---------------------------------------------------------------------
#
#  _____    _      _              _         _____ _____
# | ____|__| | ___| |_      _____(_)___ ___|  ___| ____|
# |  _| / _` |/ _ \ \ \ /\ / / _ \ / __/ __| |_  |  _|
# | |__| (_| |  __/ |\ V  V /  __/ \__ \__ \  _| | |___
# |_____\__,_|\___|_| \_/\_/ \___|_|___/___/_|   |_____|
#
#
#  Unit of Strength of Materials and Structural Analysis
#  University of Innsbruck,
#  2017 - today
#
#  Matthias Neuner matthias.neuner@uibk.ac.at
#
#  This file is part of EdelweissFE.
#
#  This library is free software; you can redistribute it and/or
#  modify it under the terms of the GNU Lesser General Public
#  License as published by the Free Software Foundation; either
#  version 2.1 of the License, or (at your option) any later version.
#
#  The full text of the license can be found in the file LICENSE.md at
#  the top level directory of EdelweissFE.
#  ---------------------------------------------------------------------

"""
FieldOutputs store all kind of analysis results,
and are defined via the keyword ``*fieldOutput``.
All fieldOutputs are accessable to all outputmanagers at the end of each
increment, step and job.
Furthermore, they can be exported to ``*.csv`` files at the end of the analysis job.

ATTENTION:
    If the results are exported to a .csv file with enabled "saveHistory",
    the time History is automatically appended to the .csv file"
"""

import os
from dataclasses import dataclass
from typing import Callable, Union

import numpy as np

from edelweissfe.fields.nodefield import NodeField
from edelweissfe.journal.journal import Journal
from edelweissfe.models.femodel import FEModel
from edelweissfe.sets.elementset import ElementSet
from edelweissfe.sets.orderedset import OrderedSet
from edelweissfe.utils.elementresultcollector import ElementResultCollector
from edelweissfe.utils.schema import schemaField, subKeywordField


@dataclass(frozen=True)
class FieldOutputPerNodeSchema:
    """The options of a single ``>>perNode`` field-output block."""

    name: str | None = schemaField(description="Name of the field output.", dtype=str, default=None, required=True)
    field: str | None = schemaField(description="Field of the result.", dtype=str, default=None, required=True)
    result: str | None = schemaField(description="Result name.", dtype=str, default=None, required=True)
    elSet: str | None = schemaField(description="Element set.", dtype=str, default=None)
    nSet: str | None = schemaField(description="Node set.", dtype=str, default=None)
    rigidBody: str | None = schemaField(
        description="Rigid body (as registered in model.rigidBodies).", dtype=str, default=None
    )
    saveHistory: bool = schemaField(
        description="Save complete History or only last (increment) result", dtype=bool, default=False
    )
    f_x: str | None = schemaField(
        description="Function to apply in each increment.", dtype=str, default=None, optionName="f(x)"
    )
    f_export_x: str | None = schemaField(
        description="Function to apply on final result (table).", dtype=str, default=None, optionName="f_export(x)"
    )
    export: str | None = schemaField(
        description="Export the field output to a file at the end of the job.", dtype=str, default=None
    )


@dataclass(frozen=True)
class FieldOutputPerElementSchema:
    """The options of a single ``>>perElement`` field-output block."""

    name: str | None = schemaField(description="Name of the field output.", dtype=str, default=None, required=True)
    elSet: str | None = schemaField(description="Element set.", dtype=str, default=None, required=True)
    result: str | None = schemaField(description="Result name.", dtype=str, default=None, required=True)
    quadraturePoint: str | None = schemaField(description="Integer or slice.", dtype=str, default=None, required=True)
    saveHistory: bool = schemaField(
        description="Save complete History or only last (increment) result", dtype=bool, default=False
    )
    f_x: str | None = schemaField(
        description="Function to apply in each increment.", dtype=str, default=None, optionName="f(x)"
    )
    f_export_x: str | None = schemaField(
        description="Function to apply on final result (table).", dtype=str, default=None, optionName="f_export(x)"
    )
    export: str | None = schemaField(
        description="Export the field output to a file at the end of the job.", dtype=str, default=None
    )


@dataclass(frozen=True)
class FieldOutputFromExpressionSchema:
    """The options of a single ``>>fromExpression`` field-output block."""

    name: str | None = schemaField(description="Name of the field output.", dtype=str, default=None, required=True)
    elSet: str | None = schemaField(description="Element set.", dtype=str, default=None)
    nSet: str | None = schemaField(description="Node set.", dtype=str, default=None)
    expression: str | None = schemaField(
        description="Expression for retrieving field output.", dtype=str, default=None, required=True
    )
    saveHistory: bool = schemaField(
        description="Save complete History or only last (increment) result", dtype=bool, default=False
    )
    f_x: str | None = schemaField(
        description="Function to apply in each increment.", dtype=str, default=None, optionName="f(x)"
    )
    f_export_x: str | None = schemaField(
        description="Function to apply on final result (table).", dtype=str, default=None, optionName="f_export(x)"
    )
    export: str | None = schemaField(
        description="Export the field output to a file at the end of the job.", dtype=str, default=None
    )


@dataclass(frozen=True)
class FieldOutputSchema:
    """The sub-keyword blocks of the ``*fieldOutput`` keyword.

    ``*fieldOutput`` itself declares no line options of its own (see
    ``edelweissfe.keywords.fieldoutput``'s module docstring) -- its entire grammar is these three
    repeatable ``>>`` blocks. Documentation/parser-validation only: ``_FieldOutputBase`` and friends
    still construct from the raw parsed dict via ``abqmodelconstructor``/``inputfilehelpers``,
    unchanged.
    """

    perNode: tuple[FieldOutputPerNodeSchema, ...] = subKeywordField(
        description="Create node-based field output.", schema=FieldOutputPerNodeSchema
    )
    perElement: tuple[FieldOutputPerElementSchema, ...] = subKeywordField(
        description="Create element-based field output.", schema=FieldOutputPerElementSchema
    )
    fromExpression: tuple[FieldOutputFromExpressionSchema, ...] = subKeywordField(
        description="Create field output from expression.", schema=FieldOutputFromExpressionSchema
    )


[docs] class _FieldOutputBase: """ Entity of a fieldOutput request. Carries the history or the latest result. Parameters ---------- name The name of this FieldOutput. model A dictionary containing the model tree. journal The journal object for logging. saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. reshape_to_dimensions Reshape the result to a specific number of dimensions. """ def __init__( self, name: str, model: FEModel, journal: Journal, saveHistory: bool = False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, reshape_to_dimensions: int = None, ): self.timeTotal = 0.0 self.name = name self.model = model self.journal = journal self.appendResults = saveHistory self.result = [] if self.appendResults else None # bumped on every change of the result, see resultVersion self._version = 0 self.f = f_x self.f_export = fExport_x self.timeHistory = [] self.export = export #: The size of the export file the run has written so far. Before its first write after #: a start or a restored checkpoint, the file is truncated to it: a cold run starts the file #: afresh, a resumed run drops whatever the interrupted run wrote after its checkpoint, and #: a step falling back to a checkpoint drops the rows of the increments it discards. self._exportedBytes = 0 self._exportFileTruncated = False self._reshape_to_dimensions = reshape_to_dimensions
[docs] def hasResult(self) -> bool: """Whether a result was computed yet: ``False`` before the first converged increment. Returns ------- bool ``True`` once :meth:`getLastResult` has something to return. """ return bool(self.result) if self.appendResults else self.result is not None
@property def resultVersion(self) -> int: """The number of changes of :attr:`result` so far -- its updates and a restart's load: the key of whatever is derived from the result and kept until it changes (e.g. a marker's :meth:`~edelweissfe.adaptivity.marking.MarkerBase.selectionStamp`). A cache key, not state: a resumed run counts from its restart, not from the start of the analysis. It is backed by the change counter ``_version``, which, like the ``_version`` of a set, a surface or a node field, a restart need not reproduce. """ return self._version
[docs] def getLastResult( self, ) -> np.ndarray: """Get the last result, no matter if the history is stored or not. Returns ------- np.ndarray The result array. """ return self.result[-1] if self.appendResults else self.result
[docs] def getResultHistory( self, ) -> np.ndarray: """Get the history. Throws an exception if the history is not stored, or if the stored entries do not all share the same shape (e.g. because the mesh changed under adaptive refinement, in which case a rectangular per-node history is not well-defined). Returns ------- np.ndarray The result history. """ if not self.appendResults: raise Exception( "fieldOutput {:} does not save any history; please define it with saveHistory=True!".format(self.name) ) if self.result: firstShape = np.shape(self.result[0]) if any(np.shape(r) != firstShape for r in self.result): raise Exception( "fieldOutput {:} has a per-node/per-element history with varying shapes across " "increments (e.g. due to adaptive mesh refinement); a rectangular history is not " "well-defined. Use a reducing f(x) (as for RF in examples/WinklerL/WinklerL_blockamg_amr.inp) " "instead of saveHistory=True in this case.".format(self.name) ) return np.asarray(self.result)
[docs] def getTimeHistory( self, ) -> np.ndarray: """Get the time history. Returns ------- np.ndarray The time history. """ return np.asarray(self.timeHistory)
def _applyResultsPipleline(self, result): """Apply the pipeline of operations onto the results. Called by inheriting classes. Parameters ---------- model The model tree. """ self.timeHistory.append(self.model.time) if self._reshape_to_dimensions is not None: result = np.reshape(result, (-1, self._reshape_to_dimensions)) if self.f: result = self.f(result) if self.appendResults: self.result.append(result.copy()) else: self.result = result self._version += 1 def _resultTableForExport(self) -> np.ndarray: """Assemble the result table that ``f_export`` is applied to. Normally this is the whole stored history, so that a table-wide ``f_export`` such as ``x[:,:,1]`` sees the increment axis it expects. Under adaptive mesh refinement the per-increment shape changes, and the history as a whole is no longer representable as a single rectangular array; in that case only the last increment is tabulated, as a table of one row. This keeps the dimensionality (and hence ``f_export``) valid while exporting the current increment, which is all ``writeLastResult`` consumes. Returns ------- np.ndarray The result table, with the increment as leading axis. """ if not self.appendResults: return np.asarray(self.result) lastShape = np.shape(self.result[-1]) if any(np.shape(entry) != lastShape for entry in self.result): return np.asarray(self.result[-1:]) return np.asarray(self.result)
[docs] def writeLastResult(self): """Update file output. Parameters ---------- model The model tree. """ res = self._resultTableForExport() if self.f_export: res = self.f_export(res) if res.ndim > 2: self.journal.message("Reshaping fieldOutput result for export in .csv file", self.name) res = res.reshape((res.shape[0], -1)) fileName = f"{self.export}.csv" if not self._exportFileTruncated: open(fileName, "a").close() if os.path.getsize(fileName) > self._exportedBytes: os.truncate(fileName, self._exportedBytes) self._exportFileTruncated = True with open(fileName, "a") as f: np.savetxt( f, np.hstack( ( self.timeHistory[-1], res[-1,], ) ).reshape((1, -1)), ) self._exportedBytes = f.tell()
[docs] def initializeJob(self): """Initalize everything. Will also update the results based on the proved start time and solution.""" self.updateResults(self.model)
[docs] def getRestartData(self) -> dict[str, np.ndarray]: """The history this field output carries from one increment to the next -- the times, the results (flattened, since refinement changes their shape), and how much of the export file belongs to the run. Returns ------- dict[str, numpy.ndarray] The state. """ state = {"timeHistory": np.array(self.timeHistory, dtype=float), "exportedBytes": np.array(self._exportedBytes)} if self.appendResults: state["resultShapes"] = np.array([np.shape(r) for r in self.result], dtype=np.int64) state["resultValues"] = np.concatenate([np.ravel(r) for r in self.result]) if self.result else np.empty(0) elif self.result is not None: state["result"] = np.asarray(self.result) return state
[docs] def setRestartData(self, data: dict[str, np.ndarray]): """Restore the state :meth:`getRestartData` returned. Parameters ---------- data The state. """ self.timeHistory = [float(t) for t in data["timeHistory"]] self._exportedBytes = int(data["exportedBytes"]) self._exportFileTruncated = False if self.appendResults: self.result, offset = [], 0 for shape in data["resultShapes"]: size = int(np.prod(shape)) self.result.append(np.array(data["resultValues"][offset : offset + size]).reshape(shape)) offset += size else: self.result = np.array(data["result"]) if "result" in data else None self._version += 1
[docs] def initializeStep(self, step): """Write the current (just-updated) result as the first row of this step. Only at the start of a step: a step resumed from a checkpoint already exported this state, as the last row before the checkpoint was written. """ if not step.timeStepper.isAtStepStart(): return if self.export: self.writeLastResult()
[docs] def finalizeIncrement( self, ): """Finalize an increment, i.e. store the current results.""" self.updateResults(self.model) if self.export: self.writeLastResult()
[docs] def finalizeStep( self, ): """Bring the stored result up to date with the model's final state, if the step advanced past the increment that stored the current one. The output managers write one final frame at the end of a step, from the result stored here -- see the Ensight export's ``finalizeStep``, which writes whenever time has advanced since its last frame. If the topology changed after that result was stored, which is exactly what live h-adaptivity does between two output increments, the result belongs to a different mesh than the one being written and the export rejects it:: Variable displacement result size (32) does not match the number of nodes (141) Guarded on time having advanced, by the same rule the managers use, so the two decisions cannot disagree: whenever a manager writes a final frame the result behind it has just been refreshed, and a step ending on an increment that already stored a result does not store a second one at the same time and duplicate the last point of a history export. """ if not self.timeHistory or self.model.time - self.timeHistory[-1] > 1e-12: self.updateResults(self.model)
def finalizeJob( self, ): pass
[docs] def setResults(self, values: np.ndarray): """Modifies a result at it's origin, if possible. Throws an exception if not possible. Parameters ---------- values The values. """ raise Exception("setting field output currently not implemented for this type of output!")
def __eq__(self, other): if type(other) is str: return other == self.name return self.getLastResult() == other def __ne__(self, other): return self.getLastResult() != other def __lt__(self, other): return self.getLastResult() < other def __le__(self, other): return self.getLastResult() <= other def __gt__(self, other): return self.getLastResult() > other def __ge__(self, other): return self.getLastResult() >= other def __getitem__(self, index): return self.getLastResult()[index] def __add__(self, other): return self.getLastResult() + other def __sub__(self, other): return self.getLastResult() - other
[docs] class NodeFieldOutput(_FieldOutputBase): """ This is a Node based FieldOutput. It operates on NodeFields. Parameters ---------- name The name of this FieldOutput. nodeField The NodeField, on which this FieldOutput operates. result The name of the result entry in the NodeField, one of :attr:`~edelweissfe.fields.nodefield.NodeField.ENTRIES`. model The model tree instance. journal The journal object for logging. saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. Raises ------ KeyError If the result is not an entry a node field holds. """ def __init__( self, name: str, nodeField, result: str, model: FEModel, journal: Journal, saveHistory: bool = False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, ): if result not in NodeField.ENTRIES: raise KeyError( "field output {:}: a node field holds the results {:}, not '{:}'".format( name, ", ".join(NodeField.ENTRIES), result ) ) self.entry = result self.fieldName = nodeField.name self._nodeField = nodeField self.associatedSet = nodeField.associatedSet super().__init__(name, model, journal, saveHistory, f_x, export, fExport_x)
[docs] def updateResults(self, model: FEModel): """Update the field output. Will use the current solution and reaction vector if result is a nodal result. Parameters ---------- model The model tree. """ result = self._nodeField[self.entry] return super()._applyResultsPipleline(result)
[docs] class RigidBodyFieldOutput(_FieldOutputBase): """ This is a FieldOutput operating directly on a RigidBody. A rigid body's visualization (e.g., surface) nodes are not independent degrees of freedom -- they are fully determined by the rigid body's reference point, which is itself part of an ordinary NodeField. This FieldOutput therefore does not operate on a NodeField at all: it queries :meth:`~edelweissfe.rigidbodies.rigidbody.RigidBody.getVisualizationField`, which computes the requested field directly from the rigid body's (already solved) kinematics. Parameters ---------- name The name of this FieldOutput. rigidBody The RigidBody on which this FieldOutput operates. field The name of the field to retrieve from the rigid body (e.g., "displacement"). model The model tree instance. journal The journal object for logging. saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. """ def __init__( self, name: str, rigidBody, field: str, model: FEModel, journal: Journal, saveHistory: bool = False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, ): self._rigidBody = rigidBody self._field = field self.associatedSet = rigidBody super().__init__(name, model, journal, saveHistory, f_x, export, fExport_x)
[docs] def updateResults(self, model: FEModel): """Update the field output from the rigid body's current kinematics. Parameters ---------- model The model tree. """ result = self._rigidBody.getVisualizationField(self._field) return super()._applyResultsPipleline(result)
[docs] class ElementFieldOutput(_FieldOutputBase): """ This is a Element based FieldOutput. It operates on ElementSets. Parameters ---------- name The name of this FieldOutput. elSet The ElementSet on which this FieldOutput operates. resultName The name of the result entry in the :class:`ElementBase`. model The model tree instance. journal The journal object for logging. saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. quadraturePoints The list of quadrature points for which the results should be extracted. """ def __init__( self, name: str, elSet: ElementSet, resultName: str, model: FEModel, journal: Journal, saveHistory: bool = False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, quadraturePoints: Union[int, slice, list[int]] = 0, ): self.associatedSet = elSet self.resultName = resultName self.quadraturePoints = quadraturePoints self._seenSetVersion = elSet._version self.elementResultCollector = ElementResultCollector( list(self.associatedSet), self.quadraturePoints, self.resultName ) super().__init__(name, model, journal, saveHistory, f_x, export, fExport_x) def _rebuildCollectorIfSetChanged(self): """Rebuild the element result collector -- which pins a fixed snapshot of the element list at construction -- if the associated ElementSet was mutated in-place (e.g. AMR replacing a refined parent element with its children) since the last check. Unlike a plain iteration over the set, this pinned snapshot does not see new elements on its own.""" if self.associatedSet._version != self._seenSetVersion: self.elementResultCollector = ElementResultCollector( list(self.associatedSet), self.quadraturePoints, self.resultName ) self._seenSetVersion = self.associatedSet._version
[docs] def updateResults(self, model: FEModel): """Update the field output. Will use the current solution and reaction vector if result is a nodal result. Parameters ---------- model The model tree. """ self._rebuildCollectorIfSetChanged() result = self.elementResultCollector.getCurrentResults() super()._applyResultsPipleline(result)
[docs] def setResults(self, values: np.ndarray): """Modifies a result at it's origin, if possible. Throws an exception if not possible. Parameters ---------- values The values. """ if self.f: raise Exception("cannot set field output for modified results (f(x) != None) !") self._rebuildCollectorIfSetChanged() for i, el in enumerate(self.associatedSet): for j, g in enumerate(self.quadraturePoints): theArray = el.getResultArray(self.resultName, g, True) theArray[:] = values[i, j, :] el.acceptLastState()
[docs] class ExpressionFieldOutput(_FieldOutputBase): """ This is a Node based FieldOutput. It operates on NodeFields. Parameters ---------- associatedSet The associated set of nodes or elements. theExpression The expression to be evaluated. name The name of this FieldOutput. nodeField The NodeField, on which this FieldOutput operates. result The name of the result entry in the NodeField. model The model tree instance. journal The journal object for logging. saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. """ def __init__( self, associatedSet: OrderedSet | None, theExpression, name: str, model: FEModel, journal: Journal, saveHistory: bool = False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, ): self.associatedSet = associatedSet self.theExpression = theExpression super().__init__(name, model, journal, saveHistory, f_x, export, fExport_x)
[docs] def updateResults(self, model: FEModel): """Update the field output. Will use the current solution and reaction vector if result is a nodal result. Parameters ---------- model The model tree. """ if self.associatedSet: result = np.reshape(np.asarray(self.theExpression()), (len(self.associatedSet), -1)) else: result = np.reshape(np.asarray(self.theExpression()), (1, -1)) return super()._applyResultsPipleline(result)
[docs] class FieldOutputController: """ The central module for managing field outputs, which can be used by output managers. """ def __init__(self, model: FEModel, journal: Journal): self.model = model self.journal = journal self.fieldOutputs = {}
[docs] def addExpressionFieldOutput( self, associatedSet: set, theExpression: Callable, name: str, saveHistory=False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, ): """Add a new FieldOutput entry to be computed during the simulation Parameters ---------- associatedSet The associated set of nodes or elements. theExpression The expression to be evaluated. name The name of this FieldOutput. saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. """ if name in self.fieldOutputs: raise Exception("FieldOutput {:} already exists!".format(name)) self.fieldOutputs[name] = ExpressionFieldOutput( associatedSet, theExpression, name, self.model, self.journal, saveHistory, f_x=f_x, export=export, fExport_x=fExport_x, )
[docs] def addPerNodeFieldOutput( self, name: str, nodeField: NodeField, result: str = None, saveHistory=False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, ): """Add a new FieldOutput entry to be computed during the simulation Parameters ---------- name The name of this FieldOutput. nodeField The :class:`NodeField` on which this FieldOutput should operate. result The name of the result entry in the :class:`NodeField`. journal The :class:`Journal` instance for logging purposes. saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. """ if not result: result = name if name in self.fieldOutputs: raise Exception("FieldOutput {:} already exists!".format(name)) self.fieldOutputs[name] = NodeFieldOutput( name, nodeField, result, self.model, self.journal, saveHistory, f_x, export, fExport_x, )
[docs] def addRigidBodyFieldOutput( self, name: str, rigidBody, field: str, saveHistory=False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, ): """Add a new FieldOutput entry operating directly on a RigidBody. Parameters ---------- name The name of this FieldOutput. rigidBody The :class:`~edelweissfe.rigidbodies.rigidbody.RigidBody` on which this FieldOutput should operate. field The name of the field to retrieve from the rigid body (e.g., "displacement"). saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. """ if name in self.fieldOutputs: raise Exception("FieldOutput {:} already exists!".format(name)) self.fieldOutputs[name] = RigidBodyFieldOutput( name, rigidBody, field, self.model, self.journal, saveHistory, f_x, export, fExport_x, )
[docs] def addPerElementFieldOutput( self, name: str, elSet: ElementSet, result: str = None, saveHistory=False, f_x: Callable = None, export: str = None, fExport_x: Callable = None, quadraturePoints=Union[int, slice, list[int]], ): """Add a new FieldOutput entry to be computed during the simulation Parameters ---------- name The name of this FieldOutput. elSet The :class:`ElementSet` on which this FieldOutput should operate. result The name of the result, which is provided by the Elements in the :class:`ElementSet`. journal The :class:`Journal` instance for logging purposes. saveHistory Save the complete history or only the last result. f_x Apply a math function on the results. export Export the results to a file. fExport_x Apply a math function on the results before exporting. quadraturePoints The indices of quadrature points for which the results should be extracted. """ if not result: result = name if name in self.fieldOutputs: raise Exception("FieldOutput {:} already exists!".format(name)) self.fieldOutputs[name] = ElementFieldOutput( name, elSet, result, self.model, self.journal, saveHistory, f_x, export, fExport_x, quadraturePoints, )
def initializeJob(self): for fieldOutput in self.fieldOutputs.values(): fieldOutput.initializeJob()
[docs] def finalizeIncrement( self, ): """Finalize all field outputs at the end of an increment.""" for output in self.fieldOutputs.values(): output.finalizeIncrement()
[docs] def finalizeStep( self, ): """Finalize all field outputs at the end of a step.""" for output in self.fieldOutputs.values(): output.finalizeStep()
[docs] def initializeStep(self, step): """Initalize an step. Parameters ---------- step The step information. """ for output in self.fieldOutputs.values(): output.initializeStep(step)
[docs] def finalizeJob( self, ): """Finalize all field outputs at the end of a job.""" for output in self.fieldOutputs.values(): output.finalizeJob()