Python Functions & Builtin Functions — Tasks & Practice¶

def celsius_to_fahrenheit(c):
    """Convert Celsius to Fahrenheit"""
    return c * 9 / 5 + 32

def fahrenheit_to_celsius(f):
    """Convert Fahrenheit to Celsius"""
    return (f - 32) * 5 / 9

def celsius_to_kelvin(c):
    """Convert Celsius to Kelvin"""
    return c + 273.15

def kelvin_to_celsius(k):
    """Convert Kelvin to Celsius"""
    return k - 273.15

def convert(value, from_unit, to_unit):
    """
    Convert temperature between units.
    Supported units: 'C', 'F', 'K'
    """
    from_unit = from_unit.upper()
    to_unit = to_unit.upper()

    valid_units = {'C', 'F', 'K'}
    if from_unit not in valid_units or to_unit not in valid_units:
        raise ValueError(f"Invalid unit. Supported: {valid_units}")

    if from_unit == to_unit:
        return value

    # Convert to Celsius first (as intermediate)
    if from_unit == 'C':
        celsius = value
    elif from_unit == 'F':
        celsius = fahrenheit_to_celsius(value)
    else:  # K
        celsius = kelvin_to_celsius(value)

    # Convert from Celsius to target
    if to_unit == 'C':
        return celsius
    elif to_unit == 'F':
        return celsius_to_fahrenheit(celsius)
    else:  # K
        return celsius_to_kelvin(celsius)

# Tests
print("=== Individual Functions ===")
print(f"100°C -> {celsius_to_fahrenheit(100)}°F")     # 212.0
print(f"212°F -> {fahrenheit_to_celsius(212)}°C")      # 100.0
print(f"0°C -> {celsius_to_kelvin(0)}K")               # 273.15
print(f"273.15K -> {kelvin_to_celsius(273.15)}°C")     # 0.0

print("\n=== Unified Convert ===")
print(f"100°C -> F: {convert(100, 'C', 'F')}")         # 212.0
print(f"32°F -> K: {convert(32, 'F', 'K')}")           # 273.15
print(f"373.15K -> F: {convert(373.15, 'K', 'F')}")    # 212.0
print(f"0°C -> C: {convert(0, 'C', 'C')}")             # 0

# Error handling
try:
    convert(100, 'C', 'X')
except ValueError as e:
    print(f"\nError: {e}")

def mean(numbers):
    """Calculate the arithmetic mean"""
    if not numbers:
        raise ValueError("Cannot compute mean of empty list")
    return sum(numbers) / len(numbers)

def median(numbers):
    """Calculate the median"""
    if not numbers:
        raise ValueError("Cannot compute median of empty list")
    sorted_nums = sorted(numbers)
    n = len(sorted_nums)
    mid = n // 2
    if n % 2 == 0:
        return (sorted_nums[mid - 1] + sorted_nums[mid]) / 2
    return sorted_nums[mid]

def mode(numbers):
    """Calculate the mode (most frequent values)"""
    if not numbers:
        raise ValueError("Cannot compute mode of empty list")
    freq = {}
    for num in numbers:
        freq[num] = freq.get(num, 0) + 1
    max_count = max(freq.values())
    modes = [num for num, count in freq.items() if count == max_count]
    return modes if len(modes) > 1 else modes[0]

def std_dev(numbers):
    """Calculate the population standard deviation"""
    if not numbers:
        raise ValueError("Cannot compute std_dev of empty list")
    avg = mean(numbers)
    variance = sum((x - avg) ** 2 for x in numbers) / len(numbers)
    return variance ** 0.5

def summary(numbers):
    """Return a dictionary with all statistics"""
    return {
        "count": len(numbers),
        "min": min(numbers),
        "max": max(numbers),
        "mean": round(mean(numbers), 4),
        "median": median(numbers),
        "mode": mode(numbers),
        "std_dev": round(std_dev(numbers), 4),
        "sum": sum(numbers),
    }

# Tests
data = [4, 8, 6, 5, 3, 7, 8, 9, 2, 8]

print("=== Individual Functions ===")
print(f"Data: {data}")
print(f"Mean:    {mean(data)}")
print(f"Median:  {median(data)}")
print(f"Mode:    {mode(data)}")
print(f"Std Dev: {std_dev(data):.4f}")

print("\n=== Summary ===")
stats = summary(data)
for key, value in stats.items():
    print(f"  {key:10s}: {value}")

