Developer Track #1 (Foundations): Python Foundations for Industrial AI

from typing import Dict, Any, Optional

# Minimal example state so the function has a defined input source.
plc_state: Dict[str, Dict[str, Any]] = {
    "Line1": {"motor_running": True, "pallets_in_buffer": 3},
}

def naive_get_fault(line_id: str) -> Optional[str]:
    """
    Attempt to read fault_code from a global PLC state dictionary.
    WARNING: This assumes fault_code always exists in the snapshot.
    """
    state = plc_state.get(line_id, {})
    # Direct indexing: raises KeyError if fault_code is missing
    try:
        return state["fault_code"]
    except KeyError as exc:
        print(f"[RUNTIME ERROR] Missing key for {line_id}: {exc}")
        return None

from typing import Dict, Any, Optional

# Simulated PLC state from two lines (notice Line1 has no fault_code)
plc_state: Dict[str, Dict[str, Any]] = {
    "Line1": {"motor_running": True, "pallets_in_buffer": 3},
    "Line2": {"motor_running": False, "pallets_in_buffer": 0, "fault_code": "E1001"},
}

def naive_get_fault(line_id: str) -> Optional[str]:
    state = plc_state.get(line_id, {})
    try:
        return state["fault_code"]  # KeyError if missing
    except KeyError as exc:
        print(f"[RUNTIME ERROR] Missing key for {line_id}: {exc}")
        return None

# Test with both lines
for line in ["Line1", "Line2"]:
    print(line, "fault:", naive_get_fault(line))

from typing import TypedDict, Optional

class PlcSnapshot(TypedDict, total=False):
    """
    Schema for PLC line snapshots.
    total=False means all fields are optional (can be missing).
    """
    line_id: str
    motor_running: bool
    pallets_in_buffer: int
    fault_code: Optional[str]  # Can be None or missing entirely

from dataclasses import dataclass
from typing import Optional, TypedDict

class PlcSnapshot(TypedDict, total=False):
    line_id: str
    motor_running: bool
    pallets_in_buffer: int
    fault_code: Optional[str]

# Minimal typed state so the agent has a defined input source.
plc_typed_state: dict[str, PlcSnapshot] = {
    "Line1": {"line_id": "Line1", "motor_running": True, "pallets_in_buffer": 3},
}

@dataclass
class DiagnosticAgent:
    """
    Agent that reads fault codes from PLC snapshots.
    Uses .get() for safe field access instead of direct indexing.
    """
    name: str

    def get_fault(self, line_id: str) -> Optional[str]:
        snapshot = plc_typed_state.get(line_id)
        if snapshot is None:
            print(f"[WARN] No snapshot for {line_id}")
            return None
        # Safe access: returns None if fault_code is missing
        return snapshot.get("fault_code")

from dataclasses import dataclass
from typing import TypedDict, Optional

class PlcSnapshot(TypedDict, total=False):
    line_id: str
    motor_running: bool
    pallets_in_buffer: int
    fault_code: Optional[str]

# Simulated PLC state using the typed schema
plc_typed_state: dict[str, PlcSnapshot] = {
    "Line1": {"line_id": "Line1", "motor_running": True, "pallets_in_buffer": 3},
    "Line2": {
        "line_id": "Line2",
        "motor_running": False,
        "pallets_in_buffer": 0,
        "fault_code": "E1001",
    },
}

@dataclass
class DiagnosticAgent:
    name: str

    def get_fault(self, line_id: str) -> Optional[str]:
        snapshot = plc_typed_state.get(line_id)
        if snapshot is None:
            print(f"[WARN] No snapshot for {line_id}")
            return None
        return snapshot.get("fault_code")

# Create and test the agent
agent = DiagnosticAgent(name="D1-baseline")

for line in ["Line1", "Line2", "Line3"]:
    print(line, "fault:", agent.get_fault(line))

import time
import random
from typing import Dict

