`visualdynamics.core.replication_segments`¶

replication_segments ¶

A replicated record judged segment by segment, each by the rule for its kind of motion.

replication.compare reads a replication the way a controller's screen does: one waveform error over the whole record, one worst deviation of a maximax spectrum, one level. Over a long record those figures are dominated by whatever carries most of its mean square, and a short event carries very little of it. A shock a fraction of a second long in a minute of road can be missed by several decibels across part of its spectrum while the whole-record error still reads under five percent, because the shock holds a percent or two of the record's mean square. The whole-record spectrum cannot see it either: a maximax over the record keeps each oscillator's largest response anywhere in it, and that may come from a different event.

So the record is cut into segments, each marked as one of three kinds of motion, and each is held to the tolerance of its kind, the way test documents commonly write a replication's tolerance:

stationary: a stretch whose statistics hold steady (a rough road at constant speed). Judged in frequency by its ASD against the reference's, line by line within a band (±3 dB by default), and in time by its mean square about zero against the reference's (within +10 percent by default; an undertest shows in the ASD band).
transient: a long non-stationary stretch whose spectrum holds its shape while its level changes (braking, a run-up). Judged by its short-time rms envelope, window by window, against the reference's (within 1 dB by default), and by its normalized ASD, the spectrum with the segment's own mean square divided out, within a band (±3 dB by default).
shock: a short event. Judged by its shock response spectrum in each polarity against the reference's, point by point (−6/+3 dB by default), and by its major peaks, the reference's largest crests and troughs, each read against the control at the same sample (within 20 percent by default).

The tolerances are examples of what test documents ask for, not any one document's requirement; every one is an argument.

A stationary or transient segment is weighted by a tapered cosine window before its ASD and its mean square are computed (10 percent of it by default, half at each end), so the segment's abrupt ends do not enter its spectrum as content the record does not have. A shock's SRS is computed on the segment as recorded: the SRS is a peak response, not an average, and a mark drawn tight on the event would put the event's first and largest cycles under the taper, weighting down the reference and the control alike there and hiding a miss in the ratio. The two local readings, the envelope and the peaks, are read on the segment as recorded too: they are ratios of the two traces at one place, from which a weight applied to both cancels.

A segment is judged only on samples that are numbers. A trace holding a NaN or an infinity is refused, and a control that is zero where the reference asks for something reads minus infinity decibels there and fails, so a dead channel never reads as a pass.

The comparison assumes the two traces are on like terms. The reference and the control should be band-limited alike to the test band (core.filters) before they are judged; replication_segments aligns them by the one integer shift replication.lag_of finds, and judge_segments takes arrays already aligned. The whole-record rms error is reported beside the segments, so the reading that hides a segment's miss is on the page next to the one that shows it.

References

The segments and the forms of their tolerances follow the first of these; the readings are standard estimators, set out in the others. Each shock segment's spectra come from core.srs, which cites its filter.

Department of Defense (2019). MIL-STD-810H, Environmental Engineering Considerations and Laboratory Tests. Method 525.2, paragraph 4.2.2c: a replication judged segment by segment, each stretch stationary, a short transient or a long transient, a stationary or long one tapered at its ends by a 10 percent cosine window, and each held to tolerances of the forms used here.
Welch, P. D. (1967). "The Use of Fast Fourier Transform for the Estimation of Power Spectra: A Method Based on Time Averaging Over Short, Modified Periodograms." IEEE Transactions on Audio and Electroacoustics, 15(2), 70-73. doi:10.1109/TAU.1967.1161901 The averaged, windowed periodogram each ASD here is, the estimator TimeHistory.compute_psds computes.
Harris, F. J. (1978). "On the Use of Windows for Harmonic Analysis with the Discrete Fourier Transform." Proceedings of the IEEE, 66(1), 51-83. doi:10.1109/PROC.1978.10837 The tapered cosine window a stationary or transient segment is weighted by.

Classes:

Name	Description
`Segment`	One stretch of the record and the kind of motion it is.
`SegmentCheck`	One reading of one segment against its tolerance.
`SegmentResult`	One segment judged: where it is, what share of the record it
`SegmentedReplication`	A replicated record judged segment by segment.

Functions:

Name	Description
`band_text`	A tolerance in words: '±3 dB', '−6/+3 dB', 'up to +10 %'.
`tapered_cosine`	The weights of a tapered cosine (Tukey) window over `samples`
`major_peaks`	The samples of the reference's major peaks: its largest crests
`as_segments`	Segment marks as `Segment`s: each a `Segment`, or a tuple of
`judge_segments`	A replicated record judged segment by segment, for a reference
`aligned_traces`	(time, reference, control, channel, lag): one channel of a run
`replication_segments`	A replicated run judged segment by segment against the reference

