safety

name	safety
description	Software safety principles
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"safety"}

What I do

• Design safe systems
• Implement safety-critical code
• Handle errors gracefully
• Prevent unsafe states
• Test for failure conditions

When to use me

When building systems where failures could cause harm.

Safety Engineering

Failure Mode Analysis

from enum import Enum

class FailureSeverity(Enum):
    CATASTROPHIC = 1  # Loss of life
    CRITICAL = 2      # Severe injury
    MARGINAL = 3      # Minor injury
    NEGLIGIBLE = 4    # No injury

class FailureModeAnalysis:
    """Analyze potential failure modes"""
    
    def __init__(self):
        self.failure_modes = []
    
    def add_failure_mode(self, component: str, failure_mode: str,
                        cause: str, effect: str, severity: FailureSeverity):
        """Add failure mode to analysis"""
        self.failure_modes.append({
            "component": component,
            "failure_mode": failure_mode,
            "cause": cause,
            "effect": effect,
            "severity": severity.value,
            "detection": "unknown",
            "mitigation": "none"
        })
    
    def prioritize_mitigation(self) -> list:
        """Order failure modes by priority"""
        return sorted(
            self.failure_modes,
            key=lambda f: f["severity"]
        )

Defensive Programming

class DefensiveInputValidation:
    """Validate all inputs"""
    
    def validate_and_convert(self, value: Any, expected_type: type,
                           default: Any = None) -> Any:
        """Validate and safely convert input"""
        try:
            if value is None:
                if default is not None:
                    return default
                raise ValueError("Required value is None")
            
            if not isinstance(value, expected_type):
                value = expected_type(value)
            
            return self._validate_range(value)
        
        except (ValueError, TypeError):
            if default is not None:
                return default
            raise
    
    def _validate_range(self, value) -> Any:
        """Validate value is within acceptable range"""
        return value  # Override with range checks

class SafeStateMachine:
    """Implement fail-safe state machine"""
    
    STATES = ["initializing", , , , , ]
    SAFE_STATE = 
    
     ():
        .state = 
        .previous_state = 
    
     () -> :
        
         new_state   .STATES:
             
        
          ._is_valid_transition(.state, new_state):
             
        
        .previous_state = .state
        .state = new_state
         
    
     () -> :
        
        valid_transitions = {
            : [, ],
            : [, , ],
            : [, , , ],
            : [, , ],
            : [],
            : []
        }
        
         to_state  valid_transitions.get(from_state, [])
    
     ():
        
        .previous_state = .state
        .state = .SAFE_STATE

Error Handling

import logging

class SafeErrorHandler:
    """Handle errors safely"""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def handle_exception(self, exception: Exception, 
                       context: Dict) -> None:
        """Handle exception with appropriate action"""
        error_type = type(exception).__name__
        
        # Log the error
        self.logger.error(
            f"Error in {context.get('operation')}: {error_type}",
            extra=context,
            exc_info=True
        )
        
        # Take corrective action based on severity
        if self._is_critical_error(exception):
            self._trigger_safe_mode()
        else:
            self._attempt_recovery(exception, context)
    
    def _is_critical_error(self, exception: Exception) -> bool:
        """Determine if error is critical"""
        critical_errors = [
            MemoryError,
            SystemError,
            KeyboardInterrupt  # Don't catch Ctrl+C
        ]
        return any(isinstance(exception, e) for e in critical_errors)
    
    def _trigger_safe_mode():
        
        
        
        
        

Resource Safety

class ResourceSafety:
    """Ensure safe resource management"""
    
    @contextmanager
    def managed_resource(self, resource, cleanup: Callable = None):
        """Safely manage resource lifecycle"""
        try:
            yield resource
        finally:
            if cleanup:
                cleanup()
            elif hasattr(resource, "close"):
                resource.close()
    
    def safe_allocation(self, size: int, max_size: int) -> bytes:
        """Safely allocate memory"""
        if size > max_size:
            raise MemoryError(f"Requested {size} exceeds max {max_size}")
        
        return np.empty(size, dtype=np.uint8)

class ConcurrencySafety:
    """Thread-safe operations"""
    
    def __init__(self):
        self.lock = threading.RLock()
    
    @contextmanager
    def critical_section(self):
        """Protect critical section"""
        with self.lock:
            yield
    
    def () -> :
        
         .lock:
             update_fn()