def read_plc_line(line_id: str) -> Dict[str, object]:
    """
    Simulate a blocking PLC read (~300ms).
    While time.sleep() runs, nothing else can execute.
    """
    time.sleep(0.3)  # Blocks for 300ms
    return {
        "line_id": line_id,
        "motor_running": random.choice([True, False]),
        "pallets_in_buffer": random.randint(0, 5),
        "fault_code": random.choice([None, "E1001", "E2002"]),
    }

from typing import List

def poll_all_lines_sequential(line_ids: List[str]) -> List[Dict[str, object]]:
    """
    Poll all lines sequentially (one at a time).
    Total time = number of lines × 0.3s per line.
    """
    start = time.perf_counter()

    # List comprehension: call read_plc_line for each line_id
    snapshots = [read_plc_line(line_id) for line_id in line_ids]

    duration = time.perf_counter() - start
    print(f"[SEQUENTIAL] Read {len(line_ids)} lines in {duration:.2f}s")

    # Extract all lines with any fault_code (including "E1001", "E2002", etc.)
    fault_lines = [s["line_id"] for s in snapshots if s["fault_code"]]
    print("[SEQUENTIAL] Faulty lines:", fault_lines)

    return snapshots

import time
import random
from typing import Dict, List

def read_plc_line(line_id: str) -> Dict[str, object]:
    """Simulate a blocking PLC read (~300ms)."""
    time.sleep(0.3)
    return {
        "line_id": line_id,
        "motor_running": random.choice([True, False]),
        "pallets_in_buffer": random.randint(0, 5),
        "fault_code": random.choice([None, "E1001", "E2002"]),
    }

def poll_all_lines_sequential(line_ids: List[str]) -> List[Dict[str, object]]:
    start = time.perf_counter()
    snapshots = [read_plc_line(line_id) for line_id in line_ids]
    duration = time.perf_counter() - start
    print(f"[SEQUENTIAL] Read {len(line_ids)} lines in {duration:.2f}s")

    fault_lines = [s["line_id"] for s in snapshots if s["fault_code"]]
    print("[SEQUENTIAL] Faulty lines:", fault_lines)
    return snapshots

# Poll 4 lines sequentially
poll_all_lines_sequential(["Line1", "Line2", "Line3", "Line4"])

import asyncio
import logging
import random
from typing import Any, Dict

# Configure logging for production-like observability
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

def plc_io_delay() -> float:
    """
    Simulate realistic PLC response time with jitter (150–350ms).
    Industrial networks have variable latency depending on load and switches.
    """
    return random.uniform(0.15, 0.35)

import asyncio
import random
from typing import Any, Dict

def plc_io_delay() -> float:
    return random.uniform(0.15, 0.35)

async def async_read_plc_line_raw(line_id: str) -> Dict[str, Any]:
    """
    Async PLC read that can fail with ~10% probability.
    Uses await asyncio.sleep() instead of time.sleep(), allowing other tasks to run.
    """
    # Non-blocking sleep: other tasks can run while this awaits
    await asyncio.sleep(plc_io_delay())

    # Simulate occasional transient failure (network timeout, switch reset, etc.)
    if random.random() < 0.1:
        raise ConnectionError(f"Timeout talking to {line_id}")

    return {
        "line_id": line_id,
        "motor_running": random.choice([True, False]),
        "pallets_in_buffer": random.randint(0, 5),
        "fault_code": random.choice([None, "E1001", "E2002"]),
    }

import logging
from typing import Any, Dict