Classes¶

Segment ¶

Bases: NamedTuple

One stretch of the record and the kind of motion it is.

start and end are in seconds on the reference's own clock, from its first sample. kind is one of SEGMENT_KINDS. name is what the readout calls it; when empty it is called by its kind and its place among the segments of that kind ('shock 1').

SegmentCheck `dataclass` ¶

SegmentCheck(reading: str, unit: str, tolerance: tuple[float, float], where: ndarray, values: ndarray, outside: int, passed: bool | None, needed: int | None = None)

One reading of one segment against its tolerance.

reading names it ('ASD', 'variance', 'envelope', 'normalized ASD', 'SRS positive', 'SRS negative', 'major peaks'). where and values are each judged item and its reading: lines in Hz for the spectra, window starts in seconds for the envelope, peak times in seconds for the peaks (both on the reference's clock), and one item for the variance; values are in unit, 'dB' or '%', as distances from the reference. low and high are the lowest and highest reading, tolerance the band it was held to, outside how many items fell outside it. passed is None when nothing could be judged.

Methods:

Name	Description
`describe`	The check in a line: the reading, its range, the tolerance,

Attributes:

Name	Type	Description
`judged`	`int`	How many items were read.
`low`	`float`	The lowest reading, NaN when there is none.
`high`	`float`	The highest reading, NaN when there is none.

Attributes¶

judged `property` ¶

judged: int

How many items were read.

low `property` ¶

low: float

The lowest reading, NaN when there is none.

high `property` ¶

high: float

The highest reading, NaN when there is none.

Methods:¶

describe ¶

describe() -> str

The check in a line: the reading, its range, the tolerance, and the verdict, as in 'SRS negative −7.4 to +0.1 dB (−6/+3 dB), 4 of 65 outside: fails'.

Source code in src/visualdynamics/core/replication_segments.py

def describe(self) -> str:
    """The check in a line: the reading, its range, the tolerance,
    and the verdict, as in 'SRS negative −7.4 to +0.1 dB (−6/+3 dB),
    4 of 65 outside: fails'."""
    if not self.judged:
        return f'{self.reading}: nothing to judge'
    span = (f'{_signed(self.low, self.unit)}' if self.judged == 1
            else f'{_signed(self.low, self.unit)} to '
                 f'{_signed(self.high, self.unit)}')
    text = f'{self.reading} {span} ({band_text(self.tolerance, self.unit)})'
    if self.judged > 1:
        text += f', {self.outside} of {self.judged} outside'
    verdict = {True: 'passes', False: 'fails', None: 'not judged'}[
        self.passed]
    return f'{text}: {verdict}'

SegmentResult `dataclass` ¶

SegmentResult(segment: Segment, label: str, first: int, samples: int, share: float, rms_error: float, checks: tuple[SegmentCheck, ...], passed: bool | None)

One segment judged: where it is, what share of the record it holds, and each of its checks.

first and samples place it in the aligned traces. share is its share of the reference's mean square over the whole record, and rms_error its own rms error, both in percent and both read on the segment as recorded: together they say how much the segment can move the whole-record figure. passed is False when any check fails, None when none fails and one or more could not be judged, and True only when every check passes.

Methods:

Name	Description
`describe`	The segment in a line: its name, kind, span and verdict, and

Methods:¶

describe ¶

describe() -> str

The segment in a line: its name, kind, span and verdict, and which checks failed.

Source code in src/visualdynamics/core/replication_segments.py

def describe(self) -> str:
    """The segment in a line: its name, kind, span and verdict, and
    which checks failed."""
    seg = self.segment
    verdict = {True: 'passes', False: 'fails', None: 'not judged'}[
        self.passed]
    text = (f'{self.label} ({seg.kind}, {seg.start:g} to {seg.end:g} s): '
            f'{verdict}')
    failed = [check.reading for check in self.checks
              if check.passed is False]
    if failed:
        text += ' on ' + ', '.join(failed)
    return text

SegmentedReplication `dataclass` ¶

SegmentedReplication(segments: tuple[SegmentResult, ...], rms_error: float, sample_rate: float, samples: int, passed: bool | None, settings: dict[str, Any] = dict(), channel: str | None = None, lag: int = 0)

A replicated record judged segment by segment.

segments are the results in the order the marks were given. rms_error is the whole record's rms error in percent, the figure a controller reports, kept beside the segments because it is the one that hides a segment's miss. passed is False when any segment fails, None when none fails and one or more could not be judged (or none is marked), and True only when every segment passes. channel and lag say which record was read and how far it was shifted to align it (None and 0 for arrays given directly). settings are the tolerances it was judged under, as judge_segments took them.

Methods:

Name	Description
`describe`	The record in a line: the whole-record rms error, and the

Attributes:

Name	Type	Description
`failed`	`list[SegmentResult]`	The segments that failed, in order.
`not_judged`	`list[SegmentResult]`	The segments that could not be judged, in order.

Attributes¶

failed `property` ¶

failed: list[SegmentResult]

The segments that failed, in order.

not_judged `property` ¶

not_judged: list[SegmentResult]

The segments that could not be judged, in order.

Methods:¶

describe ¶

describe() -> str

The record in a line: the whole-record rms error, and the verdict by segment. The counts are of the segments judged; a segment that could not be judged is named on its own.

Source code in src/visualdynamics/core/replication_segments.py

def describe(self) -> str:
    """The record in a line: the whole-record rms error, and the
    verdict by segment. The counts are of the segments judged; a
    segment that could not be judged is named on its own."""
    head = f'whole-record rms error {self.rms_error:.2f} %'
    if not self.segments:
        return f'{head}; no segment marked'
    failed, unjudged = self.failed, self.not_judged
    judged = len(self.segments) - len(unjudged)
    if failed:
        names = ', '.join(result.label for result in failed)
        text = (f'{head}; {len(failed)} of {judged} segments fail '
                f'({names})')
    elif not judged:
        text = f'{head}; not judged'
    elif not unjudged:
        return f'{head}; all {judged} segments pass'
    else:
        text = f'{head}; {judged} of {len(self.segments)} segments pass'
    if unjudged and judged:
        names = ', '.join(result.label for result in unjudged)
        text += f', {len(unjudged)} not judged ({names})'
    return text

Functions:¶

band_text ¶

band_text(tolerance: tuple[float, float], unit: str) -> str

A tolerance in words: '±3 dB', '−6/+3 dB', 'up to +10 %'.

Source code in src/visualdynamics/core/replication_segments.py

def band_text(tolerance: tuple[float, float], unit: str) -> str:
    """A tolerance in words: '±3 dB', '−6/+3 dB', 'up to +10 %'."""
    below, above = tolerance
    if np.isinf(below) and below < 0:
        text = f'up to {above:+g} {unit}'
    elif np.isinf(above):
        text = f'from {below:+g} {unit}'
    elif np.isclose(-below, above):
        text = f'\u00b1{above:g} {unit}'
    else:
        text = f'{below:+g}/{above:+g} {unit}'
    return text.replace('-', '\u2212')

tapered_cosine ¶

tapered_cosine(samples: int, percent: float = TAPER_PERCENT) -> ndarray

The weights of a tapered cosine (Tukey) window over samples samples.

percent of the window is under the cosine, half rising from zero at the start and half falling to zero at the end; the middle is weighted by one. 0 is a rectangle and 100 a Hann window. This is the symmetric form, the one a segment cut from a record is weighted by, where averaging.window_shape gives the periodic form spectral frames use.

Raises:

Type	Description
`ValueError`	For fewer than one sample, or a share outside 0 to 100.

Source code in src/visualdynamics/core/replication_segments.py

def tapered_cosine(samples: int, percent: float = TAPER_PERCENT
                   ) -> np.ndarray:
    """The weights of a tapered cosine (Tukey) window over `samples`
    samples.

    `percent` of the window is under the cosine, half rising from zero
    at the start and half falling to zero at the end; the middle is
    weighted by one. 0 is a rectangle and 100 a Hann window. This is
    the symmetric form, the one a segment cut from a record is weighted
    by, where `averaging.window_shape` gives the periodic form spectral
    frames use.

    Raises
    ------
    ValueError
        For fewer than one sample, or a share outside 0 to 100.
    """
    from scipy.signal import windows

    samples = int(samples)
    if samples < 1:
        raise ValueError('a window needs at least one sample')
    share = float(percent) / 100.0
    if not 0.0 <= share <= 1.0:
        raise ValueError(f'{percent:g} percent is not a share of a segment')
    # scipy's, not a second one written here (2026-10-16 review)
    return windows.tukey(samples, share, sym=True)

major_peaks ¶

major_peaks(reference: ArrayLike, each: int = PEAKS, share_percent: float = MAJOR_SHARE_PERCENT) -> ndarray

The samples of the reference's major peaks: its largest crests (local maxima above zero) and deepest troughs (local minima below zero), up to each of each, largest first, crests then troughs.

No more than share_percent of all its peaks are counted as major, and never fewer than one each way where there is one, so a segment with few peaks is read on its very largest.

Source code in src/visualdynamics/core/replication_segments.py

def major_peaks(reference: ArrayLike, each: int = PEAKS,
                share_percent: float = MAJOR_SHARE_PERCENT) -> np.ndarray:
    """The samples of the reference's major peaks: its largest crests
    (local maxima above zero) and deepest troughs (local minima below
    zero), up to `each` of each, largest first, crests then troughs.

    No more than `share_percent` of all its peaks are counted as major,
    and never fewer than one each way where there is one, so a segment
    with few peaks is read on its very largest.
    """
    x = np.asarray(reference, dtype=float)
    if x.size < 3:
        return np.empty(0, dtype=int)
    middle = x[1:-1]
    crests = np.flatnonzero((middle > x[:-2]) & (middle >= x[2:])
                            & (middle > 0.0)) + 1
    troughs = np.flatnonzero((middle < x[:-2]) & (middle <= x[2:])
                             & (middle < 0.0)) + 1
    allowed = int(share_percent / 100.0 * (crests.size + troughs.size) / 2.0)
    take = min(int(each), max(allowed, 1))
    top = crests[np.argsort(x[crests])[::-1][:take]]
    bottom = troughs[np.argsort(x[troughs])[:take]]
    return np.concatenate([top, bottom]).astype(int)

as_segments ¶

as_segments(marks: Sequence[Any]) -> list[Segment]

Segment marks as Segments: each a Segment, or a tuple of (start, end, kind) with an optional name.

Raises:

Type	Description
`ValueError`	For a kind that is not one of `SEGMENT_KINDS`, or a segment that does not end after it starts.

Source code in src/visualdynamics/core/replication_segments.py

def as_segments(marks: Sequence[Any]) -> list[Segment]:
    """Segment marks as `Segment`s: each a `Segment`, or a tuple of
    (start, end, kind) with an optional name.

    Raises
    ------
    ValueError
        For a kind that is not one of `SEGMENT_KINDS`, or a segment
        that does not end after it starts.
    """
    out = []
    for mark in marks:
        segment = Segment(float(mark[0]), float(mark[1]), str(mark[2]),
                          str(mark[3]) if len(mark) > 3 else '')
        if segment.kind not in SEGMENT_KINDS:
            raise ValueError(f'{segment.kind!r} is not a kind of segment; '
                             'the kinds are ' + ', '.join(SEGMENT_KINDS))
        if not segment.end > segment.start:
            raise ValueError(f'a segment from {segment.start:g} to '
                             f'{segment.end:g} s does not end after it '
                             'starts')
        out.append(segment)
    return out

judge_segments ¶

judge_segments(reference: ArrayLike, control: ArrayLike, sample_rate: float, segments: Sequence[Any], *, taper_percent: float = TAPER_PERCENT, band: tuple[float, float] | None = None, lines: float = LINES_HZ, asd_db: tuple[float, float] = ASD_DB, asd_percent: float = 100.0, variance_percent: tuple[float, float] = VARIANCE_PERCENT, envelope_db: tuple[float, float] = ENVELOPE_DB, window: float | None = None, srs_db: tuple[float, float] = SRS_DB, q: float | None = None, per_octave: int | None = None, peak_percent: float = PEAK_PERCENT, peaks: int = PEAKS) -> SegmentedReplication

A replicated record judged segment by segment, for a reference and a control trace held as plain arrays, aligned: the array form of replication_segments.

result = judge_segments(reference, control, 2048.0, [
    (6.0, 30.0, 'stationary', 'rough road'),
    (31.9, 32.7, 'shock', 'pothole'),
    (41.0, 51.0, 'transient', 'braking')], band=(5.0, 400.0))
result.rms_error, result.passed, result.failed

Each segment is held to the rule for its kind (see the module): a stationary one by its ASD and its mean square, a transient one by its short-time rms envelope and its normalized ASD, a shock by its SRS in each polarity and its major peaks. Every tolerance is an example of what test documents ask for and is an argument here.

Parameters:

Name	Type	Description	Default
`reference`	`array - like`	The two traces, one value per sample, aligned, in the same units. Ratios are taken, so the units do not matter.	required
`control`	`array - like`	The two traces, one value per sample, aligned, in the same units. Ratios are taken, so the units do not matter.	required
`sample_rate`	`float`	Samples per second.	required
`segments`	`sequence`	The marks, each a `Segment` or (start, end, kind[, name]), start and end in seconds from the first sample.	required
`taper_percent`	`float`	The share of a stationary or transient segment under its tapered cosine window's ends, before its ASD and mean square are computed. A shock's SRS is read untapered.	`TAPER_PERCENT`
`band`	`(float, float)`	The test band in Hz: the lines an ASD is judged at and the natural frequencies an SRS is computed at. By default an ASD is judged at every line above DC and an SRS from one cycle in the segment up to a fifth of the sample rate. Either way an SRS stops at a fifth of the sample rate, and a line or a point more than `ASD_FLOOR_DB` (or `replication.SRS_FLOOR_DB`) below the reference's highest is not judged.	`None`
`lines`	`float`	The ASD's line spacing in Hz; a segment shorter than one frame is one frame.	`LINES_HZ`
`asd_db`	`tuple[float, float]`	The ASD band in dB, for a stationary segment's ASD and a transient segment's normalized ASD, and the share of the judged lines that must lie inside it (every one by default).	`ASD_DB`
`asd_percent`	`tuple[float, float]`	The ASD band in dB, for a stationary segment's ASD and a transient segment's normalized ASD, and the share of the judged lines that must lie inside it (every one by default).	`ASD_DB`
`variance_percent`	`(float, float)`	A stationary segment's mean square against the reference's, in percent below and above.	`VARIANCE_PERCENT`
`envelope_db`	`(float, float)`	A transient segment's short-time rms against the reference's in every window, in dB.	`ENVELOPE_DB`
`window`	`float`	The envelope's window and step in seconds; `ENVELOPE_PERCENT` of the whole record by default.	`None`
`srs_db`	`(float, float)`	A shock segment's SRS against the reference's, each polarity, at every natural frequency judged.	`SRS_DB`
`q`	`float \| None`	The SRS's quality factor and spacing, `srs.DEFAULT_Q` and `srs.PER_OCTAVE` by default.	`None`
`per_octave`	`float \| None`	The SRS's quality factor and spacing, `srs.DEFAULT_Q` and `srs.PER_OCTAVE` by default.	`None`
`peak_percent`	`float`	How far each major peak may be from the reference's, in percent, and how many are read each way (`major_peaks`).	`PEAK_PERCENT`
`peaks`	`float`	How far each major peak may be from the reference's, in percent, and how many are read each way (`major_peaks`).	`PEAK_PERCENT`

Returns:

Type	Description
`SegmentedReplication`

Raises:

Type	Description
`ValueError`	For traces of different lengths or holding samples that are not numbers, a segment outside the record or too short to judge, an unknown kind, or a tolerance that is not a band.

Source code in src/visualdynamics/core/replication_segments.py

def judge_segments(reference: ArrayLike, control: ArrayLike,
                   sample_rate: float, segments: Sequence[Any], *,
                   taper_percent: float = TAPER_PERCENT,
                   band: tuple[float, float] | None = None,
                   lines: float = LINES_HZ,
                   asd_db: tuple[float, float] = ASD_DB,
                   asd_percent: float = 100.0,
                   variance_percent: tuple[float, float] = VARIANCE_PERCENT,
                   envelope_db: tuple[float, float] = ENVELOPE_DB,
                   window: float | None = None,
                   srs_db: tuple[float, float] = SRS_DB,
                   q: float | None = None,
                   per_octave: int | None = None,
                   peak_percent: float = PEAK_PERCENT,
                   peaks: int = PEAKS) -> SegmentedReplication:
    """A replicated record judged segment by segment, for a reference
    and a control trace held as plain arrays, aligned: the array form
    of `replication_segments`.

        result = judge_segments(reference, control, 2048.0, [
            (6.0, 30.0, 'stationary', 'rough road'),
            (31.9, 32.7, 'shock', 'pothole'),
            (41.0, 51.0, 'transient', 'braking')], band=(5.0, 400.0))
        result.rms_error, result.passed, result.failed

    Each segment is held to the rule for its kind (see the module):
    a stationary one by its ASD and its mean square, a transient one by
    its short-time rms envelope and its normalized ASD, a shock by its
    SRS in each polarity and its major peaks. Every tolerance is an
    example of what test documents ask for and is an argument here.

    Parameters
    ----------
    reference, control : array-like
        The two traces, one value per sample, aligned, in the same
        units. Ratios are taken, so the units do not matter.
    sample_rate : float
        Samples per second.
    segments : sequence
        The marks, each a `Segment` or (start, end, kind[, name]),
        start and end in seconds from the first sample.
    taper_percent : float, default `TAPER_PERCENT`
        The share of a stationary or transient segment under its
        tapered cosine window's ends, before its ASD and mean square
        are computed. A shock's SRS is read untapered.
    band : (float, float), optional
        The test band in Hz: the lines an ASD is judged at and the
        natural frequencies an SRS is computed at. By default an ASD is
        judged at every line above DC and an SRS from one cycle in the
        segment up to a fifth of the sample rate. Either way an SRS
        stops at a fifth of the sample rate, and a line or a point more
        than `ASD_FLOOR_DB` (or `replication.SRS_FLOOR_DB`) below the
        reference's highest is not judged.
    lines : float, default `LINES_HZ`
        The ASD's line spacing in Hz; a segment shorter than one frame
        is one frame.
    asd_db, asd_percent
        The ASD band in dB, for a stationary segment's ASD and a
        transient segment's normalized ASD, and the share of the judged
        lines that must lie inside it (every one by default).
    variance_percent : (float, float), default `VARIANCE_PERCENT`
        A stationary segment's mean square against the reference's, in
        percent below and above.
    envelope_db : (float, float), default `ENVELOPE_DB`
        A transient segment's short-time rms against the reference's in
        every window, in dB.
    window : float, optional
        The envelope's window and step in seconds;
        `ENVELOPE_PERCENT` of the whole record by default.
    srs_db : (float, float), default `SRS_DB`
        A shock segment's SRS against the reference's, each polarity,
        at every natural frequency judged.
    q, per_octave
        The SRS's quality factor and spacing, `srs.DEFAULT_Q` and
        `srs.PER_OCTAVE` by default.
    peak_percent, peaks
        How far each major peak may be from the reference's, in
        percent, and how many are read each way (`major_peaks`).

    Returns
    -------
    SegmentedReplication

    Raises
    ------
    ValueError
        For traces of different lengths or holding samples that are not
        numbers, a segment outside the record or too short to judge, an
        unknown kind, or a tolerance that is not a band.
    """
    from .srs import DEFAULT_Q, PER_OCTAVE

    r = np.asarray(np.real(reference), dtype=float).ravel()
    c = np.asarray(np.real(control), dtype=float).ravel()
    if r.size != c.size:
        raise ValueError(f'the reference has {r.size} samples and the '
                         f'control {c.size}; align them first')
    # a NaN would be dropped from a ratio as if it were not there, and
    # a run with a dead half would pass on the half that is
    # (2026-10-16 review)
    for name, x in (('reference', r), ('control', c)):
        bad = int(np.sum(~np.isfinite(x)))
        if bad:
            raise ValueError(f'the {name} holds {bad} samples that are '
                             'not numbers')
    rate = float(sample_rate)
    marks = as_segments(segments)
    settings = {
        'taper_percent': float(taper_percent),
        'band': None if band is None else _band('band', band),
        'lines': float(lines),
        'asd_db': _band('ASD', asd_db), 'asd_percent': float(asd_percent),
        'variance_percent': _band('variance', variance_percent),
        'envelope_db': _band('envelope', envelope_db),
        'window': None if window is None else float(window),
        'srs_db': _band('SRS', srs_db),
        'q': DEFAULT_Q if q is None else float(q),
        'per_octave': PER_OCTAVE if per_octave is None else int(per_octave),
        'peak_percent': float(peak_percent), 'peaks': int(peaks)}
    if settings['band'] is not None and settings['band'][0] <= 0.0:
        raise ValueError('the test band starts above 0 Hz')
    if not 0.0 < settings['asd_percent'] <= 100.0:
        raise ValueError(f'{asd_percent:g} percent is not a share of the '
                         'lines')
    if not settings['lines'] > 0.0:
        raise ValueError('the line spacing is above 0 Hz')
    if not settings['peak_percent'] > 0.0 or settings['peaks'] < 1:
        raise ValueError('the peaks are read one or more each way, '
                         'within a share above 0 percent')
    total = float(np.sum(r ** 2))
    whole = (100.0 * float(np.linalg.norm(c - r)) / np.sqrt(total)
             if total > 0.0 else float('nan'))
    envelope = (settings['window'] if settings['window'] is not None
                else ENVELOPE_PERCENT / 100.0 * r.size / rate)
    span = max(round(envelope * rate), 2)
    frame = max(round(rate / settings['lines']), 2)

    results = []
    for segment, label in zip(marks, _label(marks)):
        first = round(segment.start * rate)
        stop = round(segment.end * rate)
        if first < 0 or stop > r.size:
            raise ValueError(f'{label} runs {segment.start:g} to '
                             f'{segment.end:g} s, outside the record\'s '
                             f'{r.size / rate:g} s')
        n = stop - first
        if n < 8:
            raise ValueError(f'{label} holds {n} samples, too few to judge')
        rs, cs = r[first:stop], c[first:stop]
        weights = tapered_cosine(n, settings['taper_percent'])
        rw, cw = rs * weights, cs * weights
        own = float(np.sum(rs ** 2))
        share = 100.0 * own / total if total > 0.0 else float('nan')
        error = (100.0 * float(np.linalg.norm(cs - rs)) / np.sqrt(own)
                 if own > 0.0 else float('nan'))
        if segment.kind == 'stationary':
            want = float(np.mean(rw ** 2))
            got = float(np.mean(cw ** 2))
            ratio = (100.0 * (got / want - 1.0) if want > 0.0
                     else float('nan'))
            variance = (_check('variance', '%', settings['variance_percent'],
                               [segment.start], [ratio])
                        if np.isfinite(ratio) else
                        _check('variance', '%', settings['variance_percent'],
                               [], []))
            checks = [_spectral('ASD', rw, cw, rate, min(frame, n),
                                settings['band'], settings['asd_db'],
                                settings['asd_percent'], False),
                      variance]
        elif segment.kind == 'transient':
            checks = [_envelope(rs, cs, min(span, n), first, rate,
                                settings['envelope_db']),
                      _spectral('normalized ASD', rw, cw, rate,
                                min(frame, n), settings['band'],
                                settings['asd_db'], settings['asd_percent'],
                                True)]
        else:
            # untapered: a taper over a mark drawn tight on the event
            # weights down its largest cycles in both traces and hides
            # a miss in their ratio (2026-10-16 review)
            checks = [*_srs_checks(rs, cs, rate, settings['band'],
                                   settings['srs_db'], settings['q'],
                                   settings['per_octave']),
                      _peak_check(rs, cs, first, rate, settings['peaks'],
                                  settings['peak_percent'])]
        results.append(SegmentResult(
            segment, label, first, n, share, error, tuple(checks),
            _verdict([check.passed for check in checks])))
    return SegmentedReplication(
        tuple(results), whole, rate, int(r.size),
        _verdict([result.passed for result in results]), settings)

aligned_traces ¶

aligned_traces(measured: TimeHistory, specification: DataArray, channel: str | None = None, lag: int | None = None, playing: int = 0) -> tuple[ndarray, ndarray, ndarray, str, int]

(time, reference, control, channel, lag): one channel of a run and of the reference it was controlled to, aligned, over one playing.

The reference is the specification's record at channel (its first DOF by default), the control the run's record at the same DOF and dimension. The two are aligned by replication.lag_of's one shift, unless lag is given, and the control is cut to the playing'th whole playing of the reference (replication.playings): for a reference as long as the run, the one playing there is. time is the reference's own clock, from zero, in seconds. Values are as stored (SI).

Raises:

Type	Description
`ValueError`	When the reference names no such channel, the run has no record at it, or the run does not hold that playing.

Source code in src/visualdynamics/core/replication_segments.py

def aligned_traces(measured: TimeHistory, specification: DataArray,
                   channel: str | None = None, lag: int | None = None,
                   playing: int = 0
                   ) -> tuple[np.ndarray, np.ndarray, np.ndarray, str, int]:
    """(time, reference, control, channel, lag): one channel of a run
    and of the reference it was controlled to, aligned, over one playing.

    The reference is the specification's record at `channel` (its first
    DOF by default), the control the run's record at the same DOF and
    dimension. The two are aligned by `replication.lag_of`'s one shift,
    unless `lag` is given, and the control is cut to the `playing`'th
    whole playing of the reference (`replication.playings`): for a
    reference as long as the run, the one playing there is. `time` is
    the reference's own clock, from zero, in seconds. Values are as
    stored (SI).

    Raises
    ------
    ValueError
        When the reference names no such channel, the run has no record
        at it, or the run does not hold that playing.
    """
    from . import replication as rep

    dofs = [str(dof) for dof in specification.response_dof]
    dof = dofs[0] if channel is None else str(channel)
    if dof not in dofs:
        raise ValueError(f'the reference names no channel {dof}')
    index = dofs.index(dof)
    dim = specification.ordinate_dim[index]
    (row,) = rep._rows_by_dof(measured, [dof], [dim])
    if row is None:
        raise ValueError(f'the run has no {dim} record at {dof}')
    averaging = rep.averaging_for(measured, specification)
    if averaging is None:
        raise ValueError('the run is shorter than one playing of the '
                         'reference')
    if lag is None:
        lag = rep.lag_of(measured, specification, averaging)
    found = [p for p in rep.playings(measured, specification, averaging, lag)
             if p['frame'] == int(playing)]
    if not found:
        raise ValueError(f'the run holds no whole playing {int(playing) + 1} '
                         'of the reference')
    start, n = found[0]['start'], found[0]['samples']
    reference = np.real(np.asarray(specification.ordinate[index][:n],
                                   dtype=complex)).astype(float)
    control = np.real(np.asarray(measured.ordinate[row, start:start + n],
                                 dtype=complex)).astype(float)
    time = np.arange(n) / float(specification.sample_rate)
    return time, reference, control, dof, int(lag)

replication_segments ¶

replication_segments(measured: TimeHistory, specification: DataArray, segments: Sequence[Any], *, channel: str | None = None, lag: int | None = None, playing: int = 0, **settings: Any) -> SegmentedReplication

A replicated run judged segment by segment against the reference it was controlled to.

reference = TransientSpecification(...)   # the field record
result = replication_segments(run, reference, [
    (6.0, 30.0, 'stationary', 'rough road'),
    (31.9, 32.7, 'shock', 'pothole'),
    (41.0, 51.0, 'transient', 'braking')], band=(5.0, 400.0))
print(result.describe())
for segment in result.segments:
    print(segment.describe())

replication.compare reads the run as a whole, which hides a short event's miss; this reads each marked segment by the rule for its kind (judge_segments) and reports the whole-record rms error beside them. The two traces are taken at channel (the reference's first DOF by default), over the playing'th whole playing of the reference in the run, aligned by replication.lag_of unless lag is given (aligned_traces). Segment times are seconds on the reference's clock, from its first sample.