# Edge cases
print("\n=== Edge Cases ===")
print(f"Single element: {summary([5])}")
print(f"Two elements: {summary([1, 2])}")
print(f"All same: mode={mode([7, 7, 7, 7])}")
print(f"Multi-modal: mode={mode([1, 1, 2, 2, 3])}")

try:
    mean([])
except ValueError as e:
    print(f"Empty list: {e}")

def is_palindrome(s):
    """Check if string is a palindrome (ignoring case and spaces)"""
    cleaned = ''.join(c.lower() for c in s if c.isalnum())
    return cleaned == cleaned[::-1]

def count_vowels(s):
    """Count vowels in a string"""
    vowels = set('aeiouAEIOU')
    return sum(1 for c in s if c in vowels)

def title_case(s):
    """Convert to title case without using .title()"""
    words = s.split()
    result = []
    for word in words:
        if word:
            result.append(word[0].upper() + word[1:].lower())
    return ' '.join(result)

def truncate(s, max_len, suffix="..."):
    """Truncate string to max_len including suffix"""
    if len(s) <= max_len:
        return s
    return s[:max_len - len(suffix)] + suffix

def word_frequency(s):
    """Count word frequencies (case-insensitive)"""
    words = s.lower().split()
    # Remove punctuation from each word
    cleaned_words = []
    for word in words:
        cleaned = ''.join(c for c in word if c.isalnum())
        if cleaned:
            cleaned_words.append(cleaned)

    freq = {}
    for word in cleaned_words:
        freq[word] = freq.get(word, 0) + 1
    return dict(sorted(freq.items(), key=lambda x: -x[1]))

# Tests
print("=== is_palindrome ===")
for s in ["racecar", "A man a plan a canal Panama", "hello", "Was it a car or a cat I saw"]:
    print(f"  '{s}' -> {is_palindrome(s)}")

print("\n=== count_vowels ===")
for s in ["Hello World", "rhythm", "aeiou"]:
    print(f"  '{s}' -> {count_vowels(s)}")

print("\n=== title_case ===")
for s in ["hello world", "the QUICK brown FOX", "python programming"]:
    print(f"  '{s}' -> '{title_case(s)}'")

print("\n=== truncate ===")
long_text = "This is a very long string that needs to be truncated"
for length in [20, 30, 100]:
    print(f"  max={length}: '{truncate(long_text, length)}'")
print(f"  custom suffix: '{truncate(long_text, 25, suffix=' [...]')}'")

print("\n=== word_frequency ===")
text = "the cat sat on the mat the cat ate the rat"
freq = word_frequency(text)
for word, count in freq.items():
    print(f"  {word}: {count}")

# 1. Flatten a list of lists
nested = [[1, 2], [3, 4], [5, 6]]
flat = [x for sublist in nested for x in sublist]
# or: list(sum(nested, []))  — but less efficient
print(f"1. Flatten: {flat}")

# 2. Find the longest string
words = ["python", "java", "javascript", "go", "rust"]
longest = max(words, key=len)
print(f"2. Longest: {longest}")

# 3. Check if all numbers are positive
numbers = [1, 5, 3, 8, 2]
all_positive = all(n > 0 for n in numbers)
print(f"3. All positive: {all_positive}")

# 4. Filter passing students
names = ["Alice", "Bob", "Charlie", "Diana"]
scores = [85, 42, 91, 58]
passing = [name for name, score in zip(names, scores) if score >= 60]
print(f"4. Passing: {passing}")

# 5. Create dict from two lists
keys = ["a", "b", "c"]
values = [1, 2, 3]
d = dict(zip(keys, values))
print(f"5. Dict: {d}")

# 6. Unique elements preserving order
items = [3, 1, 4, 1, 5, 9, 2, 6, 5, 3, 5]
unique = list(dict.fromkeys(items))
print(f"6. Unique: {unique}")

# 7. Transpose a matrix
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
transposed = list(map(list, zip(*matrix)))
print(f"7. Transposed: {transposed}")

# 8. Second largest
nums = [10, 5, 8, 20, 15]
second = sorted(set(nums))[-2]
print(f"8. Second largest: {second}")

import functools
import time
import json

