"""Read-only regional and support-change observations of relational exchange. An explicit reference supplies real phase lifts, not a discovered equilibrium. The full centered form and phase-error vectors retain information discarded by regional means. Shared winding and transport observations add no evolution, identity certificate, reduced closure or automatic selection policy. """ from __future__ import annotations from collections.abc import Mapping, Set from dataclasses import dataclass from fractions import Fraction from itertools import islice from typing import Any import networkx as nx from .._exact_time import exact_or_represented_real, finite_represented_real from ..alias import get_attr from ..constants.aliases import ALIAS_DNFR, ALIAS_EPI, ALIAS_THETA, ALIAS_VF from ..dynamics.relational import ( RelationalExchangeField, RelationalExchangeModel, _phase_edge_storage, evaluate_relational_exchange, ) from ..utils.numeric import angle_diff from .support_transport import ( RegionalSupportBalance, RegionalSupportCut, SupportTransportReset, SupportTransportSnapshot, _from_data, observe_regional_support_balance, observe_regional_support_cut, observe_support_transport, observe_support_transport_reset, ) from .winding_certificates import WindingCertificate, certify_phase_winding __all__ = ( "RegionalRelationalWork", "RegionalExchangeBalance", "RegionalPhaseResponse", "RegionalPatternObservation", "RelationalPatternObservation", "observe_relational_pattern", "RelationalPortState", "RelationalAttachmentPort", "RelationalAttachmentObservation", "RelationalAttachmentSupplyAssessment", "observe_relational_attachment", "RelationalRelocationObservation", "observe_relational_relocation", "RelationalResetObservation", "observe_relational_reset", "RelationalCoefficientJetBounds", "bound_relational_coefficient_from_jet", "RelationalCoefficientSampleBounds", "bound_relational_coefficient_from_samples", "RelationalRateContrastBounds", "bound_relational_rate_contrast", "RelationalRateSampleBounds", "bound_relational_rate_from_samples", "RelationalSampleJetBudget", "bound_relational_sample_jet_budget", "RelationalJetSampleBounds", "bound_relational_jet_from_samples", ) @dataclass(frozen=True) class RelationalCoefficientJetBounds: """Conditional chi bounds from declared initial form-response intervals.""" form_bounds: tuple[Fraction, Fraction] rate_bounds: tuple[Fraction, Fraction] acceleration_bounds: tuple[Fraction, Fraction] squared_rate_bounds: tuple[Fraction, Fraction] restoring_gap_bounds: tuple[Fraction, Fraction] coefficient_bounds: tuple[Fraction, Fraction] | None unavailable_reasons: tuple[str, ...] arithmetic_method: str scope: tuple[str, ...] = ( "conditional_two_channel_relational_consensus_preparation", "one_nonzero_spatial_mode_with_initial_phase_consensus", "declared_value_rate_acceleration_intervals_in_one_gain_and_affine_clock", "rational_outward_bounds_with_mathematical_pi_enclosure", "no_graph_read_preparation_authentication_derivative_estimation_or_evolution", "no_automatic_precision_acceptance_physical_bridge_or_universal_coefficient", ) @dataclass(frozen=True) class RelationalCoefficientSampleBounds: """Three-sample jet bounds conditional on a supplied smoothness/noise budget.""" samples: tuple[Fraction, Fraction, Fraction] sample_step: Fraction sample_error_bound: Fraction third_derivative_bound: Fraction rate_estimate: Fraction acceleration_estimate: Fraction rate_error_bound: Fraction acceleration_error_bound: Fraction jet: RelationalCoefficientJetBounds scope: tuple[str, ...] = ( "samples_at_relative_times_zero_h_two_h_in_one_affine_clock", "supplied_uniform_sample_error_and_whole_window_C3_bound", "exact_rational_stencils_and_shared_outward_jet_enclosure", "no_source_preparation_clock_noise_or_smoothness_authentication", "no_graph_read_evolution_fit_or_physical_admission", ) @dataclass(frozen=True) class RelationalRateSampleBounds: """Initial rate from three samples and an independently supplied C3 budget.""" samples: tuple[Fraction, Fraction, Fraction] sample_step: Fraction sample_error_bound: Fraction third_derivative_bound: Fraction rate_estimate: Fraction rate_error_bound: Fraction rate_bounds: tuple[Fraction, Fraction] scope: tuple[str, ...] = ( "samples_at_relative_times_zero_h_two_h_in_one_affine_clock", "supplied_uniform_sample_error_and_whole_window_C3_bound", "exact_rational_forward_rate_stencil_and_error_enclosure", "no_source_preparation_phase_lift_clock_noise_or_smoothness_authentication", "no_graph_read_evolution_fit_or_constitutive_selection", ) @dataclass(frozen=True) class RelationalSampleJetBudget: """Sample-independent error bounds for a three-sample initial jet. The nominal sampling clock is fixed. Timestamp uncertainty is bounded jitter in that clock, not an inferred conversion to another clock. A supplied speed bound controls its effect on sampled values. The complete evidence window includes both the nominal and possible actual times. """ observation_time: Fraction sample_times: tuple[Fraction, Fraction, Fraction] sample_step: Fraction sample_error_bounds: tuple[Fraction, Fraction, Fraction] timestamp_error_bounds: tuple[Fraction, Fraction, Fraction] third_derivative_bound: Fraction first_derivative_bound: Fraction | None evidence_window: tuple[Fraction, Fraction] timing_value_error_bounds: tuple[Fraction, Fraction, Fraction] total_sample_error_bounds: tuple[Fraction, Fraction, Fraction] value_error_bound: Fraction rate_sample_error_bound: Fraction rate_timing_error_bound: Fraction rate_truncation_error_bound: Fraction rate_error_bound: Fraction acceleration_sample_error_bound: Fraction acceleration_timing_error_bound: Fraction acceleration_truncation_error_bound: Fraction acceleration_error_bound: Fraction scope: tuple[str, ...] = ( "nominal_samples_at_observation_time_plus_zero_h_two_h_in_one_clock", "separately_supplied_per_sample_value_and_timestamp_error_bounds", "supplied_C3_bound_on_nominal_window_and_speed_bound_on_enlarged_window", "first_and_second_derivatives_at_the_first_nominal_sample_time", "correlated_errors_conservatively_enclosed_not_statistically_independent", "angular_samples_require_a_supplied_continuous_real_lift", "nonzero_initial_value_error_cannot_be_replaced_by_an_exact_state", "no_sample_source_clock_lift_or_smoothness_authentication", "no_graph_read_response_evaluation_fit_or_constitutive_selection", ) @dataclass(frozen=True) class RelationalJetSampleBounds: """Joint conservative value/rate/acceleration boxes from three samples. Sharing a budget retains the uncertainty of the same measured samples. The Cartesian boxes do not assert independent errors or the existence of a model trajectory reproducing those samples. """ samples: tuple[Fraction, Fraction, Fraction] budget: RelationalSampleJetBudget rate_estimate: Fraction acceleration_estimate: Fraction value_bounds: tuple[Fraction, Fraction] rate_bounds: tuple[Fraction, Fraction] acceleration_bounds: tuple[Fraction, Fraction] scope: tuple[str, ...] = ( "exact_rational_forward_stencils_with_shared_sample_budget", "joint_conservative_enclosures_not_a_sample_compatibility_certificate", "supplied_continuous_phase_lift_without_automatic_unwrapping", "initial_visible_state_uncertainty_remains_in_value_bounds", "no_hidden_state_reconstruction_or_future_response_evaluation", ) @dataclass(frozen=True) class RelationalRateContrastBounds: """Conditional relative change of two declared rate observations. The baseline may have either sign but its interval must exclude zero. Equal intervals do not imply equal unknown rates. No correlation between their uncertainties is assumed; shared gain/clock cancellation requires the same nonzero multiplicative factor for both actual observations. """ before_bounds: tuple[Fraction, Fraction] after_bounds: tuple[Fraction, Fraction] change_bounds: tuple[Fraction, Fraction] normalized_change_bounds: tuple[Fraction, Fraction] | None unavailable_reasons: tuple[str, ...] arithmetic_method: str scope: tuple[str, ...] = ( "two_declared_rate_intervals_in_one_gain_and_clock_rate_convention", "normalized_change_is_after_over_before_minus_one", "either_baseline_sign_with_zero_separation_required", "rational_outward_bounds_without_assumed_uncertainty_correlation", "no_graph_read_preparation_authentication_derivative_estimation_or_evolution", "no_automatic_precision_acceptance_constitutive_selection_or_physical_bridge", ) @dataclass(frozen=True) class RegionalRelationalWork: """Sums of the captured nodal gradient-work contributions in one region. Positive exchange transfers phase storage toward form storage. These are contributions to the global storage derivative, not derivatives of an independently defined regional energy. Overlapping regions double count shared nodes; only a partition reproduces the global sums. """ dissipation: Fraction exchange: Fraction form_work: Fraction phase_work: Fraction form_residual: Fraction phase_residual: Fraction balance_residual: Fraction @dataclass(frozen=True) class RegionalExchangeBalance: """Two actual weighted rates using one outward form cut on full support. Form weights are full-graph degree/capacity; phase weights are the captured phase metric/capacity. The latter rate is not the derivative of a weighted phase total: its metric varies with phase. Exact arithmetic on represented rates retains phase and pressure/rounding defects rather than setting them to their ideal zero values. A zero capacity inside the region makes the divided rates unavailable, while the cut and undivided model terms remain defined. No evolution, reduced closure or provenance seal is supplied. """ cut: RegionalSupportCut form_weighted_rate: Fraction | None form_boundary_rate: Fraction form_source_rate: Fraction form_pressure_defect_rate: Fraction form_rounding_defect_rate: Fraction | None form_identity_residual: Fraction | None phase_weighted_rate: Fraction | None phase_boundary_rate: Fraction phase_rate_residual: Fraction | None weighted_rate_unavailable_reason: str | None @dataclass(frozen=True) class RegionalPhaseResponse: """Unweighted phase-rate balance from the captured phase mobility and form. Mobility is capacity divided by the positive phase metric. Population covariance retains its correlation with the exact form gradient; it need not vanish when the outward cut vanishes. The squared covariance-rate bound is Cauchy--Schwarz, not a calibrated threshold. Actual model rates retain their rounding residual. This instantaneous identity supplies no autonomous regional closure, measured derivative or monotone clock. """ mean_mobility: Fraction mean_form_gradient: Fraction mobility_variance: Fraction form_gradient_variance: Fraction mobility_gradient_covariance: Fraction mean_mobility_boundary_rate: Fraction covariance_rate: Fraction covariance_rate_squared_bound: Fraction model_total_rate: Fraction rounding_residual: Fraction total_rate: Fraction mean_rate: Fraction identity_residual: Fraction @dataclass(frozen=True) class RegionalPatternObservation: """One supplied region's complete centered coordinates and separate offsets. Means are arithmetic means, whereas ``transport.mean`` uses the shared degree/capacity metric. The two squared norms have their respective form and phase units; no combined metric or recovery threshold is introduced. """ nodes: tuple[Any, ...] form_mean: Fraction phase_error_mean: Fraction centered_form: tuple[Fraction, ...] centered_phase_error: tuple[Fraction, ...] form_norm_squared: Fraction phase_norm_squared: Fraction transport: RegionalSupportBalance | None transport_unavailable_reason: str | None work: RegionalRelationalWork | None = None boundary: RegionalExchangeBalance | None = None phase_response: RegionalPhaseResponse | None = None @dataclass(frozen=True) class RelationalPatternObservation: """Detached fresh field and observations in caller-supplied regional frames. ``reference_phase`` follows ``field.nodes`` and contains materialized real lifts. Subtracting them does not wrap angles or infer a common semicircle. A common change of phase offset disappears on centering; independently changing node representatives by full turns changes the supplied frame. """ field: RelationalExchangeField reference_phase: tuple[float, ...] regions: tuple[RegionalPatternObservation, ...] winding: tuple[WindingCertificate, ...] scope: tuple[str, ...] = ( "one_fresh_detached_relational_field_without_live_graph_writes", "supplied_real_reference_lifts_and_ordered_possibly_overlapping_regions", "exact_rational_centering_of_materialized_form_and_phase_coordinates", "separate_form_and_phase_units_without_a_combined_norm_or_threshold", "regional_transport_uses_full_support_and_independent_phase_source", "nodal_and_regional_work_retains_exact_represented_gradients_and_rate_defects", "paired_weighted_rates_share_one_cut_not_a_conserved_phase_total", "unweighted_phase_response_retains_mobility_form_covariance_and_rounding", "declared_cycle_winding_is_snapshot_geometry_not_temporal_identity", "no_reference_equilibrium_recovery_formation_closed_reduction_or_selection_certificate", ) @dataclass(frozen=True) class RelationalPortState: """One supplied port's captured state and instantaneous local interface. ``form_gradient`` is exact represented ``q=B*x``; ``relative_resultant`` contains the engine's materialized relative cosine/sine sums. Degree, gradient and resultant refer to the field's complete support. These summaries do not supply their own future evolution or a closed macro-node. """ node: Any epi: float phase: float capacity: float degree: int form_gradient: Fraction relative_resultant: tuple[float, float] pressure: float phase_metric: float form_rate: float phase_rate: float @dataclass(frozen=True) class RelationalAttachmentPort: """Captured port before and after a hypothetical unit-support change. The original attachment name is retained for compatibility; relocation observations reuse this same before/after card. """ before: RelationalPortState after: RelationalPortState @dataclass(frozen=True) class RelationalAttachmentSupplyAssessment: """Declared work compared with one represented hypothetical storage jump. The additional event-passivity premise is ``storage_change <= supplied_work``. Required supply uses the support comparison's captured storage difference, not a new physical energy or a bound on ideal trigonometric storage. Signed work may describe supply or extraction; its source is supplied by the caller. A passing balance neither authenticates that work nor selects an event. """ required_supply: Fraction supplied_work: Fraction supply_margin: Fraction represented_balance_satisfied: bool scope: tuple[str, ...] = ( "caller_supplied_work_not_authenticated", "passivity_is_an_additional_event_premise", "represented_storage_not_an_ideal_trigonometric_certificate", "no_occurrence_timing_or_live_support_selection", ) class _RelationalSupportBudget: """Shared declared-work accounting for captured support comparisons.""" storage_change: Fraction @property def represented_zero_supply_passive(self) -> bool: """Whether captured storage satisfies the additional zero-work premise.""" return self.storage_change <= 0 def assess_supply(self, supplied_work: Any) -> RelationalAttachmentSupplyAssessment: """Assess signed declared work without evaluating or modifying any graph. Positive work is supply and negative work is extraction, in the same structural-storage units as ``storage_change``. Rational inputs remain exact; other real inputs follow the shared represented-real boundary. No default work, occurrence rule, elapsed-time credit or continuous-loss funding is inferred. This is arithmetic on the report, not proof of its provenance or of exact-real passivity from rounded phase data. """ work = exact_or_represented_real(supplied_work, "supplied_work") margin = work - self.storage_change return RelationalAttachmentSupplyAssessment( required_supply=self.storage_change, supplied_work=work, supply_margin=margin, represented_balance_satisfied=margin >= 0, ) @dataclass(frozen=True) class RelationalAttachmentObservation(_RelationalSupportBudget): """Compare two admitted components with their hypothetical joined field. Full component states remain available in ``components``. Change tuples follow ``joined.nodes`` and subtract the matching component's captured value from the joined value using exact represented arithmetic. Storage changes subtract the sum of both components; ``phase_storage_change`` is the unscaled cosine cost V, and total storage includes the model's beta. ``cut`` is directed outward from the complete left component. The shared ``transport_reset`` accounts for form storage only. None of these values records an executed support event, a trajectory or a capture certificate. """ components: tuple[RelationalExchangeField, RelationalExchangeField] joined: RelationalExchangeField bridge: tuple[Any, Any] ports: tuple[RelationalAttachmentPort, RelationalAttachmentPort] form_rate_change: tuple[Fraction, ...] phase_rate_change: tuple[Fraction, ...] pressure_change: tuple[Fraction, ...] phase_metric_change: tuple[Fraction, ...] form_storage_change: Fraction phase_storage_change: Fraction storage_change: Fraction transport_reset: SupportTransportReset cut: RegionalSupportCut scope: tuple[str, ...] = ( "two_disjoint_separately_admitted_connected_components", "supplied_ordered_unit_bridge_left_port_to_right_port", "same_selected_phase_domain_with_unchanged_form_phase_and_held_capacity", "three_fresh_native_fields_without_live_graph_writes", "exact_represented_differences_not_transcendental_error_bounds", "hypothetical_support_comparison_not_an_executed_event", "no_autonomous_attachment_reduced_closure_or_recovery_certificate", ) @property def continuous_loss_change(self) -> Fraction: """Joined minus component loss rates, not available event work.""" return self.joined.continuous_loss - sum( (field.continuous_loss for field in self.components), Fraction(0) ) @dataclass(frozen=True) class RelationalRelocationObservation(_RelationalSupportBudget): """Two fresh fields for one supplied state-preserving bridge relocation. ``components`` are ordered node partitions after removing the old bridge, not separately evaluated fields. Their internal edges remain unchanged. Both cuts point outward from the first partition. Change tuples follow ``after.nodes``; ports contain each affected node once, in that order. Phase storage changes are unscaled cosine costs; total storage includes beta. A passive represented budget alone does not certify future identity. """ before: RelationalExchangeField after: RelationalExchangeField components: tuple[tuple[Any, ...], tuple[Any, ...]] remove_bridge: tuple[Any, Any] add_bridge: tuple[Any, Any] ports: tuple[RelationalAttachmentPort, ...] form_rate_change: tuple[Fraction, ...] phase_rate_change: tuple[Fraction, ...] pressure_change: tuple[Fraction, ...] phase_metric_change: tuple[Fraction, ...] form_storage_change: Fraction phase_storage_change: Fraction storage_change: Fraction transport_reset: SupportTransportReset cut_before: RegionalSupportCut cut_after: RegionalSupportCut scope: tuple[str, ...] = ( "supplied_graph_bridge_separates_two_nontrivial_connected_components", "supplied_missing_unit_edge_crosses_the_same_ordered_components", "all_internal_edges_and_primitive_form_phase_capacity_are_preserved", "same_selected_phase_domain_with_two_fresh_fields_without_live_graph_writes", "exact_represented_differences_not_transcendental_error_bounds", "hypothetical_support_comparison_not_an_executed_event", "no_autonomous_selection_timing_reduced_closure_or_recovery_certificate", ) @property def continuous_loss_change(self) -> Fraction: """New minus old loss rate, not available event work.""" return self.after.continuous_loss - self.before.continuous_loss @dataclass(frozen=True) class RelationalResetObservation(_RelationalSupportBudget): """Joint state/support storage accounting between two supplied snapshots. Form storage uses conductance; phase storage uses every bare support edge, including zero-conductance edges. The decomposition first changes state on the old support, then changes support at the new state. Phase terms are unscaled V; total storage multiplies them by the supplied positive beta. ``transport_reset.before`` retains the new state on the old support. Capacity and stored pressure use the shared transport snapshot, including its zero defaults when absent; those defaults are not measured zeros. Snapshot comparison does not authenticate an executed operator or certify that either endpoint admits the conditional unit-support relational flow. """ before: SupportTransportSnapshot after: SupportTransportSnapshot phase_before: tuple[float, ...] phase_after: tuple[float, ...] edges_before: tuple[tuple[Any, Any], ...] edges_after: tuple[tuple[Any, Any], ...] storage_scale: float form_state_change: Fraction form_support_change: Fraction form_storage_change: Fraction phase_state_change: Fraction phase_support_change: Fraction phase_storage_change: Fraction phase_storage_before: Fraction phase_storage_after: Fraction storage_before: Fraction storage_after: Fraction storage_change: Fraction identity_residual: Fraction transport_reset: SupportTransportReset scope: tuple[str, ...] = ( "supplied_nonempty_same_ordered_nodes_simple_undirected_support", "symmetric_nonnegative_conductance_with_disconnected_and_zero_weight_support_allowed", "weighted_form_storage_and_unweighted_bare_support_phase_storage", "state_change_on_old_support_then_support_change_at_new_state", "raw_phase_admission_and_shared_represented_half_sine_cost", "exact_represented_accounting_not_an_ideal_trigonometric_bound", "detached_stored_state_without_pressure_refresh_or_live_writes", "transport_snapshot_defaults_absent_capacity_and_stored_pressure_to_zero", "no_event_authentication_selection_clock_or_continuation_certificate", ) def observe_relational_reset(before_graph, after_graph, *, storage_scale): """Observe the full storage budget of a supplied joint state/support reset. Both graphs must have the same nonempty node order and simple undirected loop-free support. Shared transport admission validates scalar form, nonnegative capacity, stored pressure and symmetric nonnegative weights; absent capacity/stored pressure retain that owner's zero defaults. Explicit finite phase values are required at every node, including isolates. Graphs may be disconnected and have zero or nonunit conductances; no acute chamber, pressure law, held capacity or continuation admission is asserted. Storage is ``E_D + storage_scale * V`` with ``V`` the sum of the shared represented half-sine cost on every support edge. Raw differences retain finite-output admission before the shared signed atan2 chart is selected; binary64 tau is not substituted for the trigonometric period. The scale is materialized through the same boundary as the relational model. The result separates simultaneous nodal reorganization from support work without crediting earlier continuous dissipation or introducing an event law. """ beta, beta_q = finite_represented_real(storage_scale, "storage_scale") if beta <= 0: raise ValueError("storage_scale must be positive") for graph in (before_graph, after_graph): if not isinstance(graph, nx.Graph): raise TypeError("reset endpoints must be networkx graphs") if ( graph.is_directed() or graph.is_multigraph() or nx.number_of_selfloops(graph) ): raise ValueError("reset support must be simple undirected without loops") nodes = tuple(before_graph) if not nodes or nodes != tuple(after_graph): raise ValueError("reset endpoints require the same nonempty ordered nodes") def phases(graph): return tuple( finite_represented_real( get_attr( graph.nodes[node], ALIAS_THETA, None, conv=lambda value: value, strict=True, ), "phase", )[0] for node in nodes ) phase_before, phase_after = phases(before_graph), phases(after_graph) before = observe_support_transport(before_graph) after = observe_support_transport(after_graph) intermediate = _from_data( nodes, before.conductance, before.support_neighbors, after.epi, after.capacity, after.stored_pressure, ) transport_reset = observe_support_transport_reset(intermediate, after) old_edges = tuple( (i, j) for i, row in enumerate(before.support_neighbors) for j in row if i < j ) new_edges = tuple( (i, j) for i, row in enumerate(after.support_neighbors) for j in row if i < j ) def phase_cost(phase, edges): total = Fraction(0) for i, j in edges: difference = finite_represented_real( Fraction(phase[j]) - Fraction(phase[i]), "phase difference" )[0] total += _phase_edge_storage(angle_diff(difference, 0.0)) return total phase_old = phase_cost(phase_before, old_edges) phase_intermediate = phase_cost(phase_after, old_edges) phase_new = phase_cost(phase_after, new_edges) form_state = intermediate.dirichlet_energy - before.dirichlet_energy form_support = transport_reset.energy_change phase_state = phase_intermediate - phase_old phase_support = phase_new - phase_intermediate storage_before = before.dirichlet_energy + beta_q * phase_old storage_after = after.dirichlet_energy + beta_q * phase_new storage_change = storage_after - storage_before return RelationalResetObservation( before=before, after=after, phase_before=phase_before, phase_after=phase_after, edges_before=tuple((nodes[i], nodes[j]) for i, j in old_edges), edges_after=tuple((nodes[i], nodes[j]) for i, j in new_edges), storage_scale=beta, form_state_change=form_state, form_support_change=form_support, form_storage_change=after.dirichlet_energy - before.dirichlet_energy, phase_state_change=phase_state, phase_support_change=phase_support, phase_storage_change=phase_new - phase_old, phase_storage_before=phase_old, phase_storage_after=phase_new, storage_before=storage_before, storage_after=storage_after, storage_change=storage_change, identity_residual=( storage_change - form_state - form_support - beta_q * (phase_state + phase_support) ), transport_reset=transport_reset, ) def _ordered(value, label, *, limit=None): if isinstance(value, (str, bytes, bytearray, Mapping, Set)): raise TypeError(f"{label} must be an ordered iterable") try: return tuple(value) if limit is None else tuple(islice(iter(value), limit)) except TypeError as exc: raise TypeError(f"{label} must be an ordered iterable") from exc def _interval(raw, label): """Admit an ordered real endpoint pair into shared outward arithmetic.""" from ..mathematics._rational_interval import I values = _ordered(raw, label, limit=3) if len(values) != 2: raise ValueError(f"{label} must contain two ordered endpoints") return I(*(exact_or_represented_real(value, label) for value in values)) def bound_relational_rate_contrast( *, before_bounds, after_bounds ) -> RelationalRateContrastBounds: """Enclose (after-before)/before from supplied rate uncertainty intervals. Compute after/before-1 to avoid duplicating the uncertain baseline in the numerator and denominator. Intervals include outward arithmetic, not any missing measurement, preparation, derivative or clock error. The same nonzero gain and clock-rate factor cancels between both observations; different factors or additive rate offsets do not generally cancel. An unresolved baseline returns no normalized change, while retaining the absolute change. Neither a graph nor a law or precision target is selected. """ from ..mathematics._rational_interval import INTERVAL_METHOD before = _interval(before_bounds, "before_bounds") after = _interval(after_bounds, "after_bounds") change = after - before reasons = ("baseline_rate_not_separated_from_zero",) if before.contains(0) else () normalized = None if reasons else after / before - 1 return RelationalRateContrastBounds( before_bounds=(before.lo, before.hi), after_bounds=(after.lo, after.hi), change_bounds=(change.lo, change.hi), normalized_change_bounds=( (normalized.lo, normalized.hi) if normalized is not None else None ), unavailable_reasons=reasons, arithmetic_method=INTERVAL_METHOD, ) def bound_relational_coefficient_from_jet( *, form_bounds, rate_bounds, acceleration_bounds ) -> RelationalCoefficientJetBounds: """Enclose chi=m1²/[pi²*(m1²-m0*m2)] under declared preparation premises. Each input is an ordered pair of finite exact/represented real endpoints. Bounds include outward arithmetic, not unprovided measurement or derivative error. Unresolved signal/gap or incompatible decay returns no coefficient. This function neither validates a physical preparation nor reads a graph. """ from ..mathematics._rational_interval import INTERVAL_METHOD, pi_interval form = _interval(form_bounds, "form_bounds") rate = _interval(rate_bounds, "rate_bounds") acceleration = _interval(acceleration_bounds, "acceleration_bounds") square = rate**2 gap = square - form * acceleration reasons = [] if form.contains(0): reasons.append("initial_form_not_separated_from_zero") if (form.lo > 0 and rate.lo > 0) or (form.hi < 0 and rate.hi < 0): reasons.append("incompatible_initial_decay") if gap.hi <= 0: reasons.append("nonpositive_restoring_gap") elif gap.lo <= 0: reasons.append("unresolved_restoring_gap") coefficient = None if reasons else square / (pi_interval() ** 2 * gap) return RelationalCoefficientJetBounds( form_bounds=(form.lo, form.hi), rate_bounds=(rate.lo, rate.hi), acceleration_bounds=(acceleration.lo, acceleration.hi), squared_rate_bounds=(square.lo, square.hi), restoring_gap_bounds=(gap.lo, gap.hi), coefficient_bounds=( (coefficient.lo, coefficient.hi) if coefficient is not None else None ), unavailable_reasons=tuple(reasons), arithmetic_method=INTERVAL_METHOD, ) def _sample_triple(values, label, *, nonnegative=False): raw = _ordered(values, label, limit=4) if len(raw) != 3: raise ValueError(f"{label} must contain exactly three ordered values") result = tuple(exact_or_represented_real(value, label) for value in raw) if nonnegative and any(value < 0 for value in result): raise ValueError(f"{label} must be nonnegative") return result def _sample_jet_clock( *, sample_step, timestamp_error_bounds=(0, 0, 0), observation_time=0 ): """Admit the sampling clock without inventing observation or speed bounds.""" step = exact_or_represented_real(sample_step, "sample_step") at = exact_or_represented_real(observation_time, "observation_time") if step <= 0 or at < 0: raise ValueError( "require positive sample_step and nonnegative observation_time" ) jitter = _sample_triple( timestamp_error_bounds, "timestamp_error_bounds", nonnegative=True ) times = tuple(at + index * step for index in range(3)) window = ( min(time - error for time, error in zip(times, jitter)), max(time + error for time, error in zip(times, jitter)), ) if window[0] < 0: raise ValueError("the complete evidence window must be nonnegative") return step, times, jitter, window def bound_relational_sample_jet_budget( *, sample_step, sample_error_bounds, third_derivative_bound, timestamp_error_bounds=(0, 0, 0), first_derivative_bound=None, observation_time=0, ) -> RelationalSampleJetBudget: """Bound prospective derivative errors without reading any response. At the nominal times t0+j*h the supplied sample errors are e_j and the timestamp errors are tau_j. A nonzero timestamp error requires a separate bound L on |q'| throughout the enlarged evidence window. Then each nominal value error is at most e_j+L*tau_j. A supplied bound M on |q'''| over the nominal window bounds the forward-stencil remainders by M*h**2/3 and M*h. Neither derivative bound can be obtained from three samples alone. For phase data the bounds and samples refer to the same continuous real lift; no lift is inferred. Bounds on jitter do not admit an unknown clock scale. """ step, times, jitter, window = _sample_jet_clock( sample_step=sample_step, timestamp_error_bounds=timestamp_error_bounds, observation_time=observation_time, ) third = exact_or_represented_real(third_derivative_bound, "third_derivative_bound") if third < 0: raise ValueError("third_derivative_bound must be nonnegative") errors = _sample_triple( sample_error_bounds, "sample_error_bounds", nonnegative=True ) speed = ( None if first_derivative_bound is None else exact_or_represented_real(first_derivative_bound, "first_derivative_bound") ) if speed is not None and speed < 0: raise ValueError("first_derivative_bound must be nonnegative") if any(jitter) and speed is None: raise ValueError("nonzero timestamp error requires first_derivative_bound") timing = tuple((speed or Fraction(0)) * error for error in jitter) effective = tuple(error + shift for error, shift in zip(errors, timing)) def rate_error(values): return (3 * values[0] + 4 * values[1] + values[2]) / (2 * step) def acceleration_error(values): return (values[0] + 2 * values[1] + values[2]) / step**2 rate_sample, rate_timing = rate_error(errors), rate_error(timing) accel_sample, accel_timing = acceleration_error(errors), acceleration_error(timing) rate_truncation, accel_truncation = third * step**2 / 3, third * step return RelationalSampleJetBudget( observation_time=times[0], sample_times=times, sample_step=step, sample_error_bounds=errors, timestamp_error_bounds=jitter, third_derivative_bound=third, first_derivative_bound=speed, evidence_window=window, timing_value_error_bounds=timing, total_sample_error_bounds=effective, value_error_bound=effective[0], rate_sample_error_bound=rate_sample, rate_timing_error_bound=rate_timing, rate_truncation_error_bound=rate_truncation, rate_error_bound=rate_sample + rate_timing + rate_truncation, acceleration_sample_error_bound=accel_sample, acceleration_timing_error_bound=accel_timing, acceleration_truncation_error_bound=accel_truncation, acceleration_error_bound=accel_sample + accel_timing + accel_truncation, ) def bound_relational_jet_from_samples( samples, *, sample_step, sample_error_bounds, third_derivative_bound, timestamp_error_bounds=(0, 0, 0), first_derivative_bound=None, observation_time=0, ) -> RelationalJetSampleBounds: """Enclose the initial value, rate and acceleration in one declared clock. The first and second derivatives use the same three observed values and independent error/smoothness declarations. Their enclosing intervals are correlated; compatibility with those boxes alone does not establish a trajectory fitting the underlying samples. A nonzero initial value bound cannot be silently substituted into an exact-state inverse or forecast. """ values = _sample_triple(samples, "samples") budget = bound_relational_sample_jet_budget( sample_step=sample_step, sample_error_bounds=sample_error_bounds, timestamp_error_bounds=timestamp_error_bounds, first_derivative_bound=first_derivative_bound, third_derivative_bound=third_derivative_bound, observation_time=observation_time, ) step = budget.sample_step rate = (-3 * values[0] + 4 * values[1] - values[2]) / (2 * step) acceleration = (values[2] - 2 * values[1] + values[0]) / step**2 return RelationalJetSampleBounds( samples=values, budget=budget, rate_estimate=rate, acceleration_estimate=acceleration, value_bounds=( values[0] - budget.value_error_bound, values[0] + budget.value_error_bound, ), rate_bounds=(rate - budget.rate_error_bound, rate + budget.rate_error_bound), acceleration_bounds=( acceleration - budget.acceleration_error_bound, acceleration + budget.acceleration_error_bound, ), ) def bound_relational_rate_from_samples( samples, *, sample_step, sample_error_bound, third_derivative_bound ) -> RelationalRateSampleBounds: """Retain the uniform-error initial-rate adapter over the shared jet owner. The uniform sample error and noiseless whole-window C3 bound are supplied independently. Angular samples require a supplied consistent real lift. This adapter retains exact uniform timing and its existing report schema. """ error = exact_or_represented_real(sample_error_bound, "sample_error_bound") observation = bound_relational_jet_from_samples( samples, sample_step=sample_step, sample_error_bounds=(error,) * 3, third_derivative_bound=third_derivative_bound, ) budget = observation.budget return RelationalRateSampleBounds( samples=observation.samples, sample_step=budget.sample_step, sample_error_bound=error, third_derivative_bound=budget.third_derivative_bound, rate_estimate=observation.rate_estimate, rate_error_bound=budget.rate_error_bound, rate_bounds=observation.rate_bounds, ) def bound_relational_coefficient_from_samples( samples, *, sample_step, sample_error_bound, third_derivative_bound ) -> RelationalCoefficientSampleBounds: """Propagate shared sample/rate admission into the conditional jet observer. The noise bound covers every sample and the C3 bound covers the complete noiseless window. Neither bound, nor uniform timing, is inferred here. """ error = exact_or_represented_real(sample_error_bound, "sample_error_bound") observation = bound_relational_jet_from_samples( samples, sample_step=sample_step, sample_error_bounds=(error,) * 3, third_derivative_bound=third_derivative_bound, ) budget = observation.budget jet = bound_relational_coefficient_from_jet( form_bounds=observation.value_bounds, rate_bounds=observation.rate_bounds, acceleration_bounds=observation.acceleration_bounds, ) return RelationalCoefficientSampleBounds( samples=observation.samples, sample_step=budget.sample_step, sample_error_bound=error, third_derivative_bound=budget.third_derivative_bound, rate_estimate=observation.rate_estimate, acceleration_estimate=observation.acceleration_estimate, rate_error_bound=budget.rate_error_bound, acceleration_error_bound=budget.acceleration_error_bound, jet=jet, ) def _regions(nodes, regions): raw = _ordered(regions, "regions") if not raw: raise ValueError("regions must contain at least one region") lookup = {node: index for index, node in enumerate(nodes)} result = [] for region in raw: row = _ordered(region, "region") if not row: raise ValueError("each region must be nonempty") try: indices = tuple(lookup[node] for node in row) except (TypeError, KeyError) as exc: raise ValueError("region nodes must belong to the full support") from exc if len(set(indices)) != len(indices): raise ValueError("region nodes must be distinct") result.append(indices) return tuple(result) def _detached_graph(field): """Rebuild only the captured state consumed by the existing observers.""" graph = nx.Graph() for i, node in enumerate(field.nodes): graph.add_node( node, **{ ALIAS_EPI[0]: field.epi[i], ALIAS_VF[0]: field.capacity[i], ALIAS_THETA[0]: field.phase[i], ALIAS_DNFR[0]: field.pressure[i], }, ) graph.add_edges_from((a, b, {"weight": 1.0}) for a, b in field.edges) return graph def _port_state(field, node): if field.work is None or field.relative_resultant is None: raise RuntimeError("fresh relational field must contain port evidence") index = field.nodes.index(node) return RelationalPortState( node=node, epi=field.epi[index], phase=field.phase[index], capacity=field.capacity[index], degree=sum(node == a or node == b for a, b in field.edges), form_gradient=field.work.form_gradient[index], relative_resultant=field.relative_resultant[index], pressure=field.pressure[index], phase_metric=field.phase_metric[index], form_rate=field.form_rate[index], phase_rate=field.phase_rate[index], ) def _field_changes(before_fields, after): """Exact represented differences against one or several disjoint fields.""" origins = { node: (field, index) for field in before_fields for index, node in enumerate(field.nodes) } changes = { name + "_change": tuple( Fraction(value) - Fraction(getattr(origins[node][0], name)[origins[node][1]]) for node, value in zip(after.nodes, getattr(after, name), strict=True) ) for name in ("form_rate", "phase_rate", "pressure", "phase_metric") } changes.update( { name + "_change": getattr(after, name) - sum((getattr(field, name) for field in before_fields), Fraction(0)) for name in ("form_storage", "phase_storage", "storage") } ) return changes def observe_relational_attachment(left, right, *, model, bridge): """Evaluate one supplied unit bridge on detached admitted component states. ``left`` and ``right`` must each satisfy the native relational graph/state contract, with disjoint node labels. ``bridge`` is an ordered pair naming one left port and one right port. Components and joined field must each independently pass the selected model's phase-domain admission. Invalid components or an inadmissible joined field raise without changing either graph. Zero capacities retain the native frozen-row semantics. Admission of the disconnected components need not imply admission after joining. Both components are evaluated separately. Their disconnected union is used only for shared transport accounting, never as a relational field. A fresh native evaluation on the joined detached state supplies pressure, phase mobility and both rates. No support event is committed and no new pressure, phase or autonomous connection law is introduced. """ if not isinstance(model, RelationalExchangeModel): raise TypeError("model must be a RelationalExchangeModel") endpoints = _ordered(bridge, "bridge", limit=3) if len(endpoints) != 2: raise ValueError("bridge must contain exactly two ordered ports") components = ( evaluate_relational_exchange(left, model=model), evaluate_relational_exchange(right, model=model), ) left_nodes, right_nodes = (set(field.nodes) for field in components) if left_nodes & right_nodes: raise ValueError("component node labels must be disjoint") try: valid_ports = endpoints[0] in left_nodes and endpoints[1] in right_nodes except TypeError as exc: raise ValueError( "bridge ports must belong to their respective components" ) from exc if not valid_ports: raise ValueError("bridge ports must belong to their respective components") detached = nx.compose(*(_detached_graph(field) for field in components)) # The admitted model has no forcing. Preserve that explicit declaration # when reconstructing only its consumed state on the detached support. detached.graph["GAMMA"] = {"type": "none"} before_transport = observe_support_transport(detached) detached.add_edge(*endpoints, weight=1.0) joined = evaluate_relational_exchange(detached, model=model) after_transport = observe_support_transport(_detached_graph(joined)) reset = observe_support_transport_reset(before_transport, after_transport) cut = observe_regional_support_cut(after_transport, components[0].nodes) return RelationalAttachmentObservation( components=components, joined=joined, bridge=endpoints, ports=tuple( RelationalAttachmentPort( _port_state(field, node), _port_state(joined, node) ) for field, node in zip(components, endpoints, strict=True) ), **_field_changes(components, joined), transport_reset=reset, cut=cut, ) def observe_relational_relocation(graph, *, model, remove_bridge, add_bridge): """Compare a supplied bridge relocation at unchanged primitive state. The simple unit connected input is evaluated through the native owner in the selected phase domain. Removing ``remove_bridge`` must leave two connected components with at least two nodes each. Its ordered endpoints identify the first and second components; ``add_bridge`` must join them in that same order and must be absent from the original graph. All internal component edges remain unchanged. The new connected graph independently passes the same native field admission. Zero capacities remain valid. Only detached state is edited. No field is evaluated on the disconnected intermediate support. The report can describe either sign of storage change; its optional declared-work assessment is not an event selector, time law, ideal trigonometric bound or subsequent recovery certificate. """ if not isinstance(model, RelationalExchangeModel): raise TypeError("model must be a RelationalExchangeModel") old = _ordered(remove_bridge, "remove_bridge", limit=3) new = _ordered(add_bridge, "add_bridge", limit=3) if len(old) != 2 or len(new) != 2: raise ValueError("each bridge must contain exactly two ordered ports") before = evaluate_relational_exchange(graph, model=model) detached = _detached_graph(before) detached.graph["GAMMA"] = {"type": "none"} try: valid_nodes = all(node in detached for node in (*old, *new)) except TypeError as exc: raise ValueError("bridge ports must belong to the full support") from exc if not valid_nodes: raise ValueError("bridge ports must belong to the full support") if not detached.has_edge(*old): raise ValueError("remove_bridge must be an existing edge") if detached.has_edge(*new): raise ValueError("add_bridge must be absent from the original support") before_transport = observe_support_transport(detached) detached.remove_edge(*old) parts = tuple(nx.connected_components(detached)) if len(parts) != 2 or any(len(part) < 2 for part in parts): raise ValueError("remove_bridge must separate two nontrivial components") left = next(part for part in parts if old[0] in part) right = next(part for part in parts if old[1] in part) if new[0] not in left or new[1] not in right: raise ValueError("add_bridge must cross the same ordered components") components = tuple( tuple(node for node in before.nodes if node in part) for part in (left, right) ) detached.add_edge(*new, weight=1.0) after = evaluate_relational_exchange(detached, model=model) after_transport = observe_support_transport(_detached_graph(after)) affected = set((*old, *new)) return RelationalRelocationObservation( before=before, after=after, components=components, remove_bridge=old, add_bridge=new, ports=tuple( RelationalAttachmentPort( _port_state(before, node), _port_state(after, node) ) for node in after.nodes if node in affected ), **_field_changes((before,), after), transport_reset=observe_support_transport_reset( before_transport, after_transport ), cut_before=observe_regional_support_cut(before_transport, components[0]), cut_after=observe_regional_support_cut(after_transport, components[0]), ) def _regional_work(field, indices): work = field.work if work is None: raise RuntimeError("fresh relational field must contain work accounting") return RegionalRelationalWork( **{ name: sum((getattr(work, name)[i] for i in indices), Fraction(0)) for name in RegionalRelationalWork.__dataclass_fields__ } ) def _regional_boundary(field, cut, degrees): indices = cut.region_indices current = cut.outward_cut_current e = Fraction(field.model.epi_weight) w = Fraction(field.model.phase_weight) beta = Fraction(field.model.storage_scale) source = sum( (w * degrees[i] * Fraction(field.phase_source[i]) for i in indices), Fraction(0), ) pressure_defect = sum( (degrees[i] * field.pressure_split_residual[i] for i in indices), Fraction(0), ) reason = ( "zero_capacity_in_region" if any(field.capacity[i] == 0.0 for i in indices) else None ) form_rate = phase_rate = rounding = form_residual = phase_residual = None if reason is None: form_rate = sum( ( degrees[i] * Fraction(field.form_rate[i]) / Fraction(field.capacity[i]) for i in indices ), Fraction(0), ) rounding = sum( ( degrees[i] * field.nodal_rate_rounding_defect[i] / Fraction(field.capacity[i]) for i in indices ), Fraction(0), ) phase_rate = sum( ( Fraction(field.phase_metric[i]) * Fraction(field.phase_rate[i]) / Fraction(field.capacity[i]) for i in indices ), Fraction(0), ) form_residual = form_rate + e * current - source - pressure_defect - rounding phase_residual = phase_rate - w * current / beta if form_residual: raise RuntimeError("exact represented regional form-rate identity failed") return RegionalExchangeBalance( cut=cut, form_weighted_rate=form_rate, form_boundary_rate=-e * current, form_source_rate=source, form_pressure_defect_rate=pressure_defect, form_rounding_defect_rate=rounding, form_identity_residual=form_residual, phase_weighted_rate=phase_rate, phase_boundary_rate=w * current / beta, phase_rate_residual=phase_residual, weighted_rate_unavailable_reason=reason, ) def _regional_phase_response(field, cut): if ( field.work is None or field.phase_mobility is None or field.phase_rate_rounding_defect is None ): raise RuntimeError( "fresh relational field must contain exact phase-rate evidence" ) indices = cut.region_indices n = len(indices) mobility = tuple(field.phase_mobility[i] for i in indices) gradient = tuple(field.work.form_gradient[i] for i in indices) mean_mobility = sum(mobility, Fraction(0)) / n mean_gradient = sum(gradient, Fraction(0)) / n centered_mobility = tuple(a - mean_mobility for a in mobility) centered_gradient = tuple(q - mean_gradient for q in gradient) mobility_variance = sum((a * a for a in centered_mobility), Fraction(0)) / n gradient_variance = sum((q * q for q in centered_gradient), Fraction(0)) / n covariance = ( sum((a * q for a, q in zip(centered_mobility, centered_gradient)), Fraction(0)) / n ) k = Fraction(field.model.phase_weight) / Fraction(field.model.storage_scale) boundary_rate = k * mean_mobility * cut.outward_cut_current covariance_rate = k * n * covariance squared_bound = k * k * n * n * mobility_variance * gradient_variance model_total = k * sum((a * q for a, q in zip(mobility, gradient)), Fraction(0)) rounding = sum((field.phase_rate_rounding_defect[i] for i in indices), Fraction(0)) total = sum((Fraction(field.phase_rate[i]) for i in indices), Fraction(0)) residual = total - boundary_rate - covariance_rate - rounding if residual or model_total != boundary_rate + covariance_rate: raise RuntimeError("exact represented regional phase-rate identity failed") if covariance_rate * covariance_rate > squared_bound: raise RuntimeError("exact regional covariance-rate bound failed") return RegionalPhaseResponse( mean_mobility=mean_mobility, mean_form_gradient=mean_gradient, mobility_variance=mobility_variance, form_gradient_variance=gradient_variance, mobility_gradient_covariance=covariance, mean_mobility_boundary_rate=boundary_rate, covariance_rate=covariance_rate, covariance_rate_squared_bound=squared_bound, model_total_rate=model_total, rounding_residual=rounding, total_rate=total, mean_rate=total / n, identity_residual=residual, ) def observe_relational_pattern( graph, *, model: RelationalExchangeModel, reference_phase: Mapping, regions, cycles=(), ) -> RelationalPatternObservation: """Observe supplied regions using exactly one fresh relational evaluation. Reference phases must map the exact full node support to finite represented real lifts. Regions are ordered, nonempty selections without repeated nodes; they may overlap and need not partition support. No reference equilibrium, lift compatibility over time or automatic regional discovery is inferred. Transport is unavailable if any full-support capacity is zero, EPI weight is zero, or the region covers all support (in that order of precedence). Those are limitations of the existing regional transport contract, not failures of the retained geometry. Its forcing is independently supplied as ``w*phase_source``; native pressure-split defects are retained. Work and paired boundary balances also cover full-support regions and zero EPI weight. Their actual divided rates require positive capacity only inside the selected region. These more general read-outs do not broaden the older transport metric's admission. Each actual report populates ``work``, ``boundary`` and ``phase_response``; their optional defaults only preserve older manually constructed records. The unweighted phase response splits the actual phase-rate sum into mean-mobility cut, mobility/form covariance and exact rounding residual. It admits zero capacity and retains an exact squared covariance-rate bound without inventing a tolerance. Optional ordered cycles delegate geometric availability to the existing winding owner. No observation advances a graph or certifies a trajectory. """ field = evaluate_relational_exchange(graph, model=model) if not isinstance(reference_phase, Mapping): raise TypeError("reference_phase must map the exact full node support") if set(reference_phase) != set(field.nodes): raise ValueError("reference_phase must match the exact full node support") reference = tuple( finite_represented_real(reference_phase[node], "reference phase")[0] for node in field.nodes ) selections = _regions(field.nodes, regions) ordered_cycles = tuple( _ordered(cycle, "cycle") for cycle in _ordered(cycles, "cycles") ) detached = _detached_graph(field) unavailable = ( "zero_capacity_in_full_support" if any(capacity == 0.0 for capacity in field.capacity) else ("zero_epi_weight" if model.epi_weight == 0.0 else None) ) source = observe_support_transport(detached) degrees = tuple(len(row) for row in source.support_neighbors) forcing = tuple( Fraction(model.phase_weight) * Fraction(value) for value in field.phase_source ) observed = [] for indices in selections: nodes = tuple(field.nodes[i] for i in indices) form = tuple(Fraction(field.epi[i]) for i in indices) error = tuple( Fraction(field.phase[i]) - Fraction(reference[i]) for i in indices ) form_mean, phase_mean = sum(form) / len(form), sum(error) / len(error) centered_form = tuple(value - form_mean for value in form) centered_phase = tuple(value - phase_mean for value in error) reason = unavailable or ( "full_support_region" if len(indices) == len(field.nodes) else None ) transport = ( observe_regional_support_balance( source, nodes, epi_weight=Fraction(model.epi_weight), forcing=forcing ) if reason is None else None ) cut = ( transport.cut if transport is not None else observe_regional_support_cut(source, nodes) ) observed.append( RegionalPatternObservation( nodes=nodes, form_mean=form_mean, phase_error_mean=phase_mean, centered_form=centered_form, centered_phase_error=centered_phase, form_norm_squared=sum( (value**2 for value in centered_form), Fraction(0) ), phase_norm_squared=sum( (value**2 for value in centered_phase), Fraction(0) ), transport=transport, transport_unavailable_reason=reason, work=_regional_work(field, indices), boundary=_regional_boundary(field, cut, degrees), phase_response=_regional_phase_response(field, cut), ) ) return RelationalPatternObservation( field=field, reference_phase=reference, regions=tuple(observed), winding=tuple( certify_phase_winding(detached, cycle) for cycle in ordered_cycles ), )