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
155 changes: 153 additions & 2 deletions harmonizer/coordinate_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,9 @@
"""

import math
from typing import Tuple
from typing import Iterable, Optional, Tuple

from harmonizer.divine_invitation_engine_V2 import Coordinates
from harmonizer.divine_invitation_engine_V2 import Coordinates, Dimension


class CoordinateUtils:
Expand Down Expand Up @@ -194,6 +194,157 @@ def from_tuple(coord_tuple: Tuple[float, float, float, float]) -> Coordinates:
wisdom=coord_tuple[3],
)

@staticmethod
def blend(
base: Coordinates,
overlay: Coordinates,
ratio: float = 0.5,
) -> Coordinates:
"""
Blend two coordinates together with the supplied ratio.

Args:
base: Primary coordinate that receives the overlay.
overlay: Secondary coordinate to blend into the base.
ratio: Value between 0.0 and 1.0 indicating overlay strength.

Returns:
Blended Coordinates object.
"""
ratio = max(0.0, min(1.0, ratio))
inverse = 1.0 - ratio
return Coordinates(
love=(base.love * inverse) + (overlay.love * ratio),
justice=(base.justice * inverse) + (overlay.justice * ratio),
power=(base.power * inverse) + (overlay.power * ratio),
wisdom=(base.wisdom * inverse) + (overlay.wisdom * ratio),
)

@staticmethod
def weighted_average(
coords: Iterable[Coordinates],
weights: Optional[Iterable[float]] = None,
) -> Coordinates:
"""
Compute a weighted average of coordinates.

Args:
coords: Iterable of Coordinates objects.
weights: Optional iterable of weights (defaults to equal weighting).

Returns:
Coordinates representing the weighted average.
"""
coords = list(coords)
if not coords:
return Coordinates(0.0, 0.0, 0.0, 0.0)

if weights is None:
weights = [1.0] * len(coords)
else:
weights = list(weights)
if len(weights) != len(coords):
raise ValueError("weights length must match coords length")

total_weight = sum(weights)
if total_weight == 0.0:
return Coordinates(0.0, 0.0, 0.0, 0.0)

love_sum = justice_sum = power_sum = wisdom_sum = 0.0
for coord, weight in zip(coords, weights):
love_sum += coord.love * weight
justice_sum += coord.justice * weight
power_sum += coord.power * weight
wisdom_sum += coord.wisdom * weight

return Coordinates(
love=love_sum / total_weight,
justice=justice_sum / total_weight,
power=power_sum / total_weight,
wisdom=wisdom_sum / total_weight,
)

@staticmethod
def ensure_power_floor(
coord: Tuple[float, float, float, float],
minimum_power: float = 0.2,
) -> Tuple[float, float, float, float]:
"""
Ensure that a coordinate tuple has at least ``minimum_power`` in the power dimension.

If power is already at or above the floor, the coordinate is returned unchanged.
Otherwise the deficit is proportionally reallocated from the remaining dimensions.
"""
love, justice, power, wisdom = coord
if power >= minimum_power:
return coord

deficit = minimum_power - power
remaining = love + justice + wisdom
if remaining == 0.0:
# Nothing to redistribute, so just set the floor directly.
return (0.0, 0.0, minimum_power, 0.0)

love_ratio = love / remaining
justice_ratio = justice / remaining
wisdom_ratio = wisdom / remaining

love -= deficit * love_ratio
justice -= deficit * justice_ratio
wisdom -= deficit * wisdom_ratio
power = minimum_power

# Normalize to keep the tuple on the unit simplex.
total = love + justice + power + wisdom
if total == 0.0:
return (0.0, 0.0, power, 0.0)

return (love / total, justice / total, power / total, wisdom / total)

@staticmethod
def prioritize_dimension(
coord: Tuple[float, float, float, float],
dimension: Dimension,
boost: float = 0.05,
) -> Tuple[float, float, float, float]:
"""
Aggressively boosts a selected dimension by reallocating weight from others.

Args:
coord: Original coordinate tuple.
dimension: Dimension enum value to prioritize (e.g., Dimension.POWER).
boost: Amount to move into the prioritized dimension.

