Building zoning compliance rule engines in Python

import hashlib
import json
import logging
from enum import Enum
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime, timezone

# --- Error Categorization ---
class ComplianceErrorCategory(Enum):
    SCHEMA_VALIDATION = "SCHEMA_VALIDATION"
    PRECEDENCE_CONFLICT = "PRECEDENCE_CONFLICT"
    EXTERNAL_DEPENDENCY_FAILURE = "EXTERNAL_DEPENDENCY_FAILURE"
    SECURITY_VIOLATION = "SECURITY_VIOLATION"

class ComplianceEngineError(Exception):
    def __init__(self, category: ComplianceErrorCategory, message: str, payload: Optional[Dict] = None):
        self.category = category
        self.payload = payload or {}
        super().__init__(f"[{category.value}] {message}")

# --- Data Models & Schema Validation ---
@dataclass
class TowerAsset:
    site_id: str
    latitude: float
    longitude: float
    current_height_ft: float
    proposed_modification_ft: float

@dataclass
class ZoningOrdinance:
    jurisdiction: str
    max_height_ft: float
    setback_requirement_ft: float
    grandfathered: bool
    version_hash: str

@dataclass
class LeaseCovenant:
    lease_id: str
    landlord_approval_required: bool
    height_cap_ft: Optional[float]
    variance_on_file: bool

@dataclass
class EvaluationContext:
    asset: TowerAsset
    zoning: ZoningOrdinance
    covenant: LeaseCovenant
    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

# --- Security Boundary & Tokenization ---
def apply_security_boundary(context: EvaluationContext) -> EvaluationContext:
    """Redacts sensitive lease identifiers and tokenizes financial metadata."""
    safe_covenant = LeaseCovenant(
        lease_id=hashlib.sha256(context.covenant.lease_id.encode()).hexdigest()[:12],
        landlord_approval_required=context.covenant.landlord_approval_required,
        height_cap_ft=context.covenant.height_cap_ft,
        variance_on_file=context.covenant.variance_on_file
    )
    return EvaluationContext(asset=context.asset, zoning=context.zoning, covenant=safe_covenant, timestamp=context.timestamp)

# --- Audit Hashing ---
def generate_audit_hash(context: EvaluationContext, verdict: str) -> str:
    """Generates a deterministic SHA-256 hash for compliance traceability."""
    payload = json.dumps({
        "site_id": context.asset.site_id,
        "zoning_version": context.zoning.version_hash,
        "lease_token": context.covenant.lease_id,
        "verdict": verdict,
        "evaluated_at": context.timestamp
    }, sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

# --- Fallback Routing Protocols ---
class FallbackRouter:
    @staticmethod
    def handle_dependency_failure(error: Exception, context: EvaluationContext) -> Dict[str, Any]:
        logging.warning(f"External dependency failed: {error}. Routing to fallback evaluation.")
        # Fallback: Assume conservative municipal limits apply until manual review
        return {
            "status": "FALLBACK_REVIEW_REQUIRED",
            "reason": str(error),
            "site_id": context.asset.site_id,
            "action": "QUEUE_FOR_MANUAL_AUDIT"
        }

# --- Rule Engine Core ---
class ZoningComplianceEngine:
    def evaluate(self, context: EvaluationContext) -> Dict[str, Any]:
        try:
            # Security boundary enforcement
            secure_ctx = apply_security_boundary(context)
            
            # 1. Federal/Lease Precedence Check (Telecom Act § 332(c)(7) alignment)
            if secure_ctx.covenant.variance_on_file:
                verdict = "APPROVED_VARIANCE"
            # 2. Municipal Height & Setback Validation
            elif secure_ctx.asset.proposed_modification_ft > secure_ctx.zoning.max_height_ft:
                if secure_ctx.zoning.grandfathered:
                    verdict = "APPROVED_GRANDFATHERED"
                else:
                    verdict = "DENIED_HEIGHT_EXCEEDED"
            else:
                verdict = "APPROVED_COMPLIANT"
                
            audit_hash = generate_audit_hash(secure_ctx, verdict)
            
            return {
                "status": verdict,
                "audit_hash": audit_hash,
                "site_id": context.asset.site_id,
                "evaluated_at": context.timestamp,
                "security_tokenized": True
            }
            
        except ComplianceEngineError:
            # Domain errors carry their own category and propagate unchanged.
            raise
        except Exception as e:
            # Unexpected dependency or runtime failure: degrade to the
            # documented fallback path instead of halting the pipeline.
            return FallbackRouter.handle_dependency_failure(e, context)

# --- Execution Example ---
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    
    ctx = EvaluationContext(
        asset=TowerAsset(site_id="TX-DAL-8842", latitude=32.7767, longitude=-96.7970, current_height_ft=150, proposed_modification_ft=165),
        zoning=ZoningOrdinance(jurisdiction="Dallas_Municipal", max_height_ft=160, setback_requirement_ft=50, grandfathered=False, version_hash="v2024.1"),
        covenant=LeaseCovenant(lease_id="LSE-99102", landlord_approval_required=True, height_cap_ft=170, variance_on_file=False)
    )
    
    engine = ZoningComplianceEngine()
    try:
        result = engine.evaluate(ctx)
        print(json.dumps(result, indent=2))
    except ComplianceEngineError as err:
        print(f"Engine halted: {err.category.value} | {err}")
    except Exception as e:
        print(FallbackRouter.handle_dependency_failure(e, ctx))