Bernt
bae705aa97
- Add NFC ePassport roadmap (ICAO 9303, eIDAS) - Add TensorFlow.js edge face detection (BlazeFace) - Add structured audit logger (GDPR-compliant) - Risk scoring support Part of KYC Apple Native UX v1.1.0
457 lines
16 KiB
Python
457 lines
16 KiB
Python
"""
|
|
Decision Graph
|
|
Maps observations → decisions → consequences
|
|
"""
|
|
|
|
from typing import Dict, List, Optional, Set
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
|
|
@dataclass
|
|
class ImpactScore:
|
|
"""Impact score for an observation"""
|
|
economic: float # 0-100
|
|
social: float
|
|
safety: float
|
|
climate: float
|
|
political: float
|
|
maintenance: float
|
|
|
|
def total_impact(self) -> float:
|
|
return (self.economic + self.social + self.safety +
|
|
self.climate + self.political + self.maintenance) / 6
|
|
|
|
@property
|
|
def total(self) -> float:
|
|
return self.total_impact()
|
|
|
|
def to_dict(self) -> Dict:
|
|
return {
|
|
"economic": round(self.economic, 1),
|
|
"social": round(self.social, 1),
|
|
"safety": round(self.safety, 1),
|
|
"climate": round(self.climate, 1),
|
|
"political": round(self.political, 1),
|
|
"maintenance": round(self.maintenance, 1),
|
|
"total": round(self.total_impact(), 1)
|
|
}
|
|
|
|
|
|
class ImpactEngine:
|
|
"""Calculates impact scores for observations"""
|
|
|
|
def calculate_impact(self, observation: Dict) -> ImpactScore:
|
|
"""Calculate impact score for an observation"""
|
|
|
|
# Extract observation details
|
|
defect_code = observation.get("defect_code", "")
|
|
condition = observation.get("condition", 3)
|
|
location_type = observation.get("location_type", "urban")
|
|
|
|
# Base impact from defect type
|
|
base_impacts = {
|
|
"2300": {"economic": 60, "social": 40, "safety": 70, "climate": 20, "political": 30, "maintenance": 80}, # Surface damage
|
|
"2100": {"economic": 30, "social": 50, "safety": 40, "climate": 10, "political": 20, "maintenance": 60}, # Dirt
|
|
"2400": {"economic": 40, "social": 60, "safety": 50, "climate": 10, "political": 40, "maintenance": 50}, # Graffiti
|
|
"1100": {"economic": 50, "social": 30, "safety": 80, "climate": 30, "political": 20, "maintenance": 70}, # Rust
|
|
"5100": {"economic": 70, "social": 60, "safety": 90, "climate": 20, "political": 50, "maintenance": 60}, # Blockage
|
|
"6200": {"economic": 80, "social": 70, "safety": 85, "climate": 60, "political": 60, "maintenance": 90}, # Water damage
|
|
}
|
|
|
|
base = base_impacts.get(defect_code, {
|
|
"economic": 40, "social": 40, "safety": 40,
|
|
"climate": 40, "political": 40, "maintenance": 40
|
|
})
|
|
|
|
# Scale by condition (worse condition = higher impact)
|
|
condition_multiplier = condition / 3 # 1->0.33, 5->1.67
|
|
|
|
return ImpactScore(
|
|
economic=min(100, base["economic"] * condition_multiplier),
|
|
social=min(100, base["social"] * condition_multiplier),
|
|
safety=min(100, base["safety"] * condition_multiplier),
|
|
climate=min(100, base["climate"] * condition_multiplier),
|
|
political=min(100, base["political"] * condition_multiplier),
|
|
maintenance=min(100, base["maintenance"] * condition_multiplier)
|
|
)
|
|
|
|
|
|
class StakeholderEngine:
|
|
"""Identifies stakeholders affected by observations"""
|
|
|
|
def __init__(self):
|
|
self.stakeholder_rules = self._load_rules()
|
|
|
|
def _load_rules(self) -> Dict:
|
|
"""Load stakeholder identification rules"""
|
|
return {
|
|
"water_damage": ["water_utility", "municipality", "property_owner", "insurance"],
|
|
"road_damage": ["municipality", "traffic_authority", "insurance", "logistics"],
|
|
"lighting_failure": ["municipality", "property_owner", "safety_authority"],
|
|
"graffiti": ["municipality", "property_owner", "police"],
|
|
"blockage": ["municipality", "emergency_services", "property_owner"],
|
|
"vegetation": ["municipality", "property_owner", "environmental_agency"]
|
|
}
|
|
|
|
def identify_stakeholders(self, observation: Dict) -> List[Dict]:
|
|
"""Identify stakeholders for an observation"""
|
|
|
|
defect_code = observation.get("defect_code", "")
|
|
impact = observation.get("impact", {})
|
|
|
|
# Map defect code to category
|
|
category_map = {
|
|
"6200": "water_damage",
|
|
"2300": "road_damage",
|
|
"2100": "road_damage",
|
|
"2400": "graffiti",
|
|
"5100": "blockage",
|
|
"1100": "road_damage"
|
|
}
|
|
|
|
category = category_map.get(defect_code, "general")
|
|
stakeholders = self.stakeholder_rules.get(category, ["municipality"])
|
|
|
|
# Prioritize by impact
|
|
prioritized = []
|
|
for stakeholder in stakeholders:
|
|
priority = self._calculate_priority(stakeholder, impact)
|
|
prioritized.append({
|
|
"stakeholder": stakeholder,
|
|
"priority": priority,
|
|
"reason": self._get_reason(stakeholder, observation)
|
|
})
|
|
|
|
# Sort by priority
|
|
prioritized.sort(key=lambda x: x["priority"], reverse=True)
|
|
|
|
return prioritized
|
|
|
|
def _calculate_priority(self, stakeholder: str, impact: Dict) -> str:
|
|
"""Calculate priority for a stakeholder"""
|
|
# Simple heuristic based on impact type
|
|
if stakeholder in ["municipality", "emergency_services"]:
|
|
return "critical"
|
|
elif stakeholder in ["water_utility", "safety_authority"]:
|
|
return "high"
|
|
elif stakeholder in ["property_owner", "insurance"]:
|
|
return "medium"
|
|
else:
|
|
return "low"
|
|
|
|
def _get_reason(self, stakeholder: str, observation: Dict) -> str:
|
|
"""Get reason why stakeholder is affected"""
|
|
reasons = {
|
|
"municipality": "Responsible for public infrastructure",
|
|
"water_utility": "Responsible for water infrastructure",
|
|
"property_owner": "Property value and tenant safety affected",
|
|
"insurance": "Risk of claims and damage",
|
|
"traffic_authority": "Road safety and traffic flow",
|
|
"emergency_services": "Emergency access potentially blocked",
|
|
"safety_authority": "Public safety concern"
|
|
}
|
|
return reasons.get(stakeholder, "General interest")
|
|
|
|
|
|
class OwnershipGraph:
|
|
"""Maps ownership and responsibility for objects"""
|
|
|
|
def __init__(self):
|
|
self.ownership = {}
|
|
|
|
def register_object(
|
|
self,
|
|
goid: str,
|
|
owner: str,
|
|
maintainer: str,
|
|
insurer: Optional[str] = None,
|
|
operator: Optional[str] = None,
|
|
municipality: Optional[str] = None
|
|
):
|
|
"""Register ownership for an object"""
|
|
self.ownership[goid] = {
|
|
"goid": goid,
|
|
"owner": owner,
|
|
"maintainer": maintainer,
|
|
"insurer": insurer,
|
|
"operator": operator,
|
|
"municipality": municipality
|
|
}
|
|
|
|
def get_responsible_party(self, goid: str, issue_type: str) -> Optional[str]:
|
|
"""Get responsible party for an issue"""
|
|
obj = self.ownership.get(goid)
|
|
if not obj:
|
|
return None
|
|
|
|
# Route to appropriate party
|
|
if issue_type in ["maintenance", "repair"]:
|
|
return obj.get("maintainer", obj.get("owner"))
|
|
elif issue_type in ["insurance", "claim"]:
|
|
return obj.get("insurer", obj.get("owner"))
|
|
elif issue_type in ["operation", "service"]:
|
|
return obj.get("operator", obj.get("owner"))
|
|
else:
|
|
return obj.get("owner")
|
|
|
|
def get_object_chain(self, goid: str) -> Dict:
|
|
"""Get full ownership chain"""
|
|
return self.ownership.get(goid, {})
|
|
|
|
|
|
class CostEngine:
|
|
"""Estimates costs of inaction"""
|
|
|
|
def estimate_cost_of_inaction(
|
|
self,
|
|
observation: Dict,
|
|
time_horizon_months: int = 6
|
|
) -> Dict:
|
|
"""
|
|
Estimate cost of not fixing an issue
|
|
|
|
Returns:
|
|
Cost breakdown
|
|
"""
|
|
defect_code = observation.get("defect_code", "")
|
|
severity = observation.get("condition", 3)
|
|
|
|
# Base costs by defect type
|
|
base_costs = {
|
|
"2300": {"immediate": 5000, "escalated": 50000}, # Surface damage
|
|
"6200": {"immediate": 10000, "escalated": 100000}, # Water damage
|
|
"5100": {"immediate": 2000, "escalated": 20000}, # Blockage
|
|
"1100": {"immediate": 3000, "escalated": 30000}, # Rust
|
|
"2400": {"immediate": 1000, "escalated": 10000}, # Graffiti
|
|
}
|
|
|
|
costs = base_costs.get(defect_code, {"immediate": 5000, "escalated": 50000})
|
|
|
|
# Scale by severity
|
|
severity_multiplier = severity / 3
|
|
|
|
immediate_cost = costs["immediate"] * severity_multiplier
|
|
escalated_cost = costs["escalated"] * severity_multiplier * (time_horizon_months / 6)
|
|
|
|
# Additional costs
|
|
indirect_costs = self._calculate_indirect_costs(observation, time_horizon_months)
|
|
|
|
return {
|
|
"defect_code": defect_code,
|
|
"time_horizon_months": time_horizon_months,
|
|
"immediate_repair_cost": round(immediate_cost, 0),
|
|
"escalated_repair_cost": round(escalated_cost, 0),
|
|
"indirect_costs": round(indirect_costs, 0),
|
|
"total_cost_of_inaction": round(escalated_cost + indirect_costs, 0),
|
|
"savings_from_early_action": round(escalated_cost + indirect_costs - immediate_cost, 0)
|
|
}
|
|
|
|
def _calculate_indirect_costs(self, observation: Dict, months: int) -> float:
|
|
"""Calculate indirect costs (accidents, delays, etc.)"""
|
|
defect_code = observation.get("defect_code", "")
|
|
location = observation.get("location_type", "urban")
|
|
|
|
# Traffic impact
|
|
if defect_code in ["2300", "5100"] and location == "urban":
|
|
return 5000 * months # Traffic delays
|
|
|
|
# Safety impact
|
|
if defect_code in ["6200", "1100"]:
|
|
return 8000 * months # Accident risk
|
|
|
|
return 2000 * months # General degradation
|
|
|
|
|
|
class PriorityEngine:
|
|
"""Prioritizes observations automatically"""
|
|
|
|
def prioritize(
|
|
self,
|
|
observations: List[Dict],
|
|
budget_usd: Optional[float] = None,
|
|
max_items: int = 500
|
|
) -> List[Dict]:
|
|
"""
|
|
Prioritize observations
|
|
|
|
Returns:
|
|
Prioritized list
|
|
"""
|
|
scored = []
|
|
|
|
for obs in observations:
|
|
# Calculate priority score
|
|
score = self._calculate_priority_score(obs)
|
|
|
|
scored.append({
|
|
"observation": obs,
|
|
"priority_score": score,
|
|
"priority_level": self._score_to_level(score)
|
|
})
|
|
|
|
# Sort by score
|
|
scored.sort(key=lambda x: x["priority_score"], reverse=True)
|
|
|
|
# Filter by budget if provided
|
|
if budget_usd:
|
|
selected = []
|
|
total_cost = 0
|
|
|
|
for item in scored:
|
|
cost = item["observation"].get("repair_cost", 5000)
|
|
if total_cost + cost <= budget_usd:
|
|
selected.append(item)
|
|
total_cost += cost
|
|
|
|
if len(selected) >= max_items:
|
|
break
|
|
|
|
return selected
|
|
|
|
return scored[:max_items]
|
|
|
|
def _calculate_priority_score(self, observation: Dict) -> float:
|
|
"""Calculate priority score"""
|
|
impact = observation.get("impact", {})
|
|
cost = observation.get("repair_cost", 5000)
|
|
|
|
# Impact score
|
|
impact_score = impact.get("total", 50)
|
|
|
|
# Urgency (condition)
|
|
condition = observation.get("condition", 3)
|
|
urgency = condition * 20 # 1->20, 5->100
|
|
|
|
# Cost efficiency
|
|
efficiency = 100 / max(cost / 1000, 1)
|
|
|
|
# Combined score
|
|
score = (impact_score * 0.4 + urgency * 0.4 + efficiency * 0.2)
|
|
|
|
return score
|
|
|
|
def _score_to_level(self, score: float) -> str:
|
|
"""Convert score to priority level"""
|
|
if score >= 80:
|
|
return "critical"
|
|
elif score >= 60:
|
|
return "high"
|
|
elif score >= 40:
|
|
return "medium"
|
|
else:
|
|
return "low"
|
|
|
|
|
|
class ROICalculator:
|
|
"""Calculates ROI for interventions"""
|
|
|
|
def calculate_roi(self, intervention: Dict) -> Dict:
|
|
"""Calculate ROI for an intervention"""
|
|
cost = intervention.get("cost", 0)
|
|
expected_benefit = intervention.get("expected_benefit", 0)
|
|
|
|
if cost == 0:
|
|
return {"roi": float('inf'), "payback_months": 0}
|
|
|
|
roi = (expected_benefit - cost) / cost
|
|
payback = cost / max(expected_benefit / 12, 1) # Monthly benefit
|
|
|
|
return {
|
|
"cost": cost,
|
|
"expected_benefit": expected_benefit,
|
|
"roi": round(roi, 2),
|
|
"roi_percent": round(roi * 100, 1),
|
|
"payback_months": round(payback, 1)
|
|
}
|
|
|
|
def rank_interventions(self, interventions: List[Dict]) -> List[Dict]:
|
|
"""Rank interventions by ROI"""
|
|
ranked = []
|
|
|
|
for intervention in interventions:
|
|
roi_data = self.calculate_roi(intervention)
|
|
ranked.append({
|
|
**intervention,
|
|
**roi_data
|
|
})
|
|
|
|
# Sort by ROI
|
|
ranked.sort(key=lambda x: x["roi"], reverse=True)
|
|
|
|
return ranked
|
|
|
|
|
|
# Example usage
|
|
def example_decision_intelligence():
|
|
"""Example: Decision intelligence"""
|
|
|
|
# Impact Engine
|
|
impact_engine = ImpactEngine()
|
|
|
|
# Stakeholder Engine
|
|
stakeholder_engine = StakeholderEngine()
|
|
|
|
# Cost Engine
|
|
cost_engine = CostEngine()
|
|
|
|
# Priority Engine
|
|
priority_engine = PriorityEngine()
|
|
|
|
# ROI Calculator
|
|
roi_calculator = ROICalculator()
|
|
|
|
# Example observation
|
|
observation = {
|
|
"goid": "TRN-ROD-SUR-001",
|
|
"defect_code": "6200",
|
|
"condition": 4,
|
|
"location_type": "urban",
|
|
"impact": {"economic": 80, "social": 70, "safety": 85, "climate": 60, "political": 60, "maintenance": 90}
|
|
}
|
|
|
|
print("=== Impact Score ===")
|
|
impact = impact_engine.calculate_impact(observation)
|
|
print(f"Total impact: {impact.total_impact()}")
|
|
print(f"Safety: {impact.safety}")
|
|
print(f"Economic: {impact.economic}")
|
|
|
|
print("\n=== Stakeholders ===")
|
|
stakeholders = stakeholder_engine.identify_stakeholders(observation)
|
|
for s in stakeholders:
|
|
print(f" {s['stakeholder']}: {s['priority']} - {s['reason']}")
|
|
|
|
print("\n=== Cost of Inaction ===")
|
|
cost = cost_engine.estimate_cost_of_inaction(observation, 6)
|
|
print(f"Immediate repair: ${cost['immediate_repair_cost']:,}")
|
|
print(f"Cost of waiting 6 months: ${cost['total_cost_of_inaction']:,}")
|
|
print(f"Savings from early action: ${cost['savings_from_early_action']:,}")
|
|
|
|
print("\n=== Priority ===")
|
|
prioritized = priority_engine.prioritize([observation])
|
|
for item in prioritized:
|
|
print(f" Score: {item['priority_score']:.1f} ({item['priority_level']})")
|
|
|
|
print("\n=== ROI Ranking ===")
|
|
interventions = [
|
|
{"name": "Replace lighting", "cost": 800000, "expected_benefit": 2500000},
|
|
{"name": "Repair sidewalk", "cost": 400000, "expected_benefit": 1200000},
|
|
{"name": "Plant trees", "cost": 200000, "expected_benefit": 800000},
|
|
{"name": "Remove graffiti", "cost": 50000, "expected_benefit": 300000}
|
|
]
|
|
|
|
ranked = roi_calculator.rank_interventions(interventions)
|
|
for item in ranked:
|
|
print(f" {item['name']}: ROI {item['roi_percent']}% (payback {item['payback_months']} months)")
|
|
|
|
return {
|
|
"impact": impact.to_dict(),
|
|
"stakeholders": stakeholders,
|
|
"cost": cost,
|
|
"ranked_interventions": ranked
|
|
}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
example_decision_intelligence()
|