Parameters:

Name	Type	Description	Default
`measured`	`TimeHistory`	The run.	required
`specification`	`TransientSpecification or TimeHistory`	The reference it was controlled to.	required
`segments`	`sequence`	The marks, each a `Segment` or (start, end, kind[, name]).	required
`channel`	`str \| None`	As for `aligned_traces`.	`None`
`lag`	`str \| None`	As for `aligned_traces`.	`None`
`playing`	`str \| None`	As for `aligned_traces`.	`None`
`**settings`	`Any`	The tolerances, as `judge_segments` takes them.	`{}`

Returns:

Type	Description
`SegmentedReplication`	With `channel` and `lag` set.

Source code in src/visualdynamics/core/replication_segments.py

def replication_segments(measured: TimeHistory, specification: DataArray,
                         segments: Sequence[Any], *,
                         channel: str | None = None, lag: int | None = None,
                         playing: int = 0,
                         **settings: Any) -> SegmentedReplication:
    """A replicated run judged segment by segment against the reference
    it was controlled to.

        reference = TransientSpecification(...)   # the field record
        result = replication_segments(run, reference, [
            (6.0, 30.0, 'stationary', 'rough road'),
            (31.9, 32.7, 'shock', 'pothole'),
            (41.0, 51.0, 'transient', 'braking')], band=(5.0, 400.0))
        print(result.describe())
        for segment in result.segments:
            print(segment.describe())

    `replication.compare` reads the run as a whole, which hides a short
    event's miss; this reads each marked segment by the rule for its
    kind (`judge_segments`) and reports the whole-record rms error
    beside them. The two traces are taken at `channel` (the reference's
    first DOF by default), over the `playing`'th whole playing of the
    reference in the run, aligned by `replication.lag_of` unless `lag`
    is given (`aligned_traces`). Segment times are seconds on the
    reference's clock, from its first sample.

    Parameters
    ----------
    measured : TimeHistory
        The run.
    specification : TransientSpecification or TimeHistory
        The reference it was controlled to.
    segments : sequence
        The marks, each a `Segment` or (start, end, kind[, name]).
    channel, lag, playing
        As for `aligned_traces`.
    **settings
        The tolerances, as `judge_segments` takes them.

    Returns
    -------
    SegmentedReplication
        With `channel` and `lag` set.
    """
    _time, reference, control, dof, shift = aligned_traces(
        measured, specification, channel, lag, playing)
    result = judge_segments(reference, control,
                            float(specification.sample_rate), segments,
                            **settings)
    return replace(result, channel=dof, lag=shift)

visualdynamics.core.replication_segments¶

replication_segments ¶

Classes¶

Segment ¶

SegmentCheck dataclass ¶

Attributes¶

judged property ¶

low property ¶

high property ¶

Methods:¶

describe ¶

SegmentResult dataclass ¶

Methods:¶

describe ¶

SegmentedReplication dataclass ¶

Attributes¶

failed property ¶

not_judged property ¶

Methods:¶

describe ¶

Functions:¶

band_text ¶

tapered_cosine ¶

major_peaks ¶

as_segments ¶

judge_segments ¶

aligned_traces ¶

replication_segments ¶

`visualdynamics.core.replication_segments`¶

SegmentCheck `dataclass` ¶

judged `property` ¶

low `property` ¶

high `property` ¶

SegmentResult `dataclass` ¶

SegmentedReplication `dataclass` ¶

failed `property` ¶

not_judged `property` ¶