Python exercises

name = "Ada"
score = 91
print(name, score)

price = 12
quantity = 3
total = price * quantity
print(total)
print(7 / 2, 7 // 2, 7 % 2)
assert -7 // 2 == -4

scores = [82, 91, 76]
scores.append(95)
for score in scores:
    print(score)

def average(values):
    return sum(values) / len(values)

print(average([82, 91, 76, 95]))

def grade(score):
    if score >= 80:
        return "Distinction"
    elif score >= 50:
        return "Pass"
    else:
        return "Retry"

print(grade(80))
print(grade(49))
assert grade(50) == "Pass"

def total_of(values):
    total = 0
    for value in values:
        total += value
    return total

print(total_of([2, 4, 6]))
assert total_of([]) == 0
assert total_of([-2, 2]) == 0

def digit_sum(number):
    number = abs(number)
    total = 0
    while number > 0:
        total += number % 10
        number //= 10
    return total

print(digit_sum(204))
assert digit_sum(0) == 0
assert digit_sum(-204) == 6

matrix = [[1, 2, 3], [4, 5, 6]]
for row in matrix:
    row_total = 0
    for value in row:
        row_total += value
    print(row_total)

def is_palindrome(text):
    cleaned = text.lower().replace(" ", "")
    return cleaned == cleaned[::-1]

print(is_palindrome("Never odd or even"))
print(is_palindrome("Python"))
assert is_palindrome("") is True

def parse_age(text):
    try:
        age = int(text)
    except ValueError:
        return "Invalid integer"
    if age < 0:
        return "Age cannot be negative"
    return str(age)

print(parse_age("21"))
print(parse_age("abc"))
print(parse_age("-1"))
assert parse_age("0") == "0"

def factorial(n):
    if n < 0:
        raise ValueError("n must be nonnegative")
    if n == 0:
        return 1
    return n * factorial(n - 1)

print(factorial(4))
assert factorial(0) == 1
assert factorial(1) == 1

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as folder:
    path = Path(folder) / "notes.txt"
    path.write_text("Learn\nPractice\n", encoding="utf-8")
    with path.open(encoding="utf-8") as handle:
        lines = [line.strip() for line in handle]
    print(len(lines))
    print(lines[1])

class Counter:
    def __init__(self):
        self.value = 0
    def increment(self):
        self.value += 1

first = Counter()
second = Counter()
first.increment()
first.increment()
print(first.value, second.value)
assert second.value == 0

class Account:
    def __init__(self):
        self._balance = 0
    @property
    def balance(self):
        return self._balance
    def deposit(self, amount):
        if amount <= 0:
            raise ValueError("deposit must be positive")
        self._balance += amount

account = Account()
account.deposit(25)
print(account.balance)
try:
    account.deposit(0)
except ValueError:
    print("Rejected")
assert account.balance == 25

class Animal:
    def speak(self):
        return "sound"

class Dog(Animal):
    def speak(self):
        return "woof"

class Cat(Animal):
    def speak(self):
        return "meow"

for animal in [Dog(), Cat()]:
    print(animal.speak())

from abc import ABC, abstractmethod

class Shape(ABC):
    @abstractmethod
    def area(self):
        pass

class Rectangle(Shape):
    def __init__(self, width, height):
        self.width = width
        self.height = height
    def area(self):
        return self.width * self.height

shape = Rectangle(3, 4)
print(shape.area())
assert Rectangle(0, 4).area() == 0

class Engine:
    def start(self):
        return "Engine ready"

class Car:
    def __init__(self, engine):
        self.engine = engine
    def start(self):
        return self.engine.start() + "; car ready"

car = Car(Engine())
print(car.start())

def work(n):
    linear = 0
    quadratic = 0
    for _ in range(n):
        linear += 1
    for _ in range(n):
        for _ in range(n):
            quadratic += 1
    return linear, quadratic

print(work(3))
print(work(6))
assert work(0) == (0, 0)

first = {1, 2, 3}
second = {3, 4}
print(sorted(first & second))
print(sorted(first | second))
point = (2, 5)
x, y = point
print(x + y)
assert len({1, 1, 2}) == 2

def find(values, target):
    for index, value in enumerate(values):
        if value == target:
            return index
    return -1

print(find([4, 9, 2, 9], 9))
print(find([4, 9, 2, 9], 7))
assert find([], 9) == -1
assert find([9], 9) == 0

def binary_search(values, target):
    low, high = 0, len(values) - 1
    while low <= high:
        mid = low + (high - low) // 2
        if values[mid] == target:
            return mid
        if values[mid] < target:
            low = mid + 1
        else:
            high = mid - 1
    return -1

print(binary_search([2, 4, 6, 8, 10], 8))
print(binary_search([2, 4, 6, 8, 10], 7))
assert binary_search([], 1) == -1
assert binary_search([2], 2) == 0

def bubble_sort(values):
    result = values.copy()
    for end in range(len(result) - 1, 0, -1):
        swapped = False
        for i in range(end):
            if result[i] > result[i + 1]:
                result[i], result[i + 1] = result[i + 1], result[i]
                swapped = True
        if not swapped:
            break
    return result

print(bubble_sort([3, 2, 1]))
assert bubble_sort([]) == []
assert bubble_sort([1, 1, 2]) == [1, 1, 2]

def selection_sort(values):
    result = values.copy()
    for i in range(len(result)):
        smallest = i
        for j in range(i + 1, len(result)):
            if result[j] < result[smallest]:
                smallest = j
        result[i], result[smallest] = result[smallest], result[i]
    return result

print(selection_sort([3, 1, 2]))
assert selection_sort([]) == []
assert selection_sort([2, 2, 1]) == [1, 2, 2]

def insertion_sort(values):
    result = values.copy()
    for i in range(1, len(result)):
        key = result[i]
        j = i - 1
        while j >= 0 and result[j] > key:
            result[j + 1] = result[j]
            j -= 1
        result[j + 1] = key
    return result

print(insertion_sort([3, 1, 2]))
assert insertion_sort([]) == []
assert insertion_sort([2, 2, 1]) == [1, 2, 2]

def merge_sort(values):
    if len(values) <= 1:
        return values.copy()
    middle = len(values) // 2
    left = merge_sort(values[:middle])
    right = merge_sort(values[middle:])
    merged = []
    i = j = 0
    while i < len(left) and j < len(right):
        if left[i] <= right[j]:
            merged.append(left[i])
            i += 1
        else:
            merged.append(right[j])
            j += 1
    return merged + left[i:] + right[j:]

print(merge_sort([4, 1, 3, 2]))
assert merge_sort([]) == []
assert merge_sort([2, 1, 2]) == [1, 2, 2]

def quick_sort(values):
    if len(values) <= 1:
        return values.copy()
    pivot = values[len(values) // 2]
    lower = [value for value in values if value < pivot]
    equal = [value for value in values if value == pivot]
    upper = [value for value in values if value > pivot]
    return quick_sort(lower) + equal + quick_sort(upper)

print(quick_sort([3, 1, 2, 1]))
assert quick_sort([]) == []
assert quick_sort([2, 2]) == [2, 2]

def balanced(text):
    stack = []
    for char in text:
        if char == "(":
            stack.append(char)
        elif char == ")":
            if not stack:
                return False
            stack.pop()
    return not stack

print(balanced("(())"))
print(balanced(")("))
assert balanced("")
assert not balanced("(()")

from collections import deque

queue = deque(["A", "B"])
queue.append("C")
print(queue.popleft())
print(queue.popleft())
print(list(queue))
assert queue.popleft() == "C"
assert not queue

counts = {}
for word in "red blue red".split():
    counts[word] = counts.get(word, 0) + 1
for word in sorted(counts):
    print(word, counts[word])
assert counts.get("green", 0) == 0

class Node:
    def __init__(self, value, next_node=None):
        self.value = value
        self.next = next_node

head = Node(2, Node(3))
head = Node(1, head)
current = head
values = []
while current is not None:
    values.append(current.value)
    current = current.next
print(values)
assert values == [1, 2, 3]

class Node:
    def __init__(self, value):
        self.value = value
        self.left = self.right = None

def insert(root, value):
    if root is None:
        return Node(value)
    if value < root.value:
        root.left = insert(root.left, value)
    elif value > root.value:
        root.right = insert(root.right, value)
    return root

def inorder(root, values):
    if root is not None:
        inorder(root.left, values)
        values.append(root.value)
        inorder(root.right, values)

root = None
for value in [4, 2, 6, 1, 3]:
    root = insert(root, value)
values = []
inorder(root, values)
print(values)
assert values == [1, 2, 3, 4, 6]

import heapq

tasks = [(3, "backup"), (1, "alert"), (2, "email")]
heapq.heapify(tasks)
while tasks:
    priority, name = heapq.heappop(tasks)
    print(priority, name)
assert tasks == []

from collections import deque

def bfs(graph, start):
    seen = {start}
    queue = deque([start])
    order = []
    while queue:
        node = queue.popleft()
        order.append(node)
        for neighbor in graph.get(node, []):
            if neighbor not in seen:
                seen.add(neighbor)
                queue.append(neighbor)
    return order

graph = {"A": ["B", "C"], "B": ["A", "D"], "C": ["D"], "D": []}
print(bfs(graph, "A"))
assert bfs({}, "Z") == ["Z"]

def dfs(graph, start):
    seen = set()
    order = []
    def visit(node):
        if node in seen:
            return
        seen.add(node)
        order.append(node)
        for neighbor in graph.get(node, []):
            visit(neighbor)
    visit(start)
    return order

graph = {"A": ["B", "C"], "B": ["D"], "C": ["D"], "D": ["A"]}
print(dfs(graph, "A"))
assert dfs({}, "Z") == ["Z"]

import heapq

def shortest_paths(graph, start):
    distances = {start: 0}
    heap = [(0, start)]
    while heap:
        distance, node = heapq.heappop(heap)
        if distance != distances[node]:
            continue
        for neighbor, weight in graph.get(node, []):
            if weight < 0:
                raise ValueError("nonnegative weights required")
            candidate = distance + weight
            if candidate < distances.get(neighbor, float("inf")):
                distances[neighbor] = candidate
                heapq.heappush(heap, (candidate, neighbor))
    return distances

graph = {"A": [("B", 4), ("C", 1)], "C": [("B", 2)], "B": []}
result = shortest_paths(graph, "A")
print(sorted(result.items()))
assert shortest_paths({}, "A") == {"A": 0}

def fibonacci(n):
    if n < 0:
        raise ValueError("n must be nonnegative")
    a, b = 0, 1
    for _ in range(n):
        a, b = b, a + b
    return a

print(fibonacci(6))
assert fibonacci(0) == 0
assert fibonacci(1) == 1

def subsets(values):
    result = []
    path = []
    def visit(index):
        if index == len(values):
            result.append(path.copy())
            return
        visit(index + 1)
        path.append(values[index])
        visit(index + 1)
        path.pop()
    visit(0)
    return result

print(subsets([1, 2]))
assert subsets([]) == [[]]

def count_log(n):
    steps = 0
    while n > 1:
        n //= 2
        steps += 1
    return steps


def count_linear(n):
    steps = 0
    for _ in range(n):
        steps += 1
    return steps


def count_quadratic(n):
    steps = 0
    for _ in range(n):
        for _ in range(n):
            steps += 1
    return steps


for n in (8, 16, 32):
    print(n, count_log(n), count_linear(n), count_quadratic(n))
assert count_log(1024) == 10

from functools import lru_cache


@lru_cache(maxsize=None)
def t(n):
    if n == 1:
        return 0
    return 2 * t(n // 2) + n


def closed_form(n):
    return n * (n.bit_length() - 1)


for n in (1, 2, 4, 8, 16, 1024):
    print(n, t(n), closed_form(n))
    assert t(n) == closed_form(n)

class DynamicArray:
    def __init__(self):
        self.capacity = 1
        self.size = 0
        self.copies = 0

    def append(self):
        if self.size == self.capacity:
            self.capacity *= 2
            self.copies += self.size
        self.size += 1


array = DynamicArray()
for _ in range(16):
    array.append()
print(array.size, array.capacity, array.copies)
print(array.copies / array.size)
assert array.copies < 2 * array.size

def comparisons(items, target):
    count = 0
    for item in items:
        count += 1
        if item == target:
            break
    return count


items = [10, 20, 30, 40, 50]
best = comparisons(items, 10)
worst = comparisons(items, 50)
miss = comparisons(items, 99)
average = sum(comparisons(items, value) for value in items) / len(items)
print(best, worst, miss)
print(average)
assert average == (len(items) + 1) / 2

def insertion_sort(values):
    items = list(values)
    for i in range(1, len(items)):
        assert items[:i] == sorted(items[:i])
        key = items[i]
        j = i - 1
        while j >= 0 and items[j] > key:
            items[j + 1] = items[j]
            j -= 1
        items[j + 1] = key
    assert items == sorted(items)
    return items


print(insertion_sort([5, 2, 4, 6, 1, 3]))
print(insertion_sort([]))

calls = 0


def fib_slow(n):
    global calls
    calls += 1
    return n if n < 2 else fib_slow(n - 1) + fib_slow(n - 2)


def fib_memo(n, cache):
    global calls
    calls += 1
    if n in cache:
        return cache[n]
    cache[n] = n if n < 2 else fib_memo(n - 1, cache) + fib_memo(n - 2, cache)
    return cache[n]


calls = 0
slow = fib_slow(20)
slow_calls = calls
calls = 0
fast = fib_memo(20, {})
fast_calls = calls
print(slow, slow_calls)
print(fast, fast_calls)
assert slow == fast

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("PRAGMA foreign_keys = ON")
db.execute("CREATE TABLE department (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE)")
db.execute(
    "CREATE TABLE employee ("
    "id INTEGER PRIMARY KEY, name TEXT NOT NULL, "
    "salary INTEGER NOT NULL CHECK (salary > 0), "
    "department_id INTEGER NOT NULL REFERENCES department(id))"
)
db.execute("INSERT INTO department (id, name) VALUES (1, 'Engineering')")
db.execute("INSERT INTO employee (name, salary, department_id) VALUES ('Ada', 120, 1)")

bad_statements = [
    "INSERT INTO employee (name, salary, department_id) VALUES ('Bob', 100, 99)",
    "INSERT INTO employee (name, salary, department_id) VALUES ('Cy', -5, 1)",
    "INSERT INTO department (id, name) VALUES (2, 'Engineering')",
]
for statement in bad_statements:
    try:
        db.execute(statement)
    except sqlite3.IntegrityError as error:
        print(str(error).split(":")[0])
print(db.execute("SELECT COUNT(*) FROM employee").fetchone()[0])

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE sale (region TEXT, product TEXT, amount INTEGER)")
db.executemany(
    "INSERT INTO sale VALUES (?, ?, ?)",
    [
        ("north", "pen", 10),
        ("north", "ink", 25),
        ("south", "pen", 5),
        ("south", "ink", 40),
        ("south", "pad", 15),
        ("east", "pen", 8),
    ],
)
rows = db.execute(
    "SELECT region, COUNT(*) AS orders, SUM(amount) AS total "
    "FROM sale WHERE amount >= 8 "
    "GROUP BY region HAVING SUM(amount) > 20 "
    "ORDER BY total DESC"
).fetchall()
for row in rows:
    print(row)

import sqlite3

db = sqlite3.connect(":memory:")
db.executescript(
    """
    CREATE TABLE author (id INTEGER PRIMARY KEY, name TEXT);
    CREATE TABLE book (id INTEGER PRIMARY KEY, title TEXT, author_id INTEGER);
    INSERT INTO author VALUES (1, 'Ada'), (2, 'Grace'), (3, 'Linus');
    INSERT INTO book VALUES (1, 'Notes', 1), (2, 'Compilers', 2), (3, 'Languages', 2);
    """
)
inner = db.execute(
    "SELECT author.name, book.title FROM author "
    "JOIN book ON book.author_id = author.id ORDER BY author.name, book.title"
).fetchall()
left = db.execute(
    "SELECT author.name, COUNT(book.id) FROM author "
    "LEFT JOIN book ON book.author_id = author.id "
    "GROUP BY author.id ORDER BY author.id"
).fetchall()
print(inner)
print(left)

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE flat (order_id INTEGER, customer TEXT, city TEXT, item TEXT)")
db.executemany(
    "INSERT INTO flat VALUES (?, ?, ?, ?)",
    [(1, "Ada", "London", "pen"), (2, "Ada", "London", "ink"), (3, "Bo", "Paris", "pad")],
)
db.execute("UPDATE flat SET city = 'Cambridge' WHERE order_id = 1")
print(db.execute("SELECT DISTINCT city FROM flat WHERE customer = 'Ada' ORDER BY city").fetchall())

db.executescript(
    """
    CREATE TABLE customer (name TEXT PRIMARY KEY, city TEXT NOT NULL);
    CREATE TABLE purchase (
        order_id INTEGER PRIMARY KEY,
        customer TEXT NOT NULL REFERENCES customer(name),
        item TEXT NOT NULL
    );
    INSERT INTO customer VALUES ('Ada', 'London'), ('Bo', 'Paris');
    INSERT INTO purchase VALUES (1, 'Ada', 'pen'), (2, 'Ada', 'ink'), (3, 'Bo', 'pad');
    """
)
db.execute("UPDATE customer SET city = 'Cambridge' WHERE name = 'Ada'")
print(
    db.execute(
        "SELECT DISTINCT c.city FROM purchase p "
        "JOIN customer c ON c.name = p.customer WHERE p.customer = 'Ada'"
    ).fetchall()
)

import sqlite3

db = sqlite3.connect(":memory:", isolation_level=None)
db.execute("CREATE TABLE account (id INTEGER PRIMARY KEY, balance INTEGER NOT NULL CHECK (balance >= 0))")
db.execute("INSERT INTO account VALUES (1, 100), (2, 50)")


def transfer(source, target, amount):
    db.execute("BEGIN")
    try:
        db.execute("UPDATE account SET balance = balance + ? WHERE id = ?", (amount, target))
        db.execute("UPDATE account SET balance = balance - ? WHERE id = ?", (amount, source))
        db.execute("COMMIT")
        return True
    except sqlite3.IntegrityError:
        db.execute("ROLLBACK")
        return False


print(transfer(1, 2, 30))
print(transfer(1, 2, 500))
print(db.execute("SELECT id, balance FROM account ORDER BY id").fetchall())

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE person (id INTEGER PRIMARY KEY, email TEXT, age INTEGER)")
db.executemany(
    "INSERT INTO person (email, age) VALUES (?, ?)",
    [(f"user{i}@example.com", 20 + i % 40) for i in range(1000)],
)
query = "SELECT id FROM person WHERE email = 'user500@example.com'"


def plan(sql):
    return " ".join(row[3] for row in db.execute("EXPLAIN QUERY PLAN " + sql))


before = plan(query)
db.execute("CREATE INDEX idx_person_email ON person(email)")
after = plan(query)
print("SCAN" in before, "SEARCH" in after, "idx_person_email" in after)
print(db.execute(query).fetchall())

jobs = [("P1", 0, 8), ("P2", 1, 4), ("P3", 2, 9), ("P4", 3, 5)]


def simulate(jobs, pick):
    time = 0
    pending = sorted(jobs, key=lambda job: job[1])
    waits = {}
    order = []
    while pending:
        ready = [job for job in pending if job[1] <= time]
        if not ready:
            time = min(job[1] for job in pending)
            continue
        job = pick(ready)
        pending.remove(job)
        name, arrival, burst = job
        waits[name] = time - arrival
        time += burst
        order.append(name)
    return order, sum(waits.values()) / len(waits)


fcfs = simulate(jobs, lambda ready: min(ready, key=lambda job: job[1]))
sjf = simulate(jobs, lambda ready: min(ready, key=lambda job: (job[2], job[1])))
print(fcfs)
print(sjf)
assert sjf[1] < fcfs[1]

from collections import deque


def round_robin(bursts, quantum):
    queue = deque(bursts.items())
    time = 0
    finish = {}
    while queue:
        name, remaining = queue.popleft()
        run = min(quantum, remaining)
        time += run
        if remaining > run:
            queue.append((name, remaining - run))
        else:
            finish[name] = time
    return finish


bursts = {"A": 5, "B": 3, "C": 8}
finish = round_robin(bursts, 3)
print(finish)
waits = {name: finish[name] - bursts[name] for name in bursts}
print(waits)
assert sum(finish.values()) > 0

def fifo(refs, frames):
    memory, faults = [], 0
    for page in refs:
        if page not in memory:
            faults += 1
            if len(memory) == frames:
                memory.pop(0)
            memory.append(page)
    return faults


def lru(refs, frames):
    memory, faults = [], 0
    for page in refs:
        if page in memory:
            memory.remove(page)
        else:
            faults += 1
            if len(memory) == frames:
                memory.pop(0)
        memory.append(page)
    return faults


def optimal(refs, frames):
    memory, faults = [], 0
    for i, page in enumerate(refs):
        if page in memory:
            continue
        faults += 1
        if len(memory) == frames:
            future = refs[i + 1:]
            victim = max(memory, key=lambda p: future.index(p) if p in future else len(future) + 1)
            memory.remove(victim)
        memory.append(page)
    return faults


refs = [7, 0, 1, 2, 0, 3, 0, 4, 2, 3, 0, 3, 2, 1, 2, 0, 1, 7, 0, 1]
print(fifo(refs, 3), lru(refs, 3), optimal(refs, 3))
belady = [1, 2, 3, 4, 1, 2, 5, 1, 2, 3, 4, 5]
print(fifo(belady, 3), fifo(belady, 4))

def safe_sequence(available, allocation, maximum):
    need = [[m - a for m, a in zip(mx, al)] for mx, al in zip(maximum, allocation)]
    work = list(available)
    finished = [False] * len(allocation)
    order = []
    progress = True
    while progress:
        progress = False
        for i, row in enumerate(need):
            if not finished[i] and all(n <= w for n, w in zip(row, work)):
                work = [w + a for w, a in zip(work, allocation[i])]
                finished[i] = True
                order.append(f"P{i}")
                progress = True
    return order if all(finished) else None


def try_request(available, allocation, maximum, process, request):
    if any(r > a for r, a in zip(request, available)):
        return None
    new_available = [a - r for a, r in zip(available, request)]
    new_allocation = [row[:] for row in allocation]
    new_allocation[process] = [x + r for x, r in zip(new_allocation[process], request)]
    if safe_sequence(new_available, new_allocation, maximum) is None:
        return None
    return new_available, new_allocation


available = [3, 3, 2]
allocation = [[0, 1, 0], [2, 0, 0], [3, 0, 2], [2, 1, 1], [0, 0, 2]]
maximum = [[7, 5, 3], [3, 2, 2], [9, 0, 2], [2, 2, 2], [4, 3, 3]]
print(safe_sequence(available, allocation, maximum))

state = (available, allocation)
for process, request in ((1, [1, 0, 2]), (4, [3, 3, 0]), (0, [0, 2, 0])):
    outcome = try_request(state[0], state[1], maximum, process, request)
    print(f"P{process} {request}:", "granted" if outcome else "denied")
    if outcome:
        state = outcome

import queue
import threading

counter = 0
lock = threading.Lock()


def work():
    global counter
    for _ in range(10000):
        with lock:
            counter += 1


workers = [threading.Thread(target=work) for _ in range(4)]
for worker in workers:
    worker.start()
for worker in workers:
    worker.join()

buffer = queue.Queue(maxsize=2)
results = []


def producer():
    for item in range(1, 6):
        buffer.put(item)
    buffer.put(None)


def consumer():
    while True:
        item = buffer.get()
        if item is None:
            break
        results.append(item * item)


threads = [threading.Thread(target=producer), threading.Thread(target=consumer)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
print(results)
print(counter)

PAGE_SIZE = 4096
page_table = {0: 5, 1: 2, 2: None, 3: 7}


def translate(address):
    page, offset = divmod(address, PAGE_SIZE)
    if page not in page_table:
        return "segmentation fault"
    frame = page_table[page]
    if frame is None:
        return "page fault"
    return frame * PAGE_SIZE + offset


print(PAGE_SIZE.bit_length() - 1)
for address in (0, 4100, 8200, 16383, 20000):
    print(address, translate(address))

def encapsulate(payload):
    segment = {"src_port": 50000, "dst_port": 80, "data": payload}
    packet = {"src_ip": "192.0.2.10", "dst_ip": "198.51.100.7", "data": segment}
    frame = {"src_mac": "aa:aa:aa:aa:aa:01", "dst_mac": "bb:bb:bb:bb:bb:02", "data": packet}
    return frame


def decapsulate(frame):
    packet = frame["data"]
    segment = packet["data"]
    return segment["data"]


frame = encapsulate("GET /")
print(list(frame))
print(frame["data"]["dst_ip"], frame["data"]["data"]["dst_port"])
print(decapsulate(frame))

import ipaddress

network = ipaddress.ip_network("192.168.10.0/26")
print(network.netmask, network.num_addresses)
hosts = list(network.hosts())
print(hosts[0], hosts[-1], len(hosts))
print(ipaddress.ip_address("192.168.10.70") in network)
subnets = list(ipaddress.ip_network("192.168.10.0/24").subnets(new_prefix=26))
print([str(subnet) for subnet in subnets])

import socket
import threading

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(("127.0.0.1", 0))
server.listen(1)
port = server.getsockname()[1]


def serve():
    connection, _ = server.accept()
    with connection:
        data = connection.recv(1024)
        connection.sendall(data.upper())


thread = threading.Thread(target=serve)
thread.start()
with socket.create_connection(("127.0.0.1", port)) as client:
    client.sendall(b"hello tcp")
    reply = client.recv(1024)
thread.join()
server.close()
print(port > 0)
print(reply.decode())

import socket

receiver = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
receiver.bind(("127.0.0.1", 0))
address = receiver.getsockname()

sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sender.sendto(b"one", address)
sender.sendto(b"two", address)

first, _ = receiver.recvfrom(1024)
second, _ = receiver.recvfrom(1024)
sender.close()
receiver.close()
print(first, second)

request = "\r\n".join(["GET /index.html HTTP/1.1", "Host: example.com", "Accept: text/html", "", ""])
response = "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 5\r\n\r\nhello"


def parse_response(text):
    head, body = text.split("\r\n\r\n", 1)
    status_line, *header_lines = head.split("\r\n")
    version, status, reason = status_line.split(" ", 2)
    headers = {}
    for line in header_lines:
        name, value = line.split(":", 1)
        headers[name.strip().lower()] = value.strip()
    return int(status), reason, headers, body


method, target, version = request.split("\r\n")[0].split(" ")
print(method, target, version)
status, reason, headers, body = parse_response(response)
print(status, reason)
print(headers["content-length"], len(body.encode()))
print(body)

import ipaddress

routes = [
    (ipaddress.ip_network("0.0.0.0/0"), "default gateway"),
    (ipaddress.ip_network("10.0.0.0/8"), "corporate core"),
    (ipaddress.ip_network("10.1.0.0/16"), "branch office"),
    (ipaddress.ip_network("10.1.2.0/24"), "lab switch"),
]


def next_hop(destination):
    address = ipaddress.ip_address(destination)
    matches = [(network.prefixlen, hop) for network, hop in routes if address in network]
    return max(matches)[1]


for destination in ("10.1.2.9", "10.1.9.9", "10.200.0.1", "8.8.8.8"):
    print(destination, next_hop(destination))

import io
import unittest


def apply_discount(price_cents, percent):
    if not 0 <= percent <= 100:
        raise ValueError("percent must be between 0 and 100")
    return price_cents * (100 - percent) // 100


class DiscountTests(unittest.TestCase):
    def test_typical_discount(self):
        self.assertEqual(apply_discount(8000, 25), 6000)

    def test_zero_discount_keeps_price(self):
        self.assertEqual(apply_discount(1999, 0), 1999)

    def test_invalid_percent_is_rejected(self):
        with self.assertRaises(ValueError):
            apply_discount(1000, 120)


suite = unittest.defaultTestLoader.loadTestsFromTestCase(DiscountTests)
result = unittest.TextTestRunner(stream=io.StringIO()).run(suite)
print(result.testsRun, result.wasSuccessful())

def shipping_before(weight, express):
    if express == True:
        if weight <= 1:
            return 15
        else:
            if weight <= 5:
                return 25
            else:
                return 40
    else:
        if weight <= 1:
            return 5
        else:
            if weight <= 5:
                return 10
            else:
                return 20


RATES = [(1, 5, 15), (5, 10, 25)]


def shipping_after(weight, express):
    for limit, standard, fast in RATES:
        if weight <= limit:
            return fast if express else standard
    return 40 if express else 20


cases = [(w, e) for w in (0, 1, 1.5, 5, 5.1, 50) for e in (False, True)]
assert all(shipping_before(w, e) == shipping_after(w, e) for w, e in cases)
print(len(cases), "cases identical")
print(shipping_after(3, True), shipping_after(30, False))

class NoDiscount:
    def apply(self, amount):
        return amount


class PercentOff:
    def __init__(self, percent):
        self.percent = percent

    def apply(self, amount):
        return amount * (100 - self.percent) // 100


class FlatOff:
    def __init__(self, cents):
        self.cents = cents

    def apply(self, amount):
        return max(amount - self.cents, 0)


class Checkout:
    def __init__(self, rule):
        self.rule = rule

    def total(self, prices):
        return self.rule.apply(sum(prices))


cart = [1000, 2500]
for rule in (NoDiscount(), PercentOff(10), FlatOff(500)):
    print(type(rule).__name__, Checkout(rule).total(cart))

class EventBus:
    def __init__(self):
        self.handlers = {}

    def subscribe(self, event, handler):
        self.handlers.setdefault(event, []).append(handler)

    def publish(self, event, payload):
        for handler in self.handlers.get(event, []):
            handler(payload)


log = []
bus = EventBus()
bus.subscribe("order_placed", lambda order: log.append(f"email {order}"))
bus.subscribe("order_placed", lambda order: log.append(f"stock {order}"))
bus.publish("order_placed", "#42")
bus.publish("order_cancelled", "#43")
print(log)

import hashlib
import json

commits = {}


def commit(message, parents):
    body = json.dumps({"message": message, "parents": parents}, sort_keys=True)
    commit_id = hashlib.sha256(body.encode()).hexdigest()[:8]
    commits[commit_id] = {"message": message, "parents": parents}
    return commit_id


def ancestors(commit_id):
    seen = {commit_id}
    stack = [commit_id]
    while stack:
        for parent in commits[stack.pop()]["parents"]:
            if parent not in seen:
                seen.add(parent)
                stack.append(parent)
    return seen


def depth(commit_id):
    parents = commits[commit_id]["parents"]
    return 0 if not parents else 1 + max(depth(parent) for parent in parents)


root = commit("init", [])
parser = commit("add parser", [root])
main = commit("add tests", [parser])
feature = commit("add lexer", [parser])
merged = commit("merge feature", [main, feature])

common = ancestors(main) & ancestors(feature)
base = max(common, key=depth)
print(commits[base]["message"])
print(sorted(commits[c]["message"] for c in ancestors(merged)))
print(commit("init", []) == root)

class InsufficientFunds(Exception):
    pass


class Account:
    def __init__(self, balance):
        if balance < 0:
            raise ValueError("opening balance cannot be negative")
        self._balance = balance

    @property
    def balance(self):
        return self._balance

    def withdraw(self, amount):
        if amount <= 0:
            raise ValueError("amount must be positive")
        if amount > self._balance:
            raise InsufficientFunds(f"balance {self._balance} is less than {amount}")
        self._balance -= amount
        assert self._balance >= 0
        return self._balance


account = Account(100)
print(account.withdraw(30))
for amount in (0, 500):
    try:
        account.withdraw(amount)
    except InsufficientFunds as error:
        print("rejected:", error)
    except ValueError as error:
        print("invalid:", error)
print(account.balance)

import hashlib

print(hashlib.sha256(b"abc").hexdigest())
first = hashlib.sha256(b"transfer 100 to alice").hexdigest()
second = hashlib.sha256(b"transfer 900 to alice").hexdigest()
differing = sum(x != y for x, y in zip(first, second))
print(len(first), first == second)
print(differing > 40)

import hashlib
import hmac
import os

ITERATIONS = 600_000


def hash_password(password, salt):
    return hashlib.pbkdf2_hmac("sha256", password.encode(), salt, ITERATIONS)


def verify(password, salt, stored):
    return hmac.compare_digest(hash_password(password, salt), stored)


salt_a = os.urandom(16)
salt_b = os.urandom(16)
stored = hash_password("correct horse", salt_a)
print(verify("correct horse", salt_a, stored))
print(verify("wrong guess", salt_a, stored))
print(hash_password("correct horse", salt_a) == hash_password("correct horse", salt_b))
print(len(salt_a), len(stored))

import hashlib
import hmac

key = b"shared-secret-key"


def sign(message):
    return hmac.new(key, message, hashlib.sha256).hexdigest()


def verify(message, tag):
    return hmac.compare_digest(sign(message), tag)


message = b"amount=100&to=alice"
tag = sign(message)
print(verify(message, tag))
print(verify(b"amount=900&to=alice", tag))
print(verify(message, "0" * 64))
print(len(tag))

def xor(data, key):
    return bytes(d ^ k for d, k in zip(data, key))


plain1 = b"ATTACK AT DAWN"
plain2 = b"RETREAT AT TEN"
key = bytes(range(1, 15))
cipher1 = xor(plain1, key)
cipher2 = xor(plain2, key)
print(xor(cipher1, key) == plain1)
leak = xor(cipher1, cipher2)
print(leak == xor(plain1, plain2))
print(xor(leak, plain1))

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE users (name TEXT, secret TEXT)")
db.executemany("INSERT INTO users VALUES (?, ?)", [("ada", "s3cret-a"), ("bob", "s3cret-b")])


def unsafe(name):
    return db.execute("SELECT name FROM users WHERE name = '" + name + "'").fetchall()


def safe(name):
    return db.execute("SELECT name FROM users WHERE name = ?", (name,)).fetchall()


attack = "' OR '1'='1"
print(unsafe("ada"))
print(unsafe(attack))
print(safe(attack))

import random
import secrets
import string

first = random.Random(1234)
second = random.Random(1234)
print([first.randrange(10) for _ in range(5)] == [second.randrange(10) for _ in range(5)])

token = secrets.token_urlsafe(32)
alphabet = set(string.ascii_letters + string.digits + "-_")
print(len(token), set(token) <= alphabet)
print(secrets.token_urlsafe(32) != token)
print(secrets.compare_digest(token, token))

import time
from concurrent.futures import ThreadPoolExecutor


def fetch(n):
    time.sleep(0.2)
    return n * n


start = time.perf_counter()
sequential = [fetch(n) for n in range(5)]
sequential_time = time.perf_counter() - start

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:
    overlapped = list(pool.map(fetch, range(5)))
overlapped_time = time.perf_counter() - start

print(sequential == overlapped, overlapped)
print(overlapped_time < sequential_time / 2)

from multiprocessing import Pool


def count_primes(limit):
    return sum(1 for n in range(2, limit) if all(n % d for d in range(2, int(n ** 0.5) + 1)))


if __name__ == "__main__":
    limits = [20000, 20000, 20000, 20000]
    with Pool(2) as pool:
        results = pool.map(count_primes, limits)
    print(results)
    print(results == [count_primes(limit) for limit in limits])

def amdahl(parallel_fraction, processors):
    return 1 / ((1 - parallel_fraction) + parallel_fraction / processors)


def gustafson(parallel_fraction, processors):
    return (1 - parallel_fraction) + parallel_fraction * processors


for processors in (1, 2, 8, 64, 1024):
    print(processors, round(amdahl(0.9, processors), 2), round(gustafson(0.9, processors), 1))
print(round(1 / (1 - 0.9), 1))

from collections import Counter, defaultdict

documents = ["the cat sat", "the dog sat", "the cat ran"]


def map_phase(document):
    return [(word, 1) for word in document.split()]


def shuffle(pairs):
    groups = defaultdict(list)
    for key, value in pairs:
        groups[key].append(value)
    return groups


def reduce_phase(key, values):
    return key, sum(values)


pairs = [pair for document in documents for pair in map_phase(document)]
grouped = shuffle(pairs)
result = dict(reduce_phase(key, values) for key, values in sorted(grouped.items()))
print(result)
assert result == dict(Counter(" ".join(documents).split()))

class Process:
    def __init__(self, name):
        self.name = name
        self.clock = 0

    def local_event(self):
        self.clock += 1
        return self.clock

    def send(self):
        self.clock += 1
        return self.clock

    def receive(self, timestamp):
        self.clock = max(self.clock, timestamp) + 1
        return self.clock


a = Process("A")
b = Process("B")
a.local_event()
sent = a.send()
b.local_event()
received = b.receive(sent)
print(sent, received)
print(received > sent)
print(a.clock, b.clock)

from itertools import combinations

N = 3
replicas = [{"version": 0, "value": None} for _ in range(N)]


def write(value, version, targets):
    for i in targets:
        replicas[i] = {"version": version, "value": value}


def read(targets):
    return max((replicas[i] for i in targets), key=lambda r: r["version"])["value"]


write("v1", 1, [0, 1])
print(read([1, 2]))
print(read([2]))


def always_overlap(w, r):
    return all(set(ws) & set(rs) for ws in combinations(range(N), w) for rs in combinations(range(N), r))


print(always_overlap(2, 2))
print(always_overlap(2, 1))

def reflex_agent(location, dirty):
    if dirty:
        return "suck"
    return "right" if location == "A" else "left"


def run(world, location, steps):
    actions = []
    for _ in range(steps):
        action = reflex_agent(location, world[location])
        actions.append(action)
        if action == "suck":
            world[location] = False
        elif action == "right":
            location = "B"
        else:
            location = "A"
    return actions


world = {"A": True, "B": True}
print(run(world, "A", 4))
print(world)

from collections import deque


def solve(cap_a, cap_b, goal):
    start = (0, 0)
    parents = {start: None}
    queue = deque([start])
    while queue:
        state = queue.popleft()
        a, b = state
        if goal in state:
            path = []
            while state is not None:
                path.append(state)
                state = parents[state]
            return path[::-1]
        pour_ab = min(a, cap_b - b)
        pour_ba = min(b, cap_a - a)
        moves = ((cap_a, b), (a, cap_b), (0, b), (a, 0), (a - pour_ab, b + pour_ab), (a + pour_ba, b - pour_ba))
        for following in moves:
            if following not in parents:
                parents[following] = state
                queue.append(following)
    return None


path = solve(3, 5, 4)
print(len(path) - 1)
print(path)
print(solve(2, 4, 3))

import heapq

GRID = [
    "S.......",
    "........",
    "..####..",
    "........",
    "........",
    ".......G",
]


def find(symbol):
    for r, row in enumerate(GRID):
        if symbol in row:
            return (r, row.index(symbol))


START, GOAL = find("S"), find("G")


def manhattan(cell):
    return abs(cell[0] - GOAL[0]) + abs(cell[1] - GOAL[1])


def search(heuristic):
    frontier = [(heuristic(START), heuristic(START), 0, START)]
    best = {START: 0}
    expanded = 0
    while frontier:
        _, _, cost, cell = heapq.heappop(frontier)
        if cost > best[cell]:
            continue
        expanded += 1
        if cell == GOAL:
            return cost, expanded
        r, c = cell
        for nr, nc in ((r + 1, c), (r - 1, c), (r, c + 1), (r, c - 1)):
            if 0 <= nr < len(GRID) and 0 <= nc < len(GRID[0]) and GRID[nr][nc] != "#":
                new_cost = cost + 1
                if new_cost < best.get((nr, nc), 10**9):
                    best[(nr, nc)] = new_cost
                    h = heuristic((nr, nc))
                    heapq.heappush(frontier, (new_cost + h, h, new_cost, (nr, nc)))
    return None


print(search(manhattan))
print(search(lambda cell: 0))

LINES = [(0, 1, 2), (3, 4, 5), (6, 7, 8), (0, 3, 6), (1, 4, 7), (2, 5, 8), (0, 4, 8), (2, 4, 6)]
nodes = 0


def winner(board):
    for a, b, c in LINES:
        if board[a] != "." and board[a] == board[b] == board[c]:
            return board[a]
    return None


def terminal_score(board):
    won = winner(board)
    if won:
        return 1 if won == "X" else -1
    return 0 if "." not in board else None


def minimax(board, player):
    global nodes
    nodes += 1
    score = terminal_score(board)
    if score is not None:
        return score
    other = "O" if player == "X" else "X"
    scores = [minimax(board[:i] + player + board[i + 1:], other) for i, cell in enumerate(board) if cell == "."]
    return max(scores) if player == "X" else min(scores)


def alphabeta(board, player, alpha, beta):
    global nodes
    nodes += 1
    score = terminal_score(board)
    if score is not None:
        return score
    other = "O" if player == "X" else "X"
    value = -2 if player == "X" else 2
    for i, cell in enumerate(board):
        if cell != ".":
            continue
        child = alphabeta(board[:i] + player + board[i + 1:], other, alpha, beta)
        if player == "X":
            value = max(value, child)
            alpha = max(alpha, value)
        else:
            value = min(value, child)
            beta = min(beta, value)
        if alpha >= beta:
            break
    return value


start = "X...O...."
nodes = 0
plain = minimax(start, "X")
plain_nodes = nodes
nodes = 0
pruned = alphabeta(start, "X", -2, 2)
print(plain, pruned)
print(plain_nodes, nodes)

NEIGHBORS = {
    "WA": ["NT", "SA"],
    "NT": ["WA", "SA", "Q"],
    "SA": ["WA", "NT", "Q", "NSW", "V"],
    "Q": ["NT", "SA", "NSW"],
    "NSW": ["Q", "SA", "V"],
    "V": ["SA", "NSW"],
    "T": [],
}


def solve(colors):
    regions = list(NEIGHBORS)

    def backtrack(assignment):
        if len(assignment) == len(regions):
            return assignment
        region = next(r for r in regions if r not in assignment)
        for color in colors:
            if all(assignment.get(neighbor) != color for neighbor in NEIGHBORS[region]):
                result = backtrack({**assignment, region: color})
                if result:
                    return result
        return None

    return backtrack({})


solution = solve(["red", "green", "blue"])
print(dict(sorted(solution.items())))
print(all(solution[a] != solution[b] for a in NEIGHBORS for b in NEIGHBORS[a]))
print(solve(["red", "green"]))

def conflicts(board):
    n = len(board)
    return sum(
        1
        for i in range(n)
        for j in range(i + 1, n)
        if board[i] == board[j] or abs(board[i] - board[j]) == j - i
    )


def hill_climb(board):
    board = list(board)
    while True:
        current = conflicts(board)
        best, best_move = current, None
        for col in range(len(board)):
            original = board[col]
            for row in range(len(board)):
                if row != original:
                    board[col] = row
                    score = conflicts(board)
                    if score < best:
                        best, best_move = score, (col, row)
            board[col] = original
        if best_move is None:
            return board, current
        board[best_move[0]] = best_move[1]


state = 1


def next_start(n=8):
    global state
    board = []
    for _ in range(n):
        state = (state * 1103515245 + 12345) % 2**31
        board.append((state >> 16) % n)
    return board


restarts = 0
while True:
    board, score = hill_climb(next_start())
    if score == 0:
        break
    restarts += 1
print(restarts, board, score)

from itertools import product


def implies(a, b):
    return (not a) or b


def models_of(symbols, formula):
    assignments = (dict(zip(symbols, values)) for values in product([False, True], repeat=len(symbols)))
    return [model for model in assignments if formula(model)]


def entails(symbols, kb, query):
    return all(query(model) for model in models_of(symbols, kb))


symbols = ["rain", "wet", "cold"]
kb = lambda m: implies(m["rain"], m["wet"]) and m["rain"]
print(len(models_of(symbols, kb)))
print(entails(symbols, kb, lambda m: m["wet"]))
print(entails(symbols, kb, lambda m: m["cold"]))
print(entails(symbols, lambda m: False, lambda m: m["cold"]))

def forward_chain(rules, facts, query):
    known = set(facts)
    derived = []
    changed = True
    while changed:
        changed = False
        for premises, conclusion in rules:
            if conclusion not in known and all(p in known for p in premises):
                known.add(conclusion)
                derived.append(conclusion)
                changed = True
    return query in known, derived


rules = [(["P"], "Q"), (["L", "M"], "P"), (["B", "L"], "M"), (["A", "P"], "L"), (["A", "B"], "L")]
print(forward_chain(rules, ["A", "B"], "Q"))
print(forward_chain(rules, ["A"], "Q"))

from itertools import combinations


def negate(literal):
    return literal[1:] if literal.startswith("~") else "~" + literal


def resolve(first, second):
    return [(first - {lit}) | (second - {negate(lit)}) for lit in first if negate(lit) in second]


def refutes(clauses):
    clauses = set(clauses)
    while True:
        new = set()
        for first, second in combinations(clauses, 2):
            for resolvent in resolve(first, second):
                if not resolvent:
                    return True
                new.add(frozenset(resolvent))
        if new <= clauses:
            return False
        clauses |= new


kb = [frozenset({"~P", "Q"}), frozenset({"~Q", "R"}), frozenset({"P"})]
print(refutes(kb + [frozenset({"~R"})]))
print(refutes(kb + [frozenset({"~S"})]))

parents = {"tweety": "penguin", "penguin": "bird", "bird": "animal", "polly": "parrot", "parrot": "bird"}
properties = {
    "animal": {"alive": True},
    "bird": {"flies": True, "legs": 2},
    "penguin": {"flies": False},
}


def lookup(thing, attribute):
    while thing is not None:
        if attribute in properties.get(thing, {}):
            return properties[thing][attribute]
        thing = parents.get(thing)
    return None


print(lookup("polly", "flies"))
print(lookup("tweety", "flies"))
print(lookup("tweety", "legs"), lookup("tweety", "alive"))
print(lookup("tweety", "color"))

def is_variable(term):
    return isinstance(term, str) and term[:1].islower()


def occurs(var, term, theta):
    if var == term:
        return True
    if is_variable(term) and term in theta:
        return occurs(var, theta[term], theta)
    if isinstance(term, tuple):
        return any(occurs(var, item, theta) for item in term)
    return False


def unify_var(var, term, theta):
    if var in theta:
        return unify(theta[var], term, theta)
    if is_variable(term) and term in theta:
        return unify(var, theta[term], theta)
    if occurs(var, term, theta):
        return None
    return {**theta, var: term}


def unify(x, y, theta):
    if theta is None:
        return None
    if x == y:
        return theta
    if is_variable(x):
        return unify_var(x, y, theta)
    if is_variable(y):
        return unify_var(y, x, theta)
    if isinstance(x, tuple) and isinstance(y, tuple) and len(x) == len(y):
        for a, b in zip(x, y):
            theta = unify(a, b, theta)
        return theta
    return None


print(sorted(unify(("Knows", "John", "x"), ("Knows", "y", "Mary"), {}).items()))
print(unify(("P", "x"), ("P", ("F", "x")), {}))
print(unify(("Knows", "John"), ("Likes", "John"), {}))
print(sorted(unify(("P", "x", "x"), ("P", "y", "Z"), {}).items()))

def posterior(prior, sensitivity, false_positive):
    evidence = sensitivity * prior + false_positive * (1 - prior)
    return sensitivity * prior / evidence


print(round(posterior(0.01, 0.9, 0.09), 4), round(posterior(0.10, 0.9, 0.09), 4))

P_RAIN = 0.2
P_SPRINKLER = {True: 0.01, False: 0.4}
P_WET = {(True, True): 0.99, (True, False): 0.9, (False, True): 0.8, (False, False): 0.0}


def joint(rain, sprinkler, wet):
    p = P_RAIN if rain else 1 - P_RAIN
    p *= P_SPRINKLER[rain] if sprinkler else 1 - P_SPRINKLER[rain]
    p_wet = P_WET[(sprinkler, rain)]
    return p * (p_wet if wet else 1 - p_wet)


numerator = sum(joint(True, s, True) for s in (True, False))
evidence = sum(joint(r, s, True) for r in (True, False) for s in (True, False))
print(round(numerator / evidence, 4), round(evidence, 4))

def confusion(actual, predicted):
    tp = sum(a == 1 and p == 1 for a, p in zip(actual, predicted))
    fp = sum(a == 0 and p == 1 for a, p in zip(actual, predicted))
    fn = sum(a == 1 and p == 0 for a, p in zip(actual, predicted))
    tn = sum(a == 0 and p == 0 for a, p in zip(actual, predicted))
    return tp, fp, fn, tn


def metrics(actual, predicted):
    tp, fp, fn, tn = confusion(actual, predicted)
    accuracy = (tp + tn) / len(actual)
    precision = tp / (tp + fp) if tp + fp else 0.0
    recall = tp / (tp + fn) if tp + fn else 0.0
    f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
    return accuracy, precision, recall, f1


actual = [1, 1, 1, 1, 0, 0, 0, 0, 0, 0]
predicted = [1, 1, 0, 0, 0, 0, 0, 0, 1, 0]
tp, fp, fn, tn = confusion(actual, predicted)
print(f"tp={tp} fp={fp} fn={fn} tn={tn}")
print(*[round(value, 3) for value in metrics(actual, predicted)])

rare = [1] * 10 + [0] * 990
always_negative = [0] * 1000
accuracy, precision, recall, f1 = metrics(rare, always_negative)
print(accuracy, recall)

xs = [0, 1, 2, 3, 4]
ys = [1, 3, 5, 7, 9]
w = b = 0.0
n = len(xs)
for _ in range(2000):
    grad_w = sum(2 * (w * x + b - y) * x for x, y in zip(xs, ys)) / n
    grad_b = sum(2 * (w * x + b - y) for x, y in zip(xs, ys)) / n
    w -= 0.05 * grad_w
    b -= 0.05 * grad_b

mean_x = sum(xs) / n
mean_y = sum(ys) / n
closed_w = sum((x - mean_x) * (y - mean_y) for x, y in zip(xs, ys)) / sum((x - mean_x) ** 2 for x in xs)
closed_b = mean_y - closed_w * mean_x
print(round(w, 3), round(b, 3))
print(closed_w, closed_b)

import math

xs = [1, 2, 3, 4, 5, 6]
labels = [0, 0, 0, 1, 1, 1]


def sigmoid(z):
    return 1 / (1 + math.exp(-z))


w = b = 0.0
for _ in range(3000):
    grad_w = sum((sigmoid(w * x + b) - y) * x for x, y in zip(xs, labels)) / len(xs)
    grad_b = sum((sigmoid(w * x + b) - y) for x, y in zip(xs, labels)) / len(xs)
    w -= 0.5 * grad_w
    b -= 0.5 * grad_b

predictions = [1 if sigmoid(w * x + b) >= 0.5 else 0 for x in xs]
print(predictions == labels)
print(round(sigmoid(w * 2 + b), 3), round(sigmoid(w * 3.5 + b), 3), round(sigmoid(w * 5 + b), 3))
print(round(-b / w, 2))

train = [((1.0, 1000), "A"), ((1.1, 3000), "A"), ((5.0, 1100), "B"), ((5.2, 3100), "B")]


def make_scaler(points):
    lows = [min(p[i] for p in points) for i in range(2)]
    highs = [max(p[i] for p in points) for i in range(2)]
    return lambda p: tuple((p[i] - lows[i]) / (highs[i] - lows[i]) for i in range(2))


def knn(train, query, k, scaled):
    scale = make_scaler([point for point, _ in train]) if scaled else (lambda p: p)
    q = scale(query)
    ordered = sorted(train, key=lambda item: sum((a - b) ** 2 for a, b in zip(scale(item[0]), q)))
    votes = [label for _, label in ordered[:k]]
    return max(set(votes), key=votes.count)


print(knn(train, (1.2, 2000), 1, False))
print(knn(train, (1.2, 2000), 1, True))

import math

rows = [
    ("sunny", False, "no"),
    ("sunny", True, "no"),
    ("overcast", False, "yes"),
    ("rain", False, "yes"),
    ("rain", False, "yes"),
    ("rain", True, "no"),
    ("overcast", True, "yes"),
    ("sunny", False, "no"),
]


def entropy(labels):
    total = len(labels)
    return -sum((labels.count(v) / total) * math.log2(labels.count(v) / total) for v in set(labels))


def gain(rows, index):
    remainder = 0.0
    for value in set(row[index] for row in rows):
        subset = [row[-1] for row in rows if row[index] == value]
        remainder += len(subset) / len(rows) * entropy(subset)
    return entropy([row[-1] for row in rows]) - remainder


print(round(entropy([row[-1] for row in rows]), 3))
gains = {"outlook": gain(rows, 0), "windy": gain(rows, 1)}
print({name: round(value, 3) for name, value in gains.items()})
print(max(gains, key=gains.get))

points = [(1.0, 1.0), (1.5, 2.0), (3.0, 4.0), (5.0, 7.0), (3.5, 5.0), (4.5, 5.0), (3.5, 4.5)]


def distance2(a, b):
    return (a[0] - b[0]) ** 2 + (a[1] - b[1]) ** 2


def kmeans(points, centroids):
    iterations = 0
    while True:
        iterations += 1
        clusters = [[] for _ in centroids]
        for point in points:
            nearest = min(range(len(centroids)), key=lambda i: distance2(point, centroids[i]))
            clusters[nearest].append(point)
        updated = [(sum(p[0] for p in c) / len(c), sum(p[1] for p in c) / len(c)) for c in clusters]
        if updated == centroids:
            return centroids, clusters, iterations
        centroids = updated


centroids, clusters, iterations = kmeans(points, [points[0], points[3]])
print([tuple(round(value, 2) for value in c) for c in centroids])
print([len(c) for c in clusters], iterations)

xs = list(range(1, 13))
labels = [0, 0, 0, 0, 1, 0, 0, 1, 1, 1, 1, 1]


def predict(train_x, train_y, x, k):
    order = sorted(range(len(train_x)), key=lambda i: abs(train_x[i] - x))
    votes = [train_y[i] for i in order[:k]]
    return 1 if sum(votes) * 2 > len(votes) else 0


def train_accuracy(k):
    return sum(predict(xs, labels, x, k) == y for x, y in zip(xs, labels)) / len(xs)


def leave_one_out_accuracy(k):
    correct = 0
    for i in range(len(xs)):
        rest_x = xs[:i] + xs[i + 1:]
        rest_y = labels[:i] + labels[i + 1:]
        correct += predict(rest_x, rest_y, xs[i], k) == labels[i]
    return correct / len(xs)


for k in (1, 3, 5):
    print(k, round(train_accuracy(k), 3), round(leave_one_out_accuracy(k), 3))

def train_perceptron(data, epochs=20):
    w = [0, 0]
    b = 0
    for _ in range(epochs):
        mistakes = 0
        for (x1, x2), target in data:
            output = 1 if w[0] * x1 + w[1] * x2 + b > 0 else 0
            error = target - output
            if error:
                w[0] += error * x1
                w[1] += error * x2
                b += error
                mistakes += 1
        if mistakes == 0:
            return w, b, True
    return w, b, False


AND = [((0, 0), 0), ((0, 1), 0), ((1, 0), 0), ((1, 1), 1)]
XOR = [((0, 0), 0), ((0, 1), 1), ((1, 0), 1), ((1, 1), 0)]
print(train_perceptron(AND))
print(train_perceptron(XOR)[2])

import math


def sigmoid(z):
    return 1 / (1 + math.exp(-z))


def relu(z):
    return max(0.0, z)


def softmax(values):
    peak = max(values)
    exps = [math.exp(v - peak) for v in values]
    total = sum(exps)
    return [e / total for e in exps]


for z in (-2, 0, 2):
    print(z, round(sigmoid(z), 3), round(math.tanh(z), 3), relu(z))
probabilities = softmax([2.0, 1.0, 0.1])
print([round(p, 3) for p in probabilities], round(sum(probabilities), 6))
print([round(p, 3) for p in softmax([1000, 1001])])

def dense(inputs, weights, biases, activation):
    return [
        activation(sum(w * x for w, x in zip(row, inputs)) + bias)
        for row, bias in zip(weights, biases)
    ]


def step(z):
    return 1 if z > 0 else 0


hidden_weights = [[1, 1], [1, 1]]
hidden_biases = [-0.5, -1.5]
output_weights = [[1, -1]]
output_biases = [-0.5]

for x in ((0, 0), (0, 1), (1, 0), (1, 1)):
    hidden = dense(x, hidden_weights, hidden_biases, step)
    output = dense(hidden, output_weights, output_biases, step)[0]
    print(x, hidden, output)

import math

DATA = [((0, 0), 0), ((0, 1), 1), ((1, 0), 1), ((1, 1), 0)]
HIDDEN = 3
LEARNING_RATE = 1.0
w1 = [[0.5, -0.4], [-0.3, 0.8], [0.9, 0.2]]
b1 = [0.1, -0.1, 0.05]
w2 = [0.7, -0.6, 0.4]
b2 = [0.0]


def sigmoid(z):
    return 1 / (1 + math.exp(-z))


def forward(x):
    h = [sigmoid(sum(w1[j][k] * x[k] for k in range(2)) + b1[j]) for j in range(HIDDEN)]
    y = sigmoid(sum(w2[j] * h[j] for j in range(HIDDEN)) + b2[0])
    return h, y


def loss():
    return sum(0.5 * (forward(x)[1] - t) ** 2 for x, t in DATA)


def gradients():
    gw1 = [[0.0, 0.0] for _ in range(HIDDEN)]
    gb1 = [0.0] * HIDDEN
    gw2 = [0.0] * HIDDEN
    gb2 = 0.0
    for x, t in DATA:
        h, y = forward(x)
        dz2 = (y - t) * y * (1 - y)
        gb2 += dz2
        for j in range(HIDDEN):
            gw2[j] += dz2 * h[j]
            dz1 = dz2 * w2[j] * h[j] * (1 - h[j])
            gb1[j] += dz1
            for k in range(2):
                gw1[j][k] += dz1 * x[k]
    return gw1, gb1, gw2, gb2


analytic = gradients()[0][1][0]
eps = 1e-6
w1[1][0] += eps
up = loss()
w1[1][0] -= 2 * eps
down = loss()
w1[1][0] += eps
print(abs((up - down) / (2 * eps) - analytic) < 1e-8)

for _ in range(10000):
    gw1, gb1, gw2, gb2 = gradients()
    for j in range(HIDDEN):
        for k in range(2):
            w1[j][k] -= LEARNING_RATE * gw1[j][k]
        b1[j] -= LEARNING_RATE * gb1[j]
        w2[j] -= LEARNING_RATE * gw2[j]
    b2[0] -= LEARNING_RATE * gb2

print(round(loss(), 4))
print([round(forward(x)[1], 2) for x, _ in DATA])

import math


def softmax(values):
    peak = max(values)
    exps = [math.exp(v - peak) for v in values]
    total = sum(exps)
    return [e / total for e in exps]


def cross_entropy(scores, target):
    return -math.log(softmax(scores)[target])


def sigmoid(z):
    return 1 / (1 + math.exp(-z))


scores = [2.0, 1.0, 0.1]
target = 0
p = softmax(scores)
analytic = [p[i] - (1 if i == target else 0) for i in range(3)]
numeric = []
for i in range(3):
    up = scores[:]
    up[i] += 1e-6
    down = scores[:]
    down[i] -= 1e-6
    numeric.append((cross_entropy(up, target) - cross_entropy(down, target)) / 2e-6)
print(all(abs(a - n) < 1e-6 for a, n in zip(analytic, numeric)))
print(round(cross_entropy(scores, 0), 4), round(cross_entropy(scores, 2), 4))
y = sigmoid(-6)
print(round((y - 1) * y * (1 - y), 4), round(y - 1, 4))

def conv_out(n, k, stride=1, pad=0):
    return (n + 2 * pad - k) // stride + 1


size, channels = 28, 1
size = conv_out(size, 5)
conv1_params = (5 * 5 * channels + 1) * 6
channels = 6
print(size, channels, conv1_params)
size = conv_out(size, 2, stride=2)
print(size, channels)
size = conv_out(size, 5)
conv2_params = (5 * 5 * channels + 1) * 16
channels = 16
print(size, channels, conv2_params)
size = conv_out(size, 2, stride=2)
print(size, channels, size * size * channels)
print((28 * 28) * (24 * 24 * 6) + 24 * 24 * 6)
print(256 * 120 + 120)

from itertools import product

activations = [0.2, 0.8, 0.5, 1.0]
keep = 0.5
masks = list(product([0, 1], repeat=len(activations)))
mean = [sum(a * mask[i] / keep for mask in masks) / len(masks) for i, a in enumerate(activations)]
print([round(value, 10) for value in mean] == activations, len(masks))

xs = [1, 2, 3, 4]
ys = [2, 4, 6, 8]


def fit(penalty):
    w = 0.0
    n = len(xs)
    for _ in range(5000):
        grad = sum(2 * (w * x - y) * x for x, y in zip(xs, ys)) / n + 2 * penalty * w
        w -= 0.02 * grad
    return w


print(round(fit(0.0), 3), round(fit(1.0), 3))
closed = sum(x * y for x, y in zip(xs, ys)) / (sum(x * x for x in xs) + len(xs) * 1.0)
print(round(closed, 3))

image = [
    [0, 50, 100],
    [150, 200, 250],
]


def to_gray(pixel):
    red, green, blue = pixel
    return round(0.299 * red + 0.587 * green + 0.114 * blue)


def brighten(img, amount):
    return [[min(255, max(0, value + amount)) for value in row] for row in img]


def normalize(img):
    return [[value / 255 for value in row] for row in img]


print(len(image), len(image[0]))
print(to_gray((200, 100, 50)), to_gray((255, 255, 255)))
print(brighten(image, 100))
print([[round(value, 2) for value in row] for row in normalize(image)])
print(image[1][2])

def convolve(image, kernel):
    kh, kw = len(kernel), len(kernel[0])
    output = []
    for r in range(len(image) - kh + 1):
        row = []
        for c in range(len(image[0]) - kw + 1):
            row.append(sum(image[r + i][c + j] * kernel[i][j] for i in range(kh) for j in range(kw)))
        output.append(row)
    return output


image = [[0, 0, 90, 90, 90] for _ in range(5)]
box = [[1 / 9] * 3 for _ in range(3)]
identity = [[0, 0, 0], [0, 1, 0], [0, 0, 0]]
blurred = convolve(image, box)
print(len(blurred), len(blurred[0]))
print([round(value, 1) for value in blurred[0]])
print(convolve(image, identity)[0])

import math


def convolve(image, kernel):
    kh, kw = len(kernel), len(kernel[0])
    return [
        [
            sum(image[r + i][c + j] * kernel[i][j] for i in range(kh) for j in range(kw))
            for c in range(len(image[0]) - kw + 1)
        ]
        for r in range(len(image) - kh + 1)
    ]


SOBEL_X = [[-1, 0, 1], [-2, 0, 2], [-1, 0, 1]]
SOBEL_Y = [[-1, -2, -1], [0, 0, 0], [1, 2, 1]]

vertical_edge = [[0, 0, 100, 100, 100] for _ in range(5)]
gx = convolve(vertical_edge, SOBEL_X)
gy = convolve(vertical_edge, SOBEL_Y)
magnitude = [[round(math.hypot(a, b)) for a, b in zip(row_x, row_y)] for row_x, row_y in zip(gx, gy)]
print(magnitude[1])
print(round(math.degrees(math.atan2(gy[1][0], gx[1][0]))))

horizontal_edge = [[0] * 5, [0] * 5, [100] * 5, [100] * 5, [100] * 5]
print(convolve(horizontal_edge, SOBEL_Y)[0])

def otsu(pixels):
    histogram = [0] * 256
    for pixel in pixels:
        histogram[pixel] += 1
    total = len(pixels)
    total_sum = sum(level * count for level, count in enumerate(histogram))
    best_variance, best_threshold = -1.0, 0
    weight_dark = sum_dark = 0
    for t in range(256):
        weight_dark += histogram[t]
        if weight_dark == 0:
            continue
        weight_bright = total - weight_dark
        if weight_bright == 0:
            break
        sum_dark += t * histogram[t]
        mean_dark = sum_dark / weight_dark
        mean_bright = (total_sum - sum_dark) / weight_bright
        between = weight_dark * weight_bright * (mean_dark - mean_bright) ** 2
        if between > best_variance:
            best_variance, best_threshold = between, t
    return best_threshold


pixels = [28, 30, 35, 40, 33, 31, 38, 29, 205, 210, 215, 200, 220, 208, 212, 218]
threshold = otsu(pixels)
print(threshold)
print([1 if pixel > threshold else 0 for pixel in pixels])

from collections import deque

GRID = [
    "11000",
    "11001",
    "00001",
    "10100",
    "11100",
]


def label_components(grid):
    rows, cols = len(grid), len(grid[0])
    labels = [[0] * cols for _ in range(rows)]
    count = 0
    sizes = []
    for r in range(rows):
        for c in range(cols):
            if grid[r][c] == "1" and labels[r][c] == 0:
                count += 1
                size = 0
                queue = deque([(r, c)])
                labels[r][c] = count
                while queue:
                    y, x = queue.popleft()
                    size += 1
                    for ny, nx in ((y + 1, x), (y - 1, x), (y, x + 1), (y, x - 1)):
                        if 0 <= ny < rows and 0 <= nx < cols and grid[ny][nx] == "1" and labels[ny][nx] == 0:
                            labels[ny][nx] = count
                            queue.append((ny, nx))
                sizes.append(size)
    return count, sizes, labels


count, sizes, labels = label_components(GRID)
print(count, sizes)
for row in labels:
    print(*row)

image = [
    [10, 10, 10, 10, 10, 10],
    [10, 10, 10, 10, 10, 10],
    [10, 10, 10, 80, 90, 10],
    [10, 10, 10, 70, 200, 10],
    [10, 10, 10, 60, 50, 10],
    [10, 10, 10, 10, 10, 10],
]
template = [[80, 90], [70, 200], [60, 50]]


def ssd_map(image, template):
    th, tw = len(template), len(template[0])
    scores = {}
    for r in range(len(image) - th + 1):
        for c in range(len(image[0]) - tw + 1):
            scores[(r, c)] = sum((image[r + i][c + j] - template[i][j]) ** 2 for i in range(th) for j in range(tw))
    return scores


scores = ssd_map(image, template)
best = min(scores, key=scores.get)
print(best, scores[best])
print(sorted(scores.values())[1] > 0)
print(len(scores))