Returns:
Tuple with the prioritized dimension amplified.
"""
index_map = {
Dimension.LOVE: 0,
Dimension.JUSTICE: 1,
Dimension.POWER: 2,
Dimension.WISDOM: 3,
}
if dimension not in index_map:
raise ValueError("dimension must be a primary LJPW dimension")

values = list(coord)
idx = index_map[dimension]
boost = max(0.0, boost)
available_indices = [i for i in range(4) if i != idx]
available_total = sum(values[i] for i in available_indices)
if available_total == 0.0:
values[idx] = min(1.0, values[idx] + boost)
else:
for i in available_indices:
share = boost * (values[i] / available_total)
values[i] = max(0.0, values[i] - share)
values[idx] = min(1.0, values[idx] + boost)

total = sum(values)
if total == 0.0:
return (0.0, 0.0, 0.0, 0.0)
return tuple(v / total for v in values)

@staticmethod
def get_dominant_dimension(coord: Tuple[float, float, float, float]) -> str:
"""
Expand Down
58 changes: 58 additions & 0 deletions harmonizer/divine_invitation_engine_V2.py
Original file line number Diff line number Diff line change
Expand Up @@ -704,6 +704,9 @@ def analyze_ice(
context_result.coordinates, execution_result.coordinates
)

execution_intensity = self._calculate_execution_intensity(execution_result)
alignment_pressure = self._calculate_alignment_pressure(intent_result, execution_result)

# Calculate ICE metrics
avg_disharmony = (intent_context_dist + intent_exec_dist + context_exec_dist) / 3.0
ice_coherence = max(0.0, 1.0 - (avg_disharmony / 2.0))
Expand Down Expand Up @@ -767,6 +770,8 @@ def analyze_ice(
"ice_coherence": ice_coherence,
"ice_balance": ice_balance,
"benevolence_score": benevolence_score,
"execution_intensity": execution_intensity,
"alignment_pressure": alignment_pressure,
"intent_execution_disharmony": intent_exec_dist,
# Baseline-enhanced metrics
"baseline_disharmony": baseline_disharmony,
Expand Down Expand Up @@ -841,6 +846,36 @@ def _determine_ice_harmony_level(self, coherence: float, balance: float) -> str:
else:
return "POOR_ICE_BALANCE"

def _calculate_execution_intensity(self, execution_result: SemanticResult) -> float:
"""
Measure how forceful the execution semantics are.

High concept counts with strong power coordinates produce higher intensity.
"""
coords = execution_result.coordinates
base_intensity = coords.power
if execution_result.concept_count == 0:
return 0.0
# Scale up with concept density but clamp to 1.0
density_bonus = min(1.0, execution_result.concept_count / 10.0)
return min(1.0, base_intensity * 0.6 + density_bonus * 0.4)

def _calculate_alignment_pressure(
self, intent_result: SemanticResult, execution_result: SemanticResult
) -> float:
"""
Determine how much force is required to align execution with intent.

Larger distances and weak execution intensity increase the pressure score.
"""
distance = self.vocab.get_distance(intent_result.coordinates, execution_result.coordinates)
intensity = self._calculate_execution_intensity(execution_result)
if intensity == 0.0:
return min(1.0, distance)
# When execution is already forceful, pressure drops significantly
pressure = distance / (1.0 + (intensity * 2.0))
return min(1.0, pressure)


class PhiOptimizer:
"""Optimized phi-enhanced mathematical optimization"""
Expand Down Expand Up @@ -987,6 +1022,29 @@ def perform_phi_optimization(self, concepts: List[str]) -> Dict:
"""Perform phi-enhanced optimization"""
return self.phi_optimizer.calculate_phi_optimization(concepts)

def evaluate_action_gap(
self,
intent_words: List[str],
execution_words: List[str],
context_words: Optional[List[str]] = None,
) -> Dict:
"""
Rapid assessment focused on action: quantifies how forcefully code executes intent.

Returns execution intensity, alignment pressure, and baseline disharmony metrics so that
callers can decide whether to refactor the implementation or rename the API surface.
"""
context_words = context_words or []
ice_result = self.perform_ice_analysis(intent_words, context_words, execution_words)
metrics = ice_result["ice_metrics"]
return {
"execution_intensity": metrics.get("execution_intensity", 0.0),
"alignment_pressure": metrics.get("alignment_pressure", 0.0),
"baseline_disharmony": metrics.get("baseline_disharmony", 0.0),
"intent_execution_disharmony": metrics.get("intent_execution_disharmony", 0.0),
"ice_harmony_level": ice_result.get("ice_harmony_level"),
}


def run_comprehensive_demo():
"""Optimized demonstration of DIVE-V2 capabilities"""
Expand Down
Loading