130 lines
3.3 KiB
Python
130 lines
3.3 KiB
Python
|
import sys
|
||
|
|
import matrix
|
||
|
|
import shared
|
||
|
|
import scanf
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import List, Dict
|
||
|
|
from pprint import pprint
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Valve:
|
||
|
|
label: str
|
||
|
|
rate: int
|
||
|
|
tunnels: List[str]
|
||
|
|
opened_at: int = -1
|
||
|
|
potential: Dict[str,int] = None
|
||
|
|
|
||
|
|
def set_potential(self, valves):
|
||
|
|
self.potential = {}
|
||
|
|
for tunnel in self.tunnels:
|
||
|
|
self.potential[tunnel] = valves[tunnel].rate
|
||
|
|
|
||
|
|
def highest_potential(self):
|
||
|
|
return max(self.potential, key=self.potential.get)
|
||
|
|
|
||
|
|
def parse(rows):
|
||
|
|
valves = {}
|
||
|
|
for row in rows:
|
||
|
|
try:
|
||
|
|
left,right = row.split(" valves ")
|
||
|
|
except ValueError:
|
||
|
|
left,right = row.split(" valve ")
|
||
|
|
valve, rate = scanf.scanf("Valve %s has flow rate=%d; %*s %*s to", left)
|
||
|
|
if "," in right:
|
||
|
|
tunnels = right.split(", ")
|
||
|
|
else:
|
||
|
|
tunnels = [right,]
|
||
|
|
valves[valve] = Valve(label=valve,rate=rate,tunnels=tunnels)
|
||
|
|
|
||
|
|
for _,v in valves.items():
|
||
|
|
v.set_potential(valves)
|
||
|
|
return valves
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def part1(rows, sample=False):
|
||
|
|
p1 = Part1(rows,sample, 30)
|
||
|
|
p1.run()
|
||
|
|
|
||
|
|
|
||
|
|
class Part1:
|
||
|
|
def __init__(self, rows, sample, minutes):
|
||
|
|
self.rows = rows
|
||
|
|
self.sample = sample
|
||
|
|
self.valves = parse(rows)
|
||
|
|
self.cur = self.valves["AA"]
|
||
|
|
self.tick = 1
|
||
|
|
self.minutes = minutes
|
||
|
|
|
||
|
|
def do_tick(self, minute):
|
||
|
|
pressure = 0
|
||
|
|
opened = []
|
||
|
|
for _, valve in self.valves.items():
|
||
|
|
if valve.opened_at > 0:
|
||
|
|
continue
|
||
|
|
pressure += valve.rate
|
||
|
|
print(f"== Minute {minute} == [ {self.cur.label} ]")
|
||
|
|
print(f"\tValves {', '.join(opened)} are open, releasing {pressure} pressure")
|
||
|
|
|
||
|
|
|
||
|
|
def open_or_move(self):
|
||
|
|
#print(self.cur)
|
||
|
|
hi = self.cur.highest_potential()
|
||
|
|
potential_elsewhere = self.cur.potential
|
||
|
|
pe = potential_elsewhere[hi]
|
||
|
|
if pe > self.cur.rate:
|
||
|
|
#print(f"should move to {hi}")
|
||
|
|
self.move(hi)
|
||
|
|
elif self.cur.rate > pe:
|
||
|
|
print(f"should open {self.cur.label}")
|
||
|
|
self.open()
|
||
|
|
|
||
|
|
def move(self, where):
|
||
|
|
self.tick += 1
|
||
|
|
self.cur = self.valves[where]
|
||
|
|
print("\tMove to valve", where)
|
||
|
|
|
||
|
|
def open(self):
|
||
|
|
if self.cur.opened_at >= 0 :
|
||
|
|
raise Exception("tried to open valve already opened")
|
||
|
|
self.tick += 1
|
||
|
|
self.valves[self.cur.label].opened_at = self.tick
|
||
|
|
self.cur = self.valves[self.cur.label]
|
||
|
|
print(self.cur)
|
||
|
|
|
||
|
|
def calculate_total_flow(self):
|
||
|
|
total = 0
|
||
|
|
for label, valve in self.valves.items():
|
||
|
|
if valve.is_open:
|
||
|
|
total += valve.rate * (30 - valve.opened_at)
|
||
|
|
return total
|
||
|
|
|
||
|
|
def run(self):
|
||
|
|
for minute in range(1, self.minutes+1):
|
||
|
|
self.do_tick(minute)
|
||
|
|
self.open_or_move()
|
||
|
|
|
||
|
|
print("total flow:")
|
||
|
|
print(self.calculate_total_flow())
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
sample = False
|
||
|
|
if sys.argv[-1] == "--sample":
|
||
|
|
sample = True
|
||
|
|
rows = [row for row in shared.load_rows(16)]
|
||
|
|
|
||
|
|
with shared.elapsed_timer() as elapsed:
|
||
|
|
part1(rows, sample)
|
||
|
|
print("🕒", elapsed())
|
||
|
|
|
||
|
|
#with shared.elapsed_timer() as elapsed:
|
||
|
|
# part2(rows, sample)
|
||
|
|
# print("🕒", elapsed())
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|