import unittest, ast, importlib.util, types
import torch
import torch.nn as nn

def _dynamic(tc, ok, ok_msg, fail_msg):
    tc._testMethodName = ok_msg if ok else fail_msg
    (tc.assertTrue if ok else tc.fail)(tc._testMethodName)

def _load_user_source():
    spec = importlib.util.find_spec("user_code")
    if spec is None or not spec.origin:
        raise FileNotFoundError("Cannot locate user_code module file")
    with open(spec.origin, "r", encoding="utf-8") as f:
        return f.read()

def _extract_class_from_source(src, class_name="TimeSeriesPredictor"):
    tree = ast.parse(src)
    for node in tree.body:
        if isinstance(node, ast.ClassDef) and node.name == class_name:
            return node
    return None

def _compile_class(class_node):
    module = types.ModuleType("isolated_user_model")
    module.torch = torch
    module.nn = nn
    mod_ast = ast.Module(body=[class_node], type_ignores=[])
    ast.fix_missing_locations(mod_ast)
    code = compile(mod_ast, filename="<isolated>", mode="exec")
    exec(code, module.__dict__)
    return module

# ------- AST helpers for robust detection -------
def _attr_chain(node):
    """Return dotted name for Attribute/Name, e.g. optim.Adam / nn.MSELoss."""
    parts = []
    while isinstance(node, ast.Attribute):
        parts.append(node.attr)
        node = node.value
    if isinstance(node, ast.Name):
        parts.append(node.id)
    return ".".join(reversed(parts)) if parts else None

def _find_assign_calls(src, target_names):
    """Find assignments like target = CALL(...) and return {name: callee_str}."""
    tree = ast.parse(src)
    found = {}
    for node in ast.walk(tree):
        if isinstance(node, ast.Assign) and isinstance(node.value, ast.Call):
            callee = node.value.func
            callee_name = _attr_chain(callee)
            for tgt in node.targets:
                if isinstance(tgt, ast.Name) and tgt.id in target_names:
                    found[tgt.id] = callee_name
    return found

class TestTimeSeriesPredictor_NoImport(unittest.TestCase):

    def test_architecture_and_forward(self):
        reasons = []
        try:
            src = _load_user_source()
        except Exception as e:
            _dynamic(self, False, "", f"Cannot read user_code: {e}")
            return

        class_node = _extract_class_from_source(src, "TimeSeriesPredictor")
        if class_node is None:
            _dynamic(self, False, "", "Class TimeSeriesPredictor is not defined")
            return

        try:
            isolated = _compile_class(class_node)
            Model = getattr(isolated, "TimeSeriesPredictor")
            model = Model(1, 64, 1, 1)

            lstm = getattr(model, "lstm", None)
            fc   = getattr(model, "fc", None)
            if not isinstance(lstm, nn.LSTM):
                reasons.append("lstm layer must be nn.LSTM")
            else:
                if not getattr(lstm, "batch_first", False):
                    reasons.append("lstm must use batch_first=True")
                if lstm.input_size != 1: reasons.append(f"lstm.input_size {lstm.input_size} â  1")
                if lstm.hidden_size != 64: reasons.append(f"lstm.hidden_size {lstm.hidden_size} â  64")
                if lstm.num_layers != 1: reasons.append(f"lstm.num_layers {lstm.num_layers} â  1")

            if not (isinstance(fc, nn.Linear) and fc.in_features == 64 and fc.out_features == 1):
                reasons.append("fc must be Linear(hidden_size â output_size)")

            with torch.no_grad():
                X = torch.randn(8, 10, 1)
                pred = model(X)
            if not (isinstance(pred, torch.Tensor) and pred.shape == (8, 1)):
                reasons.append(f"forward output shape {tuple(pred.shape)} â  (8, 1)")
        except Exception as e:
            reasons.append(f"forward failed: {e}")

        _dynamic(self,
                 len(reasons) == 0,
                 "PASS: architecture & forward are correct",
                 "FAIL: " + "; ".join(reasons))

    def test_criterion_and_optimizer_ast(self):
        # AST-based: accepts nn.MSELoss / torch.nn.MSELoss; optim.Adam / torch.optim.Adam
        try:
            src = _load_user_source()
        except Exception as e:
            _dynamic(self, False, "", f"Cannot read user_code: {e}")
            return

        assigns = _find_assign_calls(src, {"criterion", "optimizer"})
        reasons = []

        crit = assigns.get("criterion")
        opt  = assigns.get("optimizer")

        valid_crit = crit in {"nn.MSELoss", "torch.nn.MSELoss"}
        valid_opt  = opt in {"optim.Adam", "torch.optim.Adam"}

        if not valid_crit:
            reasons.append(f"criterion must be MSELoss; found {crit!r}")
        if not valid_opt:
            reasons.append(f"optimizer must be Adam; found {opt!r}")

        _dynamic(self,
                 len(reasons) == 0,
                 "PASS: criterion & optimizer are declared (AST)",
                 "FAIL: " + "; ".join(reasons))


test_code.py

Keskittyy yksinomaan rekursiivisiin neuroverkkoihin (RNN) ja niiden soveltamiseen sekvenssien generointiin. Tämä edistynyt moduuli kattaa RNN:ien ydinteorian sekä käytännön lähestymistavat sekventiaalisen datan mallintamiseen, valmistaen sinut työskentelemään LSTM- ja GRU-verkkojen kanssa Pythonissa.

Haaste: Aikasarjojen Ennustaminen LSTM:llä

Ratkaisu