Advent of code 2019 day 7

day7-data.txt

test1:3,15,3,16,1002,16,10,16,1,16,15,15,4,15,99,0,0
test2:3,23,3,24,1002,24,10,24,1002,23,-1,23,101,5,23,23,1,24,23,23,4,23,99,0,0
test3:3,31,3,32,1002,32,10,32,1001,31,-2,31,1007,31,0,33,1002,33,7,33,1,33,31,31,1,32,31,31,4,31,99,0,0,0
test1b:3,26,1001,26,-4,26,3,27,1002,27,2,27,1,27,26,27,4,27,1001,28,-1,28,1005,28,6,99,0,0,5
test2b:3,52,1001,52,-5,52,3,53,1,52,56,54,1007,54,5,55,1005,55,26,1001,54,-5,54,1105,1,12,1,53,54,53,1008,54,0,55,1001,55,1,55,2,53,55,53,4,53,1001,56,-1,56,1005,56,6,99,0,0,0,0,10
data:3,8,1001,8,10,8,105,1,0,0,21,38,55,72,93,118,199,280,361,442,99999,3,9,1001,9,2,9,1002,9,5,9,101,4,9,9,4,9,99,3,9,1002,9,3,9,1001,9,5,9,1002,9,4,9,4,9,99,3,9,101,4,9,9,1002,9,3,9,1001,9,4,9,4,9,99,3,9,1002,9,4,9,1001,9,4,9,102,5,9,9,1001,9,4,9,4,9,99,3,9,101,3,9,9,1002,9,3,9,1001,9,3,9,102,5,9,9,101,4,9,9,4,9,99,3,9,101,1,9,9,4,9,3,9,1001,9,1,9,4,9,3,9,102,2,9,9,4,9,3,9,101,2,9,9,4,9,3,9,1001,9,1,9,4,9,3,9,102,2,9,9,4,9,3,9,1001,9,1,9,4,9,3,9,102,2,9,9,4,9,3,9,102,2,9,9,4,9,3,9,1002,9,2,9,4,9,99,3,9,1001,9,1,9,4,9,3,9,1002,9,2,9,4,9,3,9,1001,9,2,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,2,9,9,4,9,3,9,102,2,9,9,4,9,3,9,102,2,9,9,4,9,3,9,102,2,9,9,4,9,3,9,101,1,9,9,4,9,3,9,101,1,9,9,4,9,99,3,9,101,2,9,9,4,9,3,9,101,1,9,9,4,9,3,9,101,1,9,9,4,9,3,9,102,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,1001,9,2,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,1,9,9,4,9,99,3,9,1001,9,1,9,4,9,3,9,1002,9,2,9,4,9,3,9,1001,9,1,9,4,9,3,9,1001,9,2,9,4,9,3,9,102,2,9,9,4,9,3,9,1001,9,1,9,4,9,3,9,1002,9,2,9,4,9,3,9,1001,9,2,9,4,9,3,9,1001,9,2,9,4,9,3,9,102,2,9,9,4,9,99,3,9,101,1,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,1,9,9,4,9,3,9,101,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,1,9,9,4,9,99

day7a.py

#!python
"""
https://adventofcode.com/2019/day/7


"""

import copy
import operator
import itertools