# 1. Timer decorator
def timer(func):
    """Log execution time of a function"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"[TIMER] {func.__name__} took {elapsed:.6f}s")
        return result
    return wrapper

# 2. Retry decorator with arguments
def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    """Retry a function on failure"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_error = e
                    if attempt < max_attempts:
                        print(f"[RETRY] {func.__name__} attempt {attempt}/{max_attempts} "
                              f"failed: {e}. Retrying in {delay}s...")
                        time.sleep(delay)
                    else:
                        print(f"[RETRY] {func.__name__} failed after {max_attempts} attempts")
            raise last_error
        return wrapper
    return decorator

# 3. Type validation decorator
def validate_types(**type_map):
    """Validate argument types at runtime"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            import inspect
            sig = inspect.signature(func)
            bound = sig.bind(*args, **kwargs)
            bound.apply_defaults()

            for param_name, expected_type in type_map.items():
                if param_name in bound.arguments:
                    value = bound.arguments[param_name]
                    if not isinstance(value, expected_type):
                        raise TypeError(
                            f"Argument '{param_name}' expected {expected_type.__name__}, "
                            f"got {type(value).__name__}: {value!r}"
                        )
            return func(*args, **kwargs)
        return wrapper
    return decorator

# 4. Memoize decorator (handles unhashable args)
def memoize(func):
    """Cache function results, handles unhashable args gracefully"""
    cache = {}

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # Try to create a hashable key
        try:
            key = (args, tuple(sorted(kwargs.items())))
            hash(key)  # Test if hashable
        except TypeError:
            # Unhashable args — fall back to JSON key or skip cache
            try:
                key = json.dumps((args, kwargs), sort_keys=True, default=str)
            except (TypeError, ValueError):
                # Cannot cache — just call the function
                return func(*args, **kwargs)

        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]

    wrapper.cache = cache
    wrapper.cache_clear = lambda: cache.clear()
    return wrapper

# 5. Rate limit decorator
def rate_limit(calls=5, period=60.0):
    """Limit function to N calls per time period (seconds)"""
    def decorator(func):
        call_times = []

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            # Remove old entries
            while call_times and call_times[0] < now - period:
                call_times.pop(0)

            if len(call_times) >= calls:
                wait_time = call_times[0] + period - now
                raise RuntimeError(
                    f"Rate limit exceeded for {func.__name__}. "
                    f"Max {calls} calls per {period}s. "
                    f"Try again in {wait_time:.1f}s"
                )

            call_times.append(now)
            return func(*args, **kwargs)

        wrapper.reset = lambda: call_times.clear()
        return wrapper
    return decorator


# ===== TESTS =====

# Test @timer
@timer
def slow_function():
    time.sleep(0.05)
    return "done"

print("=== @timer ===")
slow_function()

# Test @retry
print("\n=== @retry ===")
call_count = 0

@retry(max_attempts=3, delay=0.1, exceptions=(ValueError,))
def flaky_function():
    global call_count
    call_count += 1
    if call_count < 3:
        raise ValueError("Not ready yet!")
    return "success!"

call_count = 0
print(flaky_function())

# Test @validate_types
print("\n=== @validate_types ===")

@validate_types(name=str, age=int)
def create_user(name, age):
    return {"name": name, "age": age}

print(create_user("Alice", 30))
try:
    create_user("Bob", "thirty")
except TypeError as e:
    print(f"Type error: {e}")

# Test @memoize
print("\n=== @memoize ===")

@memoize
def expensive_calculation(n):
    print(f"  Computing for n={n}...")
    time.sleep(0.01)
    return n ** 2

print(expensive_calculation(5))   # Computes
print(expensive_calculation(5))   # Cached
print(expensive_calculation(10))  # Computes
print(f"Cache size: {len(expensive_calculation.cache)}")

# Test with unhashable args
@memoize
def process_list(data):
    print(f"  Processing list...")
    return sum(data)

print(process_list([1, 2, 3]))  # Lists are unhashable, but our memoize handles it
print(process_list([1, 2, 3]))  # Cached via JSON key

# Test @rate_limit
print("\n=== @rate_limit ===")

@rate_limit(calls=3, period=1.0)
def api_call(endpoint):
    return f"Response from {endpoint}"

for i in range(3):
    print(api_call(f"/api/{i}"))

try:
    api_call("/api/extra")  # Should fail
except RuntimeError as e:
    print(f"Rate limited: {e}")

from typing import Callable, Any, Iterable, Optional

class PipelineError(Exception):
    """Custom exception for pipeline failures"""
    def __init__(self, stage, original_error, value):
        self.stage = stage
        self.original_error = original_error
        self.value = value
        super().__init__(f"Pipeline failed at stage '{stage}': {original_error}")

class Pipeline:
    """A composable, lazy function pipeline"""

    def __init__(self):
        self._steps = []

    def pipe(self, func: Callable, name: str = None) -> 'Pipeline':
        """Add a transformation step"""
        self._steps.append(('pipe', func, name or func.__name__))
        return self

    def filter(self, predicate: Callable, name: str = None) -> 'Pipeline':
        """Add a filter step (for iterables)"""
        self._steps.append(('filter', predicate, name or predicate.__name__))
        return self

    def branch(self, condition: Callable,
               if_true: Callable, if_false: Callable,
               name: str = "branch") -> 'Pipeline':
        """Conditional transformation"""
        self._steps.append(('branch', (condition, if_true, if_false), name))
        return self

    def catch(self, handler: Callable, name: str = "error_handler") -> 'Pipeline':
        """Add error handling for the previous step"""
        self._steps.append(('catch', handler, name))
        return self

    def tap(self, func: Callable, name: str = "tap") -> 'Pipeline':
        """Execute a side effect without modifying the value"""
        self._steps.append(('tap', func, name))
        return self

    def execute(self, data: Any) -> Any:
        """Execute the pipeline on data"""
        result = data
        error_handler = None

        for step_type, func, name in self._steps:
            try:
                if step_type == 'pipe':
                    if isinstance(result, (list, tuple)):
                        result = type(result)(func(item) for item in result)
                    else:
                        result = func(result)

                elif step_type == 'filter':
                    if isinstance(result, (list, tuple)):
                        result = type(result)(item for item in result if func(item))
                    else:
                        if not func(result):
                            return None  # Filtered out

                elif step_type == 'branch':
                    condition, if_true, if_false = func
                    if isinstance(result, (list, tuple)):
                        result = type(result)(
                            if_true(item) if condition(item) else if_false(item)
                            for item in result
                        )
                    else:
                        if condition(result):
                            result = if_true(result)
                        else:
                            result = if_false(result)

                elif step_type == 'tap':
                    func(result)  # Side effect only

                elif step_type == 'catch':
                    error_handler = func
                    continue

            except Exception as e:
                if error_handler:
                    result = error_handler(e, result)
                    error_handler = None
                else:
                    raise PipelineError(name, e, result)

        return result

    def execute_each(self, data: Iterable) -> list:
        """Execute the pipeline on each item individually"""
        return [self.execute(item) for item in data]

    def __repr__(self):
        steps = " -> ".join(name for _, _, name in self._steps)
        return f"Pipeline({steps})"


# ===== TESTS =====

# Basic pipeline
print("=== Basic Pipeline ===")
result = (
    Pipeline()
    .pipe(lambda x: x.strip())
    .pipe(lambda x: x.lower())
    .pipe(lambda x: x.replace(" ", "_"))
    .execute("  Hello World  ")
)
print(f"Result: {result}")  # hello_world

# Pipeline with list
print("\n=== List Pipeline ===")
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
result = (
    Pipeline()
    .pipe(lambda x: x * 2)
    .filter(lambda x: x > 5)
    .pipe(lambda x: x + 100)
    .execute(numbers)
)
print(f"Result: {result}")  # [106, 108, 110, 112, 114, 116, 118, 120]

# Pipeline with branching
print("\n=== Branching Pipeline ===")
data = [1, -2, 3, -4, 5, -6]
result = (
    Pipeline()
    .branch(
        condition=lambda x: x > 0,
        if_true=lambda x: x ** 2,
        if_false=lambda x: abs(x)
    )
    .pipe(lambda x: f"val={x}")
    .execute(data)
)
print(f"Result: {result}")

# Pipeline with error handling
print("\n=== Error Handling Pipeline ===")
result = (
    Pipeline()
    .pipe(lambda x: int(x), name="parse_int")
    .catch(lambda e, val: 0)  # Default to 0 on parse error
    .pipe(lambda x: x * 10)
    .execute("not_a_number")
)
print(f"Result: {result}")  # 0 (error caught, defaulted)

# Pipeline with tap (debugging)
print("\n=== Tap Pipeline ===")
result = (
    Pipeline()
    .pipe(lambda x: x ** 2)
    .tap(lambda x: print(f"  [DEBUG] After square: {x}"))
    .pipe(lambda x: x + 10)
    .tap(lambda x: print(f"  [DEBUG] After add 10: {x}"))
    .execute(5)
)
print(f"Result: {result}")

# Pipeline representation
p = (
    Pipeline()
    .pipe(str.strip, "strip")
    .pipe(str.lower, "lower")
    .filter(lambda x: len(x) > 0, "non_empty")
)
print(f"\n{repr(p)}")

# Sample dataset
students = [
    {"name": "Alice", "grades": [90, 85, 92, 88], "age": 20},
    {"name": "Bob", "grades": [75, 60, 70, 65], "age": 22},
    {"name": "Charlie", "grades": [95, 98, 100, 97], "age": 19},
    {"name": "Diana", "grades": [50, 45, 55, 60], "age": 21},
    {"name": "Eve", "grades": [80, 85, 75, 90], "age": 20},
    {"name": "Frank", "grades": [88, 92, 85, 90], "age": 23},
]

# 1. Calculate average grade for each student using map
averages = list(map(lambda s: {**s, "avg": sum(s["grades"]) / len(s["grades"])}, students))
print("=== 1. Averages ===")
for s in averages:
    print(f"  {s['name']}: {s['avg']:.1f}")

# 2. Filter students with average >= 80
honor_roll = list(filter(lambda s: s["avg"] >= 80, averages))
print(f"\n=== 2. Honor Roll ===")
print(f"  {list(map(lambda s: s['name'], honor_roll))}")

# 3. Sort by average (descending)
ranked = sorted(averages, key=lambda s: s["avg"], reverse=True)
print(f"\n=== 3. Ranked ===")
for rank, s in enumerate(ranked, 1):
    print(f"  #{rank} {s['name']}: {s['avg']:.1f}")

# 4. Check if ALL students passed (avg >= 50)
all_passed = all(map(lambda s: s["avg"] >= 50, averages))
print(f"\n=== 4. All passed: {all_passed} ===")

# 5. Check if ANY student got a perfect score (100) in any subject
any_perfect = any(any(g == 100 for g in s["grades"]) for s in students)
print(f"\n=== 5. Any perfect score: {any_perfect} ===")

# 6. Pair students by rank (1st with last, 2nd with 2nd-to-last, etc.)
names_ranked = list(map(lambda s: s["name"], ranked))
pairs = list(zip(names_ranked, reversed(names_ranked)))
print(f"\n=== 6. Study Pairs ===")
for strong, needs_help in pairs[:len(pairs)//2]:
    print(f"  {strong} tutors {needs_help}")

# 7. Class statistics
all_grades = [g for s in students for g in s["grades"]]
print(f"\n=== 7. Class Statistics ===")
print(f"  Highest grade: {max(all_grades)}")
print(f"  Lowest grade:  {min(all_grades)}")
print(f"  Total grades:  {len(all_grades)}")
print(f"  Class average: {sum(all_grades) / len(all_grades):.1f}")

# 8. Create a grade distribution using builtins
grade_letters = list(map(
    lambda g: 'A' if g >= 90 else 'B' if g >= 80 else 'C' if g >= 70 else 'D' if g >= 60 else 'F',
    all_grades
))
distribution = {letter: grade_letters.count(letter) for letter in sorted(set(grade_letters))}
print(f"\n=== 8. Grade Distribution ===")
for letter, count in distribution.items():
    bar = '#' * count
    print(f"  {letter}: {bar} ({count})")

# 9. Find the student with the most consistent grades (lowest std dev)
def variance(grades):
    avg = sum(grades) / len(grades)
    return sum((g - avg) ** 2 for g in grades) / len(grades)

most_consistent = min(students, key=lambda s: variance(s["grades"]))
print(f"\n=== 9. Most Consistent: {most_consistent['name']} ===")
print(f"  Grades: {most_consistent['grades']}")

# 10. Create a formatted report using only builtins
print(f"\n=== 10. Final Report ===")
header = f"{'Name':<12} {'Avg':>6} {'Min':>4} {'Max':>4} {'Grade':>6}"
print(f"  {header}")
print(f"  {'-' * len(header)}")
for s in ranked:
    avg = s["avg"]
    letter = 'A' if avg >= 90 else 'B' if avg >= 80 else 'C' if avg >= 70 else 'D' if avg >= 60 else 'F'
    print(f"  {s['name']:<12} {avg:>6.1f} {min(s['grades']):>4} {max(s['grades']):>4} {letter:>6}")

import inspect
import functools
from typing import Any, Callable, Dict, Optional, Type, TypeVar, get_type_hints

T = TypeVar('T')

class CircularDependencyError(Exception):
    pass

class ServiceNotFoundError(Exception):
    pass

class DIContainer:
    """Lightweight Dependency Injection Container"""

    SINGLETON = "singleton"
    TRANSIENT = "transient"
    SCOPED = "scoped"

    def __init__(self, parent: Optional['DIContainer'] = None):
        self._registry: Dict[type, dict] = {}
        self._singletons: Dict[type, Any] = {}
        self._scoped_instances: Dict[type, Any] = {}
        self._resolving: set = set()  # For circular dependency detection
        self._parent = parent

    def register(self, interface: type, implementation: type = None,
                 lifetime: str = TRANSIENT, factory: Callable = None):
        """Register a service"""
        if implementation is None and factory is None:
            implementation = interface

        self._registry[interface] = {
            "implementation": implementation,
            "lifetime": lifetime,
            "factory": factory,
        }
        return self

    def singleton(self, interface: type, implementation: type = None):
        """Shorthand for singleton registration"""
        return self.register(interface, implementation, lifetime=self.SINGLETON)

    def transient(self, interface: type, implementation: type = None):
        """Shorthand for transient registration"""
        return self.register(interface, implementation, lifetime=self.TRANSIENT)

    def resolve(self, interface: type) -> Any:
        """Resolve a service, injecting dependencies automatically"""
        # Check for circular dependencies
        if interface in self._resolving:
            chain = " -> ".join(t.__name__ for t in self._resolving)
            raise CircularDependencyError(
                f"Circular dependency detected: {chain} -> {interface.__name__}"
            )

        # Look up registration
        reg = self._registry.get(interface)
        if reg is None:
            if self._parent:
                return self._parent.resolve(interface)
            raise ServiceNotFoundError(f"No registration for {interface.__name__}")

        lifetime = reg["lifetime"]

        # Check caches
        if lifetime == self.SINGLETON and interface in self._singletons:
            return self._singletons[interface]
        if lifetime == self.SCOPED and interface in self._scoped_instances:
            return self._scoped_instances[interface]

        # Resolve
        self._resolving.add(interface)
        try:
            if reg["factory"]:
                instance = reg["factory"](self)
            else:
                instance = self._create_instance(reg["implementation"])
        finally:
            self._resolving.discard(interface)

        # Cache if needed
        if lifetime == self.SINGLETON:
            self._singletons[interface] = instance
        elif lifetime == self.SCOPED:
            self._scoped_instances[interface] = instance

        return instance

    def _create_instance(self, cls: type) -> Any:
        """Create an instance with automatic constructor injection"""
        try:
            hints = get_type_hints(cls.__init__)
        except Exception:
            hints = {}

        sig = inspect.signature(cls.__init__)
        kwargs = {}

        for name, param in sig.parameters.items():
            if name == 'self':
                continue

            # Get type from hints
            param_type = hints.get(name)
            if param_type and param_type in self._registry:
                kwargs[name] = self.resolve(param_type)
            elif param.default is not inspect.Parameter.empty:
                kwargs[name] = param.default
            else:
                raise ServiceNotFoundError(
                    f"Cannot resolve parameter '{name}' of type "
                    f"'{param_type}' for {cls.__name__}"
                )

        return cls(**kwargs)

    def create_scope(self) -> 'DIContainer':
        """Create a child scope"""
        child = DIContainer(parent=self)
        child._registry = {**self._registry}
        child._singletons = self._singletons  # Shared!
        return child

    # Decorator interface
    def injectable(self, lifetime: str = TRANSIENT):
        """Decorator to auto-register a class"""
        def decorator(cls):
            self.register(cls, cls, lifetime)
            return cls
        return decorator


# ===== USAGE DEMO =====

# Define interfaces (using abstract-like classes)
class Logger:
    def log(self, msg: str): ...

class Database:
    def query(self, sql: str): ...

class UserRepository:
    def find(self, user_id: int): ...

# Implementations
class ConsoleLogger(Logger):
    def __init__(self):
        self.logs = []

    def log(self, msg: str):
        self.logs.append(msg)
        print(f"  [LOG] {msg}")

class PostgresDB(Database):
    def __init__(self, logger: Logger):
        self.logger = logger

    def query(self, sql: str):
        self.logger.log(f"Executing: {sql}")
        return [{"id": 1, "name": "Alice"}]

class UserRepo(UserRepository):
    def __init__(self, db: Database, logger: Logger):
        self.db = db
        self.logger = logger

    def find(self, user_id: int):
        self.logger.log(f"Finding user {user_id}")
        return self.db.query(f"SELECT * FROM users WHERE id = {user_id}")

class UserService:
    def __init__(self, repo: UserRepository, logger: Logger):
        self.repo = repo
        self.logger = logger

    def get_user(self, user_id: int):
        self.logger.log(f"UserService.get_user({user_id})")
        return self.repo.find(user_id)

# Configure container
container = DIContainer()
container.singleton(Logger, ConsoleLogger)
container.singleton(Database, PostgresDB)
container.transient(UserRepository, UserRepo)
container.transient(UserService)

# Resolve with automatic injection
print("=== Resolving UserService ===")
service = container.resolve(UserService)
result = service.get_user(1)
print(f"Result: {result}")

# Singleton verification
print("\n=== Singleton Test ===")
logger1 = container.resolve(Logger)
logger2 = container.resolve(Logger)
print(f"Same logger instance: {logger1 is logger2}")  # True

# Scoped lifetime
print("\n=== Scoped Lifetime ===")
container.register(UserRepository, UserRepo, lifetime=DIContainer.SCOPED)

scope1 = container.create_scope()
scope2 = container.create_scope()

repo_1a = scope1.resolve(UserRepository)
repo_1b = scope1.resolve(UserRepository)
repo_2 = scope2.resolve(UserRepository)

print(f"Same scope, same instance: {repo_1a is repo_1b}")    # True
print(f"Different scope, different: {repo_1a is not repo_2}")  # True

# Circular dependency detection
print("\n=== Circular Dependency Detection ===")

class ServiceA:
    def __init__(self, b: 'ServiceB'): self.b = b

class ServiceB:
    def __init__(self, a: ServiceA): self.a = a

circular_container = DIContainer()
circular_container.transient(ServiceA)
circular_container.transient(ServiceB)

try:
    circular_container.resolve(ServiceA)
except CircularDependencyError as e:
    print(f"Caught: {e}")

import functools
from typing import Callable, Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, field
from enum import Enum, auto

@dataclass
class Transition:
    event: str
    source: str
    target: str
    guard: Optional[Callable] = None
    action: Optional[Callable] = None

@dataclass
class StateConfig:
    on_enter: Optional[Callable] = None
    on_exit: Optional[Callable] = None

class StateMachine:
    """Function-based state machine with guards, actions, and history"""

    def __init__(self, initial_state: str):
        self._current_state = initial_state
        self._transitions: List[Transition] = []
        self._states: Dict[str, StateConfig] = {}
        self._history: List[Tuple[str, str, str]] = []  # (from, event, to)
        self._context: Dict[str, Any] = {}
        self._listeners: Dict[str, List[Callable]] = {}

    @property
    def state(self) -> str:
        return self._current_state

    @property
    def context(self) -> dict:
        return self._context

    def add_state(self, name: str, on_enter: Callable = None,
                  on_exit: Callable = None) -> 'StateMachine':
        """Register a state with optional enter/exit callbacks"""
        self._states[name] = StateConfig(on_enter=on_enter, on_exit=on_exit)
        return self

    def add_transition(self, event: str, source: str, target: str,
                       guard: Callable = None, action: Callable = None) -> 'StateMachine':
        """Add a transition"""
        self._transitions.append(Transition(event, source, target, guard, action))
        return self

    def on(self, event_name: str, listener: Callable) -> 'StateMachine':
        """Add an event listener"""
        self._listeners.setdefault(event_name, []).append(listener)
        return self

    def send(self, event: str, **payload) -> bool:
        """Send an event to the state machine"""
        matching = [
            t for t in self._transitions
            if t.event == event and t.source == self._current_state
        ]

        if not matching:
            print(f"  No transition for event '{event}' in state '{self._current_state}'")
            return False

        for transition in matching:
            # Check guard
            if transition.guard and not transition.guard(self._context, payload):
                continue

            old_state = self._current_state

            # Exit current state
            if old_state in self._states and self._states[old_state].on_exit:
                self._states[old_state].on_exit(self._context)

            # Execute transition action
            if transition.action:
                transition.action(self._context, payload)

            # Update state
            self._current_state = transition.target
            self._history.append((old_state, event, transition.target))

            # Enter new state
            if transition.target in self._states and self._states[transition.target].on_enter:
                self._states[transition.target].on_enter(self._context)

            # Notify listeners
            for listener in self._listeners.get('transition', []):
                listener(old_state, event, transition.target, self._context)

            return True

        print(f"  All guards failed for event '{event}'")
        return False

    def get_available_events(self) -> List[str]:
        """Get events available in current state"""
        return list(set(
            t.event for t in self._transitions
            if t.source == self._current_state
        ))

    def get_history(self) -> List[Tuple[str, str, str]]:
        """Get transition history"""
        return list(self._history)

    def visualize(self):
        """Print a text visualization of the state machine"""
        print(f"\n  Current: [{self._current_state}]")
        print(f"  Available events: {self.get_available_events()}")
        print(f"  History ({len(self._history)} transitions):")
        for old, event, new in self._history[-5:]:
            print(f"    {old} --({event})--> {new}")


# ===== DEMO: Order Processing State Machine =====

print("=== Order Processing State Machine ===\n")

sm = StateMachine("pending")

# Define states with enter/exit callbacks
sm.add_state("pending",
    on_enter=lambda ctx: print(f"  [ENTER pending] Order created"),
)
sm.add_state("confirmed",
    on_enter=lambda ctx: print(f"  [ENTER confirmed] Order confirmed, total=${ctx.get('total', 0)}"),
    on_exit=lambda ctx: print(f"  [EXIT confirmed] Preparing to process..."),
)
sm.add_state("processing",
    on_enter=lambda ctx: print(f"  [ENTER processing] Starting fulfillment"),
)
sm.add_state("shipped",
    on_enter=lambda ctx: print(f"  [ENTER shipped] Tracking: {ctx.get('tracking', 'N/A')}"),
)
sm.add_state("delivered",
    on_enter=lambda ctx: print(f"  [ENTER delivered] Order complete!"),
)
sm.add_state("cancelled",
    on_enter=lambda ctx: print(f"  [ENTER cancelled] Reason: {ctx.get('cancel_reason', 'N/A')}"),
)

# Define transitions with guards and actions
sm.add_transition("confirm", "pending", "confirmed",
    guard=lambda ctx, p: p.get("total", 0) > 0,
    action=lambda ctx, p: ctx.update({"total": p["total"], "items": p.get("items", [])}),
)

sm.add_transition("cancel", "pending", "cancelled",
    action=lambda ctx, p: ctx.update({"cancel_reason": p.get("reason", "Customer request")}),
)

sm.add_transition("cancel", "confirmed", "cancelled",
    action=lambda ctx, p: ctx.update({"cancel_reason": p.get("reason", "Customer request")}),
)

sm.add_transition("process", "confirmed", "processing",
    guard=lambda ctx, p: ctx.get("total", 0) > 0,
)

sm.add_transition("ship", "processing", "shipped",
    action=lambda ctx, p: ctx.update({"tracking": p.get("tracking", "TRK-000")}),
)

sm.add_transition("deliver", "shipped", "delivered")

# Add a listener
sm.on("transition", lambda old, event, new, ctx:
    print(f"  >> Transition: {old} --({event})--> {new}"))

# Run the state machine
print("--- Creating order ---")
sm.context.update({"order_id": "ORD-001"})

print(f"\nState: {sm.state}")
print(f"Available: {sm.get_available_events()}")

print("\n--- Confirming order ---")
sm.send("confirm", total=99.99, items=["Widget", "Gadget"])

print("\n--- Processing order ---")
sm.send("process")

print("\n--- Shipping order ---")
sm.send("ship", tracking="TRK-12345")

print("\n--- Delivering order ---")
sm.send("deliver")

# Try invalid transition
print("\n--- Try to ship delivered order ---")
sm.send("ship")

# Visualize
sm.visualize()

# Test guard failure
print("\n\n=== Guard Failure Test ===")
sm2 = StateMachine("pending")
sm2.add_transition("confirm", "pending", "confirmed",
    guard=lambda ctx, p: p.get("total", 0) > 0)

sm2.send("confirm", total=0)  # Guard fails
print(f"Still in: {sm2.state}")  # Still pending

sm2.send("confirm", total=50)  # Guard passes
print(f"Now in: {sm2.state}")   # confirmed