Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,8 @@ jobs:
agencitylab/reference/_download.py \
agencitylab/exceptions.py \
agencitylab/scientific_status.py
- name: Type-check complete package
run: mypy agencitylab

tests:
name: Tests / Python ${{ matrix.python-version }}
Expand Down
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,10 @@ identifiers are not part of the stable public compatibility contract.
- Added explicit result-summary names for magnitude means and circular phase
statistics while retaining compatibility keys.
- Reconciled the documented and actual ``agencitylab.fields`` dynamics exports.
- Fixed the optional Vopson information-mass helper to use numeric physical-constant
values while keeping the relation explicitly speculative.
- Resolved the historical package-wide mypy debt and added a full-package typing
gate so internal typing regressions are no longer hidden behind the public-surface check.

### Scientific integrity

Expand Down
16 changes: 7 additions & 9 deletions agencitylab/analysis/information/agencity_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,19 +81,17 @@ def agencity_phase_information(theta, *, verbose=False):
return H


def full_information_summary(b, J=None, theta=None):
"""
Complete information description.
"""
def full_information_summary(b, J=None, theta=None, *, verbose=False):
"""Return the descriptive information summary without altering canonical data."""
out = {
"entropy_b": agencity_information_index(b),
"density_b": agencity_information_density(b),
"entropy_b": agencity_information_index(b, verbose=verbose),
"density_b": agencity_information_density(b, verbose=verbose),
}

if J is not None:
out["structure_J"] = agencity_structural_information(J)
out["structure_J"] = agencity_structural_information(J, verbose=verbose)

if theta is not None:
out["phase_entropy"] = agencity_phase_information(theta)
out["phase_entropy"] = agencity_phase_information(theta, verbose=verbose)

return out
return out
37 changes: 22 additions & 15 deletions agencitylab/analysis/information/vopson.py
Original file line number Diff line number Diff line change
@@ -1,27 +1,34 @@
"""
Vopson information mass equivalence (advanced).
"""Vopson information-mass equivalence hypothesis.

This module implements the proposed relation as a speculative information-physics
extension. It is not part of the canonical Theory of Agencity and is not treated
as an experimentally established mass law by AgencityLab.
"""

from __future__ import annotations
import numpy as np

C = 299792458 # speed of light
from agencitylab.constants.physics import SPEED_OF_LIGHT
from agencitylab.scientific_status import ScientificStatus

from .landauer import landauer_from_entropy, landauer_lower_bound

SCIENTIFIC_STATUS = ScientificStatus.SPECULATIVE
C = SPEED_OF_LIGHT.value


def information_mass(bits: float, temperature: float) -> float:
"""
Information → mass equivalence (Vopson hypothesis)
"""Return the Vopson-hypothesis mass equivalent for ``bits`` at ``temperature``.

m = (k_B * T * ln2 * bits) / c^2
The implemented relation is ``m = k_B*T*ln(2)*bits/c**2``. The Landauer
helper supplies the numerical Boltzmann constant and validates finite,
non-negative inputs; division by ``c**2`` is the speculative mass-equivalence
step.
"""
from agencitylab.constants.physics import BOLTZMANN_CONSTANT

return (BOLTZMANN_CONSTANT * temperature * np.log(2) * bits) / (C**2)
return float(landauer_lower_bound(bits, temperature) / (C**2))


def vopson_mass_equivalent(entropy_nats: float, temperature: float):
"""
Convert entropy (nats) → mass
"""
bits = entropy_nats / np.log(2)
return information_mass(bits, temperature)
def vopson_mass_equivalent(entropy_nats: float, temperature: float) -> float:
"""Return the speculative mass equivalent for entropy expressed in nats."""

return float(landauer_from_entropy(entropy_nats, temperature) / (C**2))
14 changes: 9 additions & 5 deletions agencitylab/analysis/reports.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from __future__ import annotations

from dataclasses import asdict, is_dataclass
from typing import Any, Dict, Mapping, Optional
from typing import Any, Dict, Mapping, Optional, cast

import numpy as np

