Update.
[mygptrnn.git] / rpl.py
1 #!/usr/bin/env python
2
3 # Any copyright is dedicated to the Public Domain.
4 # https://creativecommons.org/publicdomain/zero/1.0/
5
6 # Written by Francois Fleuret <francois@fleuret.org>
7
8 import math
9
10 import torch, torchvision
11
12 from torch import nn
13 from torch.nn import functional as F
14
15 ######################################################################
16
17
18 def rpl_exec(program, stack):
19     stack = stack.copy()
20     for op in program:
21         if op == "add":
22             if len(stack) > 1:
23                 a, b = stack.pop(), stack.pop()
24                 stack.append(a + b)
25         elif op == "min":
26             if len(stack) > 1:
27                 a, b = stack.pop(), stack.pop()
28                 stack.append(min(a, b))
29         elif op == "max":
30             if len(stack) > 1:
31                 a, b = stack.pop(), stack.pop()
32                 stack.append(max(a, b))
33         elif op == "swp":
34             if len(stack) > 1:
35                 a, b = stack.pop(), stack.pop()
36                 stack.append(a)
37                 stack.append(b)
38         elif op == "rep":
39             if len(stack) > 1:
40                 a, b = stack.pop(), stack.pop()
41                 stack += [b] * a
42         elif op == "dup":
43             if len(stack) > 0:
44                 a = stack.pop()
45                 stack.append(a)
46                 stack.append(a)
47         elif op == "del":
48             if len(stack) > 0:
49                 a = stack.pop()
50         else:
51             raise ValueError(f"Unknown instruction {op}")
52
53     return stack
54
55
56 rpl_ops = ["add", "min", "max", "swp", "rep", "dup", "del"]
57
58 ######################################################################
59
60
61 def generate(
62     nb_starting_values=3, nb_result_values_max=None, max_input=9, prog_len=6, nb_runs=5
63 ):
64     prog_len = (1 + torch.randint(2 * prog_len, (1,))).clamp(max=prog_len).item()
65
66     while True:
67         no_empty_stack = True
68         prog = [rpl_ops[k] for k in torch.randint(len(rpl_ops), (prog_len,))]
69
70         result = []
71         for _ in range(nb_runs):
72             stack = [
73                 x.item() for x in torch.randint(max_input + 1, (nb_starting_values,))
74             ]
75             result_stack = rpl_exec(prog, stack)
76             if len(result_stack) == 0:
77                 no_empty_stack = False
78             result = result + ["<in>"] + stack + ["<out>"] + result_stack
79
80         result = result + ["<prg>"] + prog
81         result = result + ["<end>"]
82
83         if no_empty_stack and (
84             nb_result_values_max is None or len(result_stack) <= nb_result_values_max
85         ):
86             break
87
88     return result
89
90
91 def next_marker(seq, tokens, start=0):
92     pos = None
93     for t in tokens:
94         try:
95             i = seq.index(t, start)
96             if pos is None or i < pos:
97                 pos = i
98         except ValueError:
99             pass
100     return pos
101
102
103 def decompose(seq):
104     io = []
105     k = 0
106     while seq[k] == "<in>":
107         o = next_marker(seq, ["<out>"], start=k + 1)
108         if o is None:
109             raise ValueError("Missing output markers (should be correct in the prompt)")
110         e = next_marker(seq, ["<in>", "<prg>"], start=o)
111         if e is None:
112             raise ValueError(
113                 "Missing input/output markers (should be correct in the prompt)"
114             )
115         try:
116             io.append(
117                 ([int(x) for x in seq[k + 1 : o]], [int(x) for x in seq[o + 1 : e]])
118             )
119         except ValueError:
120             raise ValueError(
121                 "Invalid input/output value (should be correct in the prompt)"
122             )
123
124         k = e
125
126     if seq[k] == "<prg>":
127         e = next_marker(seq, ["<end>"], start=k)
128         if e is None:
129             prog = []
130         else:
131             prog = seq[k + 1 : e]
132     else:
133         raise ValueError("Missing <prg> (it should be in the prompt)")
134
135     return prog, io
136
137
138 def stack_distance(target_stack, result_stack):
139     return abs(len(result_stack) - len(target_stack)) + sum(
140         [0 if x == y else 1 for x, y in zip(result_stack, target_stack)]
141     )
142
143
144 def compute_nb_errors(seq):
145     prog, io = decompose(seq)
146
147     nb_total, nb_errors = 0, 0
148
149     stacks = []
150
151     if len(set(prog) - set(rpl_ops)) > 0:
152         # Program is not valid, we count 100% error
153         for start_stack, target_stack in io:
154             stacks.append((start_stack, target_stack, ["N/A"], False))
155             nb_total += len(target_stack)
156             nb_errors += len(target_stack)
157
158     else:
159         # Program is valid
160         for start_stack, target_stack in io:
161             result_stack = rpl_exec(prog, start_stack)
162             nb_total += len(target_stack)
163             e = stack_distance(target_stack, result_stack)
164             nb_errors += e
165             stacks.append((start_stack, target_stack, result_stack, e == 0))
166
167     return nb_total, nb_errors, prog, stacks
168
169
170 ######################################################################
171
172 if __name__ == "__main__":
173     seq = generate()
174     print(seq)
175     seq[3] = 7
176     print(seq)
177     print(compute_nb_errors(seq))