def compute(instructions, inputs):
    """
    inputs is a list of numeric inputs

    returns: the final state of the "program" and the last output
    """

    def get_input(list, pos, mode):
        n = list[pos]
        if mode == 0:
            return list[n]
        elif mode == 1:
            return n
        else:
            raise Exception("Invalid mode: {}".format(mode))

    my_program = copy.deepcopy(instructions)

    ip = 0   # instruction pointer
    max_ip = len(my_program)
    result = None

    while ip < max_ip:
        output = None

        # modes: 0 = position, 1 = immediate
        opcode = my_program[ip] % 100
        mode1 = my_program[ip] // 100 % 10
        mode2 = my_program[ip] // 1000 % 10
        mode3 = my_program[ip] // 10000 % 10

        if (opcode == 1) or (opcode == 2):
            # add or multiply
            input1 = get_input(my_program, ip + 1,  mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            assert mode3 == 0, "output mode must be zero"
            out_pos = my_program[ip + 3]
            functor = operator.add if opcode == 1 else operator.mul
            output = functor(input1, input2)
            my_program[out_pos] = output
            ip += 4
        elif opcode == 3:
            # input
            input1 = inputs.pop(0)  # consume an input
            assert mode1 == 0, "output mode must be zero"
            out_pos = my_program[ip + 1]
            my_program[out_pos] = input1
            ip += 2
        elif opcode == 4:
            # output
            input1 = get_input(my_program, ip + 1, mode1)
            result = input1
            # print("Output: {}".format(result))
            ip += 2
        elif (opcode == 5) or (opcode == 6):
            # jump if true, jump if false
            input1 = get_input(my_program, ip + 1, mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            if (opcode == 5) and (input1 != 0):
                ip = input2
            elif (opcode == 6) and (input1 == 0):
                ip = input2
            else:
                ip += 3
        elif (opcode == 7) or (opcode == 8):
            # less-then or equal
            input1 = get_input(my_program, ip + 1,  mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            comparator = operator.lt if opcode == 7 else operator.eq
            assert mode3 == 0, "output mode must be zero"
            # int(True) == 1, int(False) == 0
            out_pos = my_program[ip + 3]
            my_program[out_pos] = int(comparator(input1, input2))
            ip += 4
        elif opcode == 99:
            # halt
            # print("Done")
            ip = max_ip     # the end
        else:
            raise Exception("Invalid opcode: {}".format(opcode))

    return my_program, result


def read_data(filename):
    """
    Each line is a "program":
    name1:op1,op2,...opX
    """

    alldata = {}

    with open(filename) as f:
        for line in f:
            (name, prog) = line.rstrip().split(':')
            alldata[name] = [int(x) for x in prog.split(',')]

    return alldata


def run_a_program(plist):
    inputs = [0, 1, 2, 3, 4]
    results = []
    for inputset in itertools.permutations(inputs):
        val = 0
        for phase in list(inputset):
            (_, val) = compute(plist, [phase, val])
        results.append((inputset, val))

    def grabber(x):
        return x[1]

    return max(results, key=grabber)


def main():
    data = read_data('day7-data.txt')

    for pname in ['test1', 'test2', 'test3', 'data']:
        program = data[pname]
        (maxparm, maxvalue) = run_a_program(program)
        print("{}: {} <- {}".format(pname, maxparm, maxvalue))


if __name__ == "__main__":
    main()

day7a2.py

#!python
"""
https://adventofcode.com/2019/day/7

This is a rewrite of day7a.py using coroutines via asyncio,
getting ready for the second puzzle in day 7.

"""

import copy
import operator
import itertools
import asyncio


async def compute(instructions, inputs, outputs):
    """
    inputs: async input queue
    outputs: async output queue

    returns: the final state of the "program" and the last output
    """

    def get_input(list, pos, mode):
        n = list[pos]
        if mode == 0:
            return list[n]
        elif mode == 1:
            return n
        else:
            raise Exception("Invalid mode: {}".format(mode))

    my_program = copy.deepcopy(instructions)

    ip = 0   # instruction pointer
    max_ip = len(my_program)
    result = None

    while ip < max_ip:
        output = None

        # modes: 0 = position, 1 = immediate
        opcode = my_program[ip] % 100
        mode1 = my_program[ip] // 100 % 10
        mode2 = my_program[ip] // 1000 % 10
        mode3 = my_program[ip] // 10000 % 10

        if (opcode == 1) or (opcode == 2):
            # add or multiply
            input1 = get_input(my_program, ip + 1,  mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            assert mode3 == 0, "output mode must be zero"
            out_pos = my_program[ip + 3]
            functor = operator.add if opcode == 1 else operator.mul
            output = functor(input1, input2)
            my_program[out_pos] = output
            ip += 4
        elif opcode == 3:
            # input
            assert mode1 == 0, "output mode must be zero"
            input1 = await inputs.get()  # consume an input
            inputs.task_done()
            out_pos = my_program[ip + 1]
            my_program[out_pos] = input1
            ip += 2
        elif opcode == 4:
            # output
            input1 = get_input(my_program, ip + 1, mode1)
            result = input1
            outputs.put_nowait(result)
            # print("Output: {}".format(result))
            ip += 2
        elif (opcode == 5) or (opcode == 6):
            # jump if true, jump if false
            input1 = get_input(my_program, ip + 1, mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            if (opcode == 5) and (input1 != 0):
                ip = input2
            elif (opcode == 6) and (input1 == 0):
                ip = input2
            else:
                ip += 3
        elif (opcode == 7) or (opcode == 8):
            # less-then or equal
            input1 = get_input(my_program, ip + 1,  mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            comparator = operator.lt if opcode == 7 else operator.eq
            assert mode3 == 0, "output mode must be zero"
            # int(True) == 1, int(False) == 0
            out_pos = my_program[ip + 3]
            my_program[out_pos] = int(comparator(input1, input2))
            ip += 4
        elif opcode == 99:
            # halt
            # print("Done")
            ip = max_ip     # the end
        else:
            raise Exception("Invalid opcode: {}".format(opcode))

    return my_program, result


async def run_a_program(plist):
    inputs = [0, 1, 2, 3, 4]
    results = []
    for inputset in itertools.permutations(inputs):
        # create and initialize the queues
        # Q1 - F1 - Q2 - F2 - Q3 - F3 - Q4 - F5 - Q5
        queues = []
        for q in range(len(inputs) + 1):
            queues.append(asyncio.Queue())
        for pos in range(len(inputs)):
            # "phases"
            queues[pos].put_nowait(inputset[pos])
        # first value
        queues[0].put_nowait(0)
        # set up the computationsY
        filters = []
        for pos in range(len(inputs)):
            filter = asyncio.create_task(
                compute(plist, queues[pos], queues[pos+1]))
            filters.append(filter)
        await asyncio.gather(*filters, return_exceptions=True)
        val = queues[-1].get_nowait()  # output of final filter
        results.append((inputset, val))

    def grabber(x):
        return x[1]

    return max(results, key=grabber)


def read_data(filename):
    """py
    Each line is a "program":
    name1:op1,op2,...opX
    """

    alldata = {}

    with open(filename) as f:
        for line in f:
            (name, prog) = line.rstrip().split(':')
            alldata[name] = [int(x) for x in prog.split(',')]

    return alldata


def main():
    data = read_data('day7-data.txt')

    for pname in ['test1', 'test2', 'test3', 'data']:
        program = data[pname]
        (maxparm, maxvalue) = asyncio.run(run_a_program(program))
        print("{}: {} <- {}".format(pname, maxparm, maxvalue))


if __name__ == "__main__":
    main()

day7b.py

#!python
"""
https://adventofcode.com/2019/day/7#part2


"""

import copy
import operator
import itertools
import asyncio


async def compute(instructions, inputs, outputs):
    """
    inputs: async input queue
    outputs: async output queue

    returns: the final state of the "program" and the last output
    """

    def get_input(list, pos, mode):
        n = list[pos]
        if mode == 0:
            return list[n]
        elif mode == 1:
            return n
        else:
            raise Exception("Invalid mode: {}".format(mode))

    my_program = copy.deepcopy(instructions)

    ip = 0   # instruction pointer
    max_ip = len(my_program)
    result = None

    while ip < max_ip:
        output = None

        # modes: 0 = position, 1 = immediate
        opcode = my_program[ip] % 100
        mode1 = my_program[ip] // 100 % 10
        mode2 = my_program[ip] // 1000 % 10
        mode3 = my_program[ip] // 10000 % 10

        if (opcode == 1) or (opcode == 2):
            # add or multiply
            input1 = get_input(my_program, ip + 1,  mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            assert mode3 == 0, "output mode must be zero"
            out_pos = my_program[ip + 3]
            functor = operator.add if opcode == 1 else operator.mul
            output = functor(input1, input2)
            my_program[out_pos] = output
            ip += 4
        elif opcode == 3:
            # input
            assert mode1 == 0, "output mode must be zero"
            input1 = await inputs.get()  # consume an input
            inputs.task_done()
            out_pos = my_program[ip + 1]
            my_program[out_pos] = input1
            ip += 2
        elif opcode == 4:
            # output
            input1 = get_input(my_program, ip + 1, mode1)
            result = input1
            outputs.put_nowait(result)
            # print("Output: {}".format(result))
            ip += 2
        elif (opcode == 5) or (opcode == 6):
            # jump if true, jump if false
            input1 = get_input(my_program, ip + 1, mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            if (opcode == 5) and (input1 != 0):
                ip = input2
            elif (opcode == 6) and (input1 == 0):
                ip = input2
            else:
                ip += 3
        elif (opcode == 7) or (opcode == 8):
            # less-then or equal
            input1 = get_input(my_program, ip + 1,  mode1)
            input2 = get_input(my_program, ip + 2, mode2)
            comparator = operator.lt if opcode == 7 else operator.eq
            assert mode3 == 0, "output mode must be zero"
            # int(True) == 1, int(False) == 0
            out_pos = my_program[ip + 3]
            my_program[out_pos] = int(comparator(input1, input2))
            ip += 4
        elif opcode == 99:
            # halt
            # print("Done")
            ip = max_ip     # the end
        else:
            raise Exception("Invalid opcode: {}".format(opcode))

    return my_program, result


async def run_a_program(plist):
    inputs = [5, 6, 7, 8, 9]
    results = []
    for inputset in itertools.permutations(inputs):
        # create and initialize the queues
        # Q1 - F1 - Q2 - F2 - Q3 - F3 - Q4 - F5 +
        # +--------------------------------------
        queues = []
        for q in range(len(inputs)):
            queues.append(asyncio.Queue())
        for pos in range(len(inputs)):
            # "phases"
            queues[pos].put_nowait(inputset[pos])
        # first value
        queues[0].put_nowait(0)
        # set up the computationsY
        filters = []
        max_queue = len(queues) - 1
        for pos in range(len(inputs)):
            input_q = queues[pos]
            output_q = queues[pos + 1 if pos < max_queue else 0]
            filter = asyncio.create_task(
                compute(plist, input_q, output_q))
            filters.append(filter)
        returns = await asyncio.gather(*filters, return_exceptions=True)
        # value returned by the final filter
        (_, val) = returns[-1]
        results.append((inputset, val))

    def grabber(x):
        return x[1]

    return max(results, key=grabber)


def read_data(filename):
    """py
    Each line is a "program":
    name1:op1,op2,...opX
    """

    alldata = {}

    with open(filename) as f:
        for line in f:
            (name, prog) = line.rstrip().split(':')
            alldata[name] = [int(x) for x in prog.split(',')]

    return alldata


def main():
    data = read_data('day7-data.txt')

    for pname in ['test1b', 'test2b', 'data']:
        program = data[pname]
        (maxparm, maxvalue) = asyncio.run(run_a_program(program))
        print("{}: {} <- {}".format(pname, maxparm, maxvalue))


if __name__ == "__main__":
    main()