def log_and_swallow_errors(fn):
    """
    Decorator that wraps async_read_plc_line_raw to add:
    1. Logging (INFO for success, ERROR for failure)
    2. Fallback snapshots when exceptions occur
    """
    async def wrapper(line_id: str) -> Dict[str, Any]:
        try:
            logging.info("Reading %s", line_id)
            snapshot = await fn(line_id)  # Call the original function
            logging.info("OK: %s", line_id)
            return snapshot
        except Exception as exc:
            logging.error("PLC read failed for %s: %s", line_id, exc)
            # Return a safe fallback instead of crashing
            return {
                "line_id": line_id,
                "motor_running": False,
                "pallets_in_buffer": 0,
                "fault_code": "E-CONNECTION",  # Explicit error code
            }
    return wrapper

# Continuing from previous cells (imports, plc_io_delay, async_read_plc_line_raw, log_and_swallow_errors)

# Apply the decorator to create a production-safe reader
async_read_plc_line = log_and_swallow_errors(async_read_plc_line_raw)

# Now async_read_plc_line has:
# - All the PLC read logic from async_read_plc_line_raw
# - Automatic logging for every read attempt
# - Safe fallback snapshots when network errors occur
# - No changes needed in the calling code

import asyncio
import logging
import time
from typing import Any, Dict, List

async def async_read_plc_line(line_id: str) -> Dict[str, Any]:
    return {
        "line_id": line_id,
        "motor_running": True,
        "pallets_in_buffer": 2,
        "fault_code": None,
    }

async def poll_all_lines_concurrent(line_ids: List[str]) -> List[Dict[str, Any]]:
    """
    Poll all lines concurrently using asyncio.gather.
    Total time ≈ max(individual read times), not the sum.
    """
    start = time.perf_counter()

    # asyncio.gather starts all reads simultaneously and waits for all to complete
    # The * unpacks the generator expression into individual arguments
    snapshots = await asyncio.gather(*(async_read_plc_line(line_id) for line_id in line_ids))

    duration = time.perf_counter() - start
    logging.info("[ASYNC] Read %d lines in %.2fs", len(line_ids), duration)

    # List comprehension to extract faulted lines
    fault_lines = [s["line_id"] for s in snapshots if s["fault_code"]]
    logging.info("[ASYNC] Faulty lines: %s", fault_lines)

    return snapshots

import asyncio
import logging
import random
import time
from typing import Any, Dict, List

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

def plc_io_delay() -> float:
    return random.uniform(0.15, 0.35)

async def async_read_plc_line_raw(line_id: str) -> Dict[str, Any]:
    await asyncio.sleep(plc_io_delay())
    if random.random() < 0.1:
        raise ConnectionError(f"Timeout talking to {line_id}")
    return {
        "line_id": line_id,
        "motor_running": random.choice([True, False]),
        "pallets_in_buffer": random.randint(0, 5),
        "fault_code": random.choice([None, "E1001", "E2002"]),
    }

def log_and_swallow_errors(fn):
    async def wrapper(line_id: str) -> Dict[str, Any]:
        try:
            logging.info("Reading %s", line_id)
            snapshot = await fn(line_id)
            logging.info("OK: %s", line_id)
            return snapshot
        except Exception as exc:
            logging.error("PLC read failed for %s: %s", line_id, exc)
            return {
                "line_id": line_id,
                "motor_running": False,
                "pallets_in_buffer": 0,
                "fault_code": "E-CONNECTION",
            }
    return wrapper

async_read_plc_line = log_and_swallow_errors(async_read_plc_line_raw)

async def poll_all_lines_concurrent(line_ids: List[str]) -> List[Dict[str, Any]]:
    start = time.perf_counter()
    snapshots = await asyncio.gather(*(async_read_plc_line(line_id) for line_id in line_ids))
    duration = time.perf_counter() - start
    logging.info("[ASYNC] Read %d lines in %.2fs", len(line_ids), duration)
    fault_lines = [s["line_id"] for s in snapshots if s["fault_code"]]
    logging.info("[ASYNC] Faulty lines: %s", fault_lines)
    return snapshots

# Example usage (in Jupyter or script with asyncio.run):
# asyncio.run(poll_all_lines_concurrent(["Line1", "Line2", "Line3", "Line4"]))