Expand Down Expand Up @@ -46,7 +46,7 @@
def _portable(value):
"""Convert analysis output to JSON-friendly values without losing metadata."""
if is_dataclass(value):
return _portable(asdict(value))
return _portable(asdict(cast(Any, value)))
if isinstance(value, np.ndarray):
return value.tolist()
if isinstance(value, (np.floating, np.integer, np.bool_)):
Expand Down Expand Up @@ -178,6 +178,7 @@ def build_report_dict(
crossing_indices = crossings + start
crossing_times = xi_valid[crossings] if crossings.size else np.asarray([], dtype=float)

theta_jumps: dict[str, Any]
if theta_jump_threshold is None:
theta_jumps = {
"status": "not configured",
Expand All @@ -198,6 +199,7 @@ def build_report_dict(
"times": xi_valid[jump_indices] if jump_indices.size else np.asarray([], dtype=float),
}

plateaus: dict[str, Any]
if plateau_slope_threshold is None and plateau_min_duration is None:
plateaus = {
"status": "not configured",
Expand All @@ -221,9 +223,11 @@ def build_report_dict(
),
}

orientation = orientation_statistics(
np.asarray(result.M)[start:stop],
np.asarray(result.O)[start:stop],
orientation: dict[str, Any] = dict(
orientation_statistics(
np.asarray(result.M)[start:stop],
np.asarray(result.O)[start:stop],
)
)
orientation["sigma_theta_mean"] = (
float(np.mean(sigma[finite_sigma]))
Expand Down
7 changes: 4 additions & 3 deletions agencitylab/api/analyze.py
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ def analyze_coherence(
b = np.asarray(_get_attr(result, "b"), dtype=complex)
valid = S > 0.0
sigma = sigma_theta(theta, xi, tau, valid_mask=valid)
orientation = orientation_statistics(M, O)
orientation: dict[str, Any] = dict(orientation_statistics(M, O))
orientation["structural_phase_coherence"] = phase_coherence(
theta,
valid_mask=valid,
Expand Down Expand Up @@ -184,8 +184,8 @@ def analyze_stability(result, *, verbose: bool = False) -> Dict[str, Any]:
def analyze_information(result, *, verbose: bool = False) -> Dict[str, Any]:
b = _get_attr(result, "b")
info = full_information_summary(b, verbose=verbose)
info["agencity_information_index"] = agencity_information_index(b, verbose=verbose)
info["agencity_information_density"] = agencity_information_density(b, verbose=verbose)
info["agencity_information_index"] = info["entropy_b"]
info["agencity_information_density"] = info["density_b"]
return info


Expand Down Expand Up @@ -236,6 +236,7 @@ def analyze_transitions(
crossings = critical_surface_crossings(D[start:], S[start:])
crossing_indices = crossings + start

jumps: dict[str, Any]
if theta_jump_threshold is None:
jumps = {
"status": "not configured",
Expand Down
4 changes: 2 additions & 2 deletions agencitylab/api/export.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

from dataclasses import asdict, is_dataclass
from pathlib import Path
from typing import Any
from typing import Any, cast

import numpy as np

Expand All @@ -30,7 +30,7 @@ def _flatten_for_table(data: Any, parent_key: str = "", sep: str = ".") -> dict[
if hasattr(data, "to_dict") and callable(data.to_dict):
data = data.to_dict()
elif is_dataclass(data):
data = asdict(data)
data = asdict(cast(Any, data))

items: dict[str, Any] = {}
if isinstance(data, dict):
Expand Down
5 changes: 3 additions & 2 deletions agencitylab/backends/numba_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,12 @@

from __future__ import annotations

from typing import Callable
from typing import Callable, cast

import numpy as np

from .numpy_backend import (
WindowKind,
apply_window_numpy,
causal_moving_correlation_numpy,
normalize_numpy,
Expand Down Expand Up @@ -84,7 +85,7 @@ def central_difference_numba(values, step: float):

def apply_window_numba(values, kind: str = "hann"):
"""Use NumPy for already-vectorised tapering."""
return apply_window_numpy(values, kind=kind, axis=-1)
return apply_window_numpy(values, kind=cast(WindowKind, kind), axis=-1)


def causal_moving_correlation_numba(values, window: int = 1, epsilon: float = 1e-12):
Expand Down
10 changes: 5 additions & 5 deletions agencitylab/backends/numpy_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,16 +69,16 @@ def apply_window_numpy(values, kind: WindowKind = "hann", axis: int = -1):
if values.shape[axis] < 1:
raise ValueError("values must contain at least one sample.")

kind = str(kind).lower().strip()
key = str(kind).lower().strip()
n = values.shape[axis]

if kind == "hann":
if key == "hann":
window = np.hanning(n)
elif kind == "hamming":
elif key == "hamming":
window = np.hamming(n)
elif kind == "blackman":
elif key == "blackman":
window = np.blackman(n)
elif kind == "rectangular":
elif key == "rectangular":
window = np.ones(n, dtype=float)
else:
raise ValueError("Unknown window kind.")
Expand Down
4 changes: 2 additions & 2 deletions agencitylab/core/crm.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ def _uniform_step(axis):
axis = validate_axis(axis)
diffs = np.diff(axis)
step = float(diffs[0])
tolerance = np.finfo(float).eps * max(1.0, abs(step)) * 64.0
tolerance = float(np.finfo(float).eps * max(1.0, abs(step)) * 64.0)
if not np.allclose(diffs, step, rtol=1e-10, atol=tolerance):
raise ValueError("discrete CRM requires uniformly sampled coordinates")
return step
Expand All @@ -38,7 +38,7 @@ def _window_to_samples(window, axis):
raise ValueError("CRM window is smaller than one sampling interval")

represented = n * step
tolerance = max(np.finfo(float).eps * max(1.0, abs(window)) * 128.0, abs(step) * 1e-9)
tolerance = float(max(np.finfo(float).eps * max(1.0, abs(window)) * 128.0, abs(step) * 1e-9))
if not np.isclose(represented, window, rtol=1e-9, atol=tolerance):
raise ValueError("CRM window must be an integer multiple of the sampling interval")
return n
Expand Down
6 changes: 3 additions & 3 deletions agencitylab/core/multiscale.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,20 +32,20 @@ def _sample_step(axis: np.ndarray) -> float:
axis = validate_axis(axis)
diffs = np.diff(axis)
step = float(diffs[0])
tolerance = np.finfo(float).eps * max(1.0, abs(step)) * 64.0
tolerance = float(np.finfo(float).eps * max(1.0, abs(step)) * 64.0)
if not np.allclose(diffs, step, rtol=1e-10, atol=tolerance):
raise ValueError("discrete multiscale extensions require uniformly sampled coordinates")
return step


def _window_samples(window: float, axis: np.ndarray) -> int:
window = validate_positive_scalar(window, name="w")
window = float(validate_positive_scalar(window, name="w"))
step = _sample_step(axis)
samples = int(round(window / step))
if samples < 1:
raise ValueError("w is smaller than one sampling interval")
represented = samples * step
tolerance = max(np.finfo(float).eps * max(1.0, abs(window)) * 128.0, step * 1e-9)
tolerance = max(float(np.finfo(float).eps) * max(1.0, abs(window)) * 128.0, step * 1e-9)
if not np.isclose(represented, window, rtol=1e-9, atol=tolerance):
raise ValueError("w must be an integer multiple of the sampling interval")
return samples
Expand Down
12 changes: 6 additions & 6 deletions agencitylab/data/feature_extract/embeddings.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,30 +30,30 @@ def build_embedding_signal(
if arr.ndim != 2:
raise ValueError("vectors must be 2D (n_samples, n_features)")

mode = mode.lower().strip()
normalized_mode = str(mode).lower().strip()

# =========================
# 🔥 1. FULL INFORMATION (RECOMMENDED)
# =========================
if mode == "raw":
if normalized_mode == "raw":
return arr # shape (n, d)

# =========================
# ⚠️ LOSSY MODES
# =========================
if mode == "norm":
if normalized_mode == "norm":
return np.linalg.norm(arr, axis=1)

if mode == "mean":
if normalized_mode == "mean":
return np.mean(arr, axis=1)

if mode == "sum":
if normalized_mode == "sum":
return np.sum(arr, axis=1)

# =========================
# 🔬 PCA REDUCTION (SMART)
# =========================
if mode == "pca":
if normalized_mode == "pca":
# centrage
X = arr - np.mean(arr, axis=0)

Expand Down
7 changes: 6 additions & 1 deletion agencitylab/data/pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,13 @@ def window(self, kind: str = "hann") -> "DataPipeline":
def build(self) -> SignalData:
"""Return a canonical SignalData object."""
self._require_signal()
assert self.xi is not None and self.u is not None
self.steps.append("build")
return SignalData(xi=self.xi.copy(), u=self.u.copy(), metadata=dict(self.metadata))
return SignalData(
xi=self.xi.copy(),
u=self.u.copy(),
metadata=dict(self.metadata),
)

def _require_signal(self) -> None:
"""Ensure that a signal has already been loaded."""
Expand Down
9 changes: 6 additions & 3 deletions agencitylab/dynamics/delays.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ def interpolate_history(history_xi, history_y, xi_query):


def solve_delay_euler(
rhs: Callable[[float, np.ndarray, Callable[[float], np.ndarray]], np.ndarray],
rhs: Callable[[float, np.ndarray, np.ndarray], np.ndarray],
history_function: Callable[[float], np.ndarray],
xi_grid,
delay: float,
Expand Down Expand Up @@ -71,14 +71,17 @@ def solve_delay_euler(
def state_at(query_xi: float) -> np.ndarray:
if query_xi <= xi_grid[0]:
return np.asarray(history_function(query_xi), dtype=float)
idx = np.searchsorted(xi_grid[: len(trajectory)], query_xi, side="right") - 1
idx = int(np.searchsorted(xi_grid, query_xi, side="right")) - 1
idx = max(0, min(idx, len(trajectory) - 1))
return trajectory[idx]

for i in range(1, xi_grid.size):
h = float(xi_grid[i] - xi_grid[i - 1])
delayed_state = state_at(float(xi_grid[i - 1] - delay))
derivative = np.asarray(rhs(xi_grid[i - 1], trajectory[i - 1], delayed_state), dtype=float)
derivative = np.asarray(
rhs(float(xi_grid[i - 1]), trajectory[i - 1], delayed_state),
dtype=float,
)
trajectory[i] = trajectory[i - 1] + h * derivative

return trajectory
4 changes: 2 additions & 2 deletions agencitylab/fields/dynamics/simulation.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,8 +87,8 @@ def _set_boundary_faces(field: np.ndarray, value: complex | float) -> np.ndarray
dtype = np.result_type(field.dtype, np.asarray(value).dtype)
projected = np.array(field, dtype=dtype, copy=True)
for axis in range(projected.ndim):
lower = [slice(None)] * projected.ndim
upper = [slice(None)] * projected.ndim
lower: list[slice | int] = [slice(None)] * projected.ndim
upper: list[slice | int] = [slice(None)] * projected.ndim
lower[axis] = 0
upper[axis] = -1
projected[tuple(lower)] = value
Expand Down
2 changes: 1 addition & 1 deletion agencitylab/fields/numerics/grid.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ def _validate_axes(axes: Sequence[Iterable[float]]) -> tuple[np.ndarray, ...]:
if np.any(differences <= 0.0):
raise ValueError(f"axis {index} must be strictly increasing")
spacing = float(differences[0])
tolerance = max(abs(spacing) * 1e-12, np.finfo(float).eps * 32.0)
tolerance = max(abs(spacing) * 1e-12, float(np.finfo(float).eps) * 32.0)
if not np.allclose(differences, spacing, rtol=1e-10, atol=tolerance):
raise ValueError(f"axis {index} must be uniformly spaced")
copy = np.array(axis, dtype=float, copy=True)
Expand Down
Loading
Loading