Несоответствие типов в дереве соединений ⇐ Python

[Anonymous]

Я строю связи с функциональными нейронами, но встречаю несоответствие типов, которое не понимаю. Проблема возникает в рекурсии, поэтому это непросто. Функции applyInputs и output ведут себя аналогично, но первая не работает, а вторая работает. Я подозреваю, что это трюк Python...
import copy
import itertools
import math
import numpy as np
import pydot
import pytest
import random
import sympy
import textdistance
import time
import typing

eps = 1e-6

class Dict(dict):
def __init__(self, d: dict = {}):
super().__init__(d)
self.removedItems = {}
self.addedItems = {}

def __sub__(self, d):
newDict = copy.deepcopy(self)

for k, v in self.items():
if (not k in d.keys()):
newDict.removedItems[k] = v
else:
del newDict[k]

for k, v in d.items():
if (not k in self.keys()):
newDict.addedItems[k] = v

return newDict

def __add__(self, d):
newDict = copy.deepcopy(self)

for k, v in d.removedItems.items():
del newDict[k]

for k, v in d.addedItems.items():
newDict[k] = v

return newDict

def index(l: list, value):
bestCost = math.inf
bestIndex = -1

for i in range(0, len(l)):
cost = heuristic(l, value)

if (cost < bestCost):
bestCost = cost
bestIndex = i

return bestIndex

def fit(target: object, function, parameters: list

• , limit: int = 1e5, percentage: float = 0.001) -> list:
  initialCells = []
  
  barycenter = []
  
  for p in parameters:
  barycenter.append(p[len(p) // 2])
  
  initialCells.append(barycenter)
  
  """
  count = 1
  
  for p in parameters:
  count *= len(p)
  
  while (len(initialCells) < percentage * count):
  point = []
  
  for p in parameters:
  point.append(random.choice(p))
  
  initialCells.append(point)
  """
  
  visitedCells = []
  cells = []
  
  for c in initialCells:
  directions = []
  
  for i in range(0, len(c)):
  direction = []
  
  for v in c:
  if (isinstance(v, np.ndarray)):
  x = copy.deepcopy(v)
  x.fill(0)
  else:
  x = type(v)()
  
  direction.append(x)
  
  j = index(parameters, c)
  direction = parameters[j - 1] - c
  directions.append(direction)
  direction = copy.deepcopy(direction)
  direction = parameters[(j + 1) % len(parameters)] - c
  directions.append(direction)
  
  try:
  value = function(c)
  cost = heuristic(value, target)
  
  cells.append((cost, c, directions))
  except:
  pass
  
  cells = sorted(cells, key = lambda x: x[0])
  
  if (len(cells) == 0):
  return None
  
  bestCost = cells[0][0]
  bestParams = cells[0][1]
  
  while (len(cells)):
  cell = cells[0]
  del cells[0]
  
  if (index(visitedCells, cell) != -1):
  continue
  
  visitedCells.append(cell)
  
  if (len(visitedCells) > limit):
  break
  
  for direction in cell[2]:
  c = copy.deepcopy(cell[1])
  
  for i in range(0, len(c)):
  c[i] += direction[i]
  
  d = []
  
  for i in range(0, len(c)):
  direction = []
  
  for v in c:
  if (isinstance(v, np.ndarray)):
  x = copy.deepcopy(v)
  x.fill(0)
  else:
  x = type(v)()
  
  direction.append(x)
  
  j = index(parameters[i], c[i])
  direction[i] = parameters[i][j - 1] - c[i]
  d.append(direction)
  direction = copy.deepcopy(direction)
  direction[i] = parameters[i][(j + 1) % len(parameters[i])] - c[i]
  d.append(direction)
  
  try:
  value = function(c)
  cost = heuristic(value, target)
  
  if (cost < bestCost):
  bestCost = cost
  bestParams = c
  
  if (cost < eps):
  return bestParams
  
  cells.append((cost, c, d))
  except:
  pass
  
  cells = sorted(cells, key = lambda x: x[0])
  cells = cells[:50]
  
  return bestParams
  
  def heuristic(val, target):
  if (isinstance(target, str)):
  s = val
  
  cost = 0 if isinstance(val, str) else 1
  
  try:
  s = str(val)
  except:
  pass
  
  if (isinstance(s, str)):
  a, b = s, target
  
  if (b in a):
  a, b = b, a
  
  if (a in b):
  return cost + 1 - 1 / target.count(a) + 1 / (1 + len(a)) - 1 / (1 + len(b))
  
  return cost + 1 / (1 + len(a)) - 1 / (1 + len(b)) + textdistance.Levenshtein().distance(a, b)
  else:
  return abs(hash(val) - hash(target))
  elif (isinstance(target, sympy.Expr)):
  if (isinstance(val, sympy.Expr)):
  if (val == target):
  return 0
  else:
  return heuristic(str(val), str(target))
  else:
  return 1 + heuristic(str(val), str(target))
  
  try:
  v = np.array(val)
  t = np.array(target)
  
  if (v.shape != t.shape):
  return 100 + abs(np.sum(v) - np.sum(t))
  
  return np.linalg.norm(np.subtract(v, t))
  except:
  try:
  return 999.0
  #return abs(hash(val) - hash(target))
  except:
  return 999.0
  
  def toTuple(x) -> tuple:
  if (isinstance(x, dict)):
  return tuple(map(tuple, x.items()))
  
  if (isinstance(x, np.ndarray) or isinstance(x, list) or isinstance(x, set)):
  l = []
  
  for v in x:
  l.append(normalizedOutput(v))
  
  return tuple(l)
  
  return tuple(x)
  
  def normalizedOutput(output):
  if (isinstance(output, list) or isinstance(output, np.ndarray) or isinstance(output, set) or isinstance(output, dict)):
  return toTuple(output)
  
  return output
  
  class Neuron:
  def __init__(self, name, function, inputTypes: tuple[type], outputType: type):
  self.name = name
  self.function = function
  self.inputTypes = inputTypes
  self.outputType = outputType
  
  def dot(self, index: int = 0):
  s = ""
  startIndex = copy.deepcopy(index)
  
  for i in range(0, len(self.inputTypes)):
  s += "n" + str(index) + ' [label="' + str(self.inputTypes[i]).replace("", "") + '", shape=circle, style=fill];\n'
  s += "n" + str(index) + " -> n" + str(startIndex + len(self.inputTypes)) + ";\n"
  index += 1
  
  s += "n" + str(index) + ' [label="' + self.name + '", shape=circle, style=fill];\n'
  index += 1
  
  s += "n" + str(index) + ' [label="' + str(self.outputType).replace("", "") + '", shape=circle, style=fill];\n'
  s += "n" + str(startIndex + len(self.inputTypes)) + " -> n" + str(index) + ";\n"
  index += 1
  
  return s, index
  
  def __eq__(self, other):
  if (isinstance(other, Neuron)):
  return self.name == other.name and self.function == other.function and self.inputTypes == other.inputTypes and self.outputType == other.outputType
  
  return False
  
  def __hash__(self):
  return hash(self.function)
  
  class Connection:
  def __init__(self, neuron: Neuron, inputs: list):
  self.neuron = neuron
  self.inputs = list(inputs)
  
  for i in range(0, len(self.inputs)):
  if (isinstance(self.inputs[i], Connection)):
  self.inputs[i] = copy.copy(self.inputs[i])
  
  def __eq__(self, other):
  if (isinstance(other, Connection)):
  return self.neuron == other.neuron and self.inputs == other.inputs
  
  return False
  
  def __hash__(self):
  return hash(self.neuron) + int(np.sum([hash(x) for x in self.inputs]))
  
  def applyInputs(self, inputs: list):
  index = 0
  
  for i in range(0, len(self.inputs)):
  input_ = self.inputs[i]
  
  if (isinstance(input_, Connection)):
  size = len(input_.inputTypes())
  input_.applyInputs(inputs[index:index + size])
  index += size
  else:
  if (index < len(inputs)):
  if (isinstance(inputs[index], Connection)):
  assert(inputs[index].neuron.outputType == self.neuron.inputTypes[i])
  elif (isinstance(inputs[index], Neuron)):
  assert(inputs[index].outputType == self.neuron.inputTypes[i])
  elif (isinstance(inputs[index], type)):
  assert(inputs[index] == self.neuron.inputTypes[i])
  else:
  assert(type(inputs[index]) == self.neuron.inputTypes[i])
  
  if (isinstance(inputs[index], Connection)):
  self.inputs[i] = copy.copy(inputs[index])
  else:
  self.inputs[i] = inputs[index]
  
  index += 1
  
  def output(self, inputs: list = None):
  args = []
  index = 0
  
  for i in range(0, len(self.inputs)):
  input_ = self.inputs[i]
  
  if (isinstance(input_, Connection)):
  if (inputs == None):
  args.append(input_.output())
  else:
  size = len(input_.inputTypes())
  args.append(input_.output(inputs[index:index + size]))
  index += size
  else:
  if (inputs != None):
  if (index < len(inputs)):
  args.append(inputs[index])
  index += 1
  elif (isinstance(input_, Neuron)):
  args.append(input_.function())
  else:
  args.append(input_)
  
  return self.neuron.function(*args)
  
  def inputTypes(self) -> list:
  types = []
  
  for input_ in self.inputs:
  if (isinstance(input_, Connection)):
  types.extend(input_.inputTypes())
  elif (isinstance(input_, Neuron)):
  types.extend(input_.inputTypes)
  elif (isinstance(input_, typing.GenericAlias)):
  types.append(input_)
  elif (isinstance(input_, type)):
  types.append(input_)
  else:
  types.append(type(input_))
  
  return types
  
  def cost(self) -> int:
  cost = 0
  
  for i in self.inputs:
  cost += 1
  
  if (type(i) is Connection):
  cost += i.cost()
  
  return cost
  
  def toStr(self) -> str:
  args = []
  
  for i in self.inputs:
  if (type(i) is Connection):
  args.append(i.toStr())
  elif (type(i) is Neuron):
  args.append(i.name)
  else:
  args.append(str(i))
  
  s = self.neuron.name
  
  if (len(self.neuron.inputTypes)):
  s += "(" + ", ".join(args) + ")"
  
  return s
  
  class Brain:
  def __init__(self, neurons: list[Neuron]):
  self.neurons = neurons
  
  def directLearn(self, targets: list = [], targetTypes: list = None, level: int = 3):
  if (targetTypes == None):
  targetTypes = []
  
  for t in targets:
  targetTypes.append(type(t))
  
  assert(len(targets) == len(targetTypes))
  
  parameters = dict()
  types = dict()
  
  for neuron in self.neurons:
  if (len(neuron.inputTypes) == 0):
  d = parameters.get(neuron.outputType, dict())
  value = neuron.function()
  types[neuron.outputType] = neuron.outputType
  value = normalizedOutput(value)
  l = d.get(value, [])
  l.append(neuron)
  d[value] = l
  parameters[neuron.outputType] = d
  
  learnedConnections = [None] * len(targets)
  
  connectionMapping = {}
  
  connections = set()
  
  for neuron in self.neurons:
  if (len(neuron.inputTypes)):
  connections.add(Connection(neuron, neuron.inputTypes))
  
  for ll in range(0, level):
  mapping = copy.deepcopy(connectionMapping)
  
  for connection in connections:
  l = mapping.get(connection.neuron.outputType, set())
  
  args = connection.neuron.inputTypes
  args = [[x] for x in args]
  
  for j in range(0, len(connection.neuron.inputTypes)):
  args[j].extend(connectionMapping.get(connection.neuron.inputTypes[j], []))
  
  if (len(args)):
  product = itertools.product(*args)
  
  for p in product:
  c = copy.deepcopy(connection)
  c.applyInputs(p)
  l.add(c)
  
  if (len(l)):
  if (connection.neuron.outputType in targetTypes):
  newL = set()
  
  for c in l:
  args = []
  
  connectionInputTypes = c.inputTypes()
  
  for t in connectionInputTypes:
  l = list(parameters[t].keys())
  
  for i in range(0, len(l)):
  if (types[t] == np.ndarray):
  l[i] = np.asarray(l[i])
  else:
  l[i] = types[t](l[i])
  
  args.append(l)
  
  tt = [x == connection.neuron.outputType for x in targetTypes]
  
  evalConnections = []
  
  for j in range(0, len(tt)):
  if (tt[j]):
  params = fit(targets[j], c.output, args)
  
  if (params == None):
  continue
  
  inputs = []
  
  for k in range(0, len(connectionInputTypes)):
  inputs.append(parameters[connectionInputTypes[k]][normalizedOutput(params[k])][0])
  
  try:
  value = c.output(params)
  
  evalConnections.append((heuristic(value, targets[j]), c, inputs))
  
  for kk in range(0, len(inputs)):
  assert(inputs[kk].outputType == connectionInputTypes[kk])
  except:
  pass
  
  evalConnections = sorted(evalConnections, key = lambda x: (x[0], x[1].cost()))
  
  if (len(evalConnections)):
  h = math.inf
  
  if (learnedConnections[j] != None):
  s = learnedConnections[j].toStr()
  h = heuristic(learnedConnections[j].output(), targets[j])
  
  c = copy.deepcopy(evalConnections[0][1])
  c.applyInputs(evalConnections[0][2])
  
  if (h > heuristic(c.output(), targets[j])):
  learnedConnections[j] = c
  
  for x in evalConnections[:100]:
  newL.add(x[1])
  
  l = newL
  
  mapping[connection.neuron.outputType] = l
  
  h = [math.inf] * len(learnedConnections)
  
  for j in range(0, len(learnedConnections)):
  if (learnedConnections[j] != None):
  h[j] = heuristic(targets[j], learnedConnections[j].output())
  
  if (np.sum(h) < eps):
  ll = level
  break
  
  connectionMapping = mapping
  
  connections = []
  
  for k, v in connectionMapping.items():
  connections += v
  
  connections = set(connections)
  
  return learnedConnections
  
  def taskPairs(task: str) -> tuple:
  import json
  import urllib.request
  
  url = urllib.request.urlopen("https://raw.githubusercontent.com/arcpr ... /training/" + task + ".json")
  data = json.loads(url.read().decode())
  
  train = data["train"]
  trainPairs = []
  
  for v in train:
  trainPairs.append((np.array(v["input"]), np.array(v["output"])))
  
  test = data["test"]
  testPairs = []
  
  for v in test:
  testPairs.append((np.array(v["input"]), np.array(v["output"])))
  
  return (trainPairs, testPairs)
  
  def inputOutputPair(pairs):
  inputs = []
  outputs = []
  
  for p in pairs:
  inputs.append(p[0])
  outputs.append(p[1])
  
  return (inputs, outputs)
  
  def map_(x: np.ndarray, mapping: Dict):
  y = copy.deepcopy(x)
  
  for i in range(0, y.shape[0]):
  for j in range(0, y.shape[1]):
  y[i, j] = mapping.get(y[i, j], y[i, j])
  
  return y
  
  def associate(mapping: Dict, a, b):
  m = copy.deepcopy(mapping)
  
  m[a] = b
  
  return m
  
  def test_task0d3d703e(): #Color mapping
  digitNeurons = []
  
  for i in range(0, 10):
  digitNeurons.append(Neuron(str(i), lambda i = i: i, [], int))
  
  mapNeuron = Neuron("map", lambda x, m: [map_(v, m) for v in x], [np.ndarray, Dict], np.ndarray)
  mappingNeuron = Neuron("mapping", lambda: Dict(), [], Dict)
  associateNeuron = Neuron("associate", lambda mapping, a, b: associate(mapping, a, b), [Dict, int, int], Dict)
  inputNeuron = Neuron("input", lambda: np.array([]), [], np.ndarray)
  
  neurons = []
  neurons += digitNeurons
  neurons.append(mapNeuron)
  neurons.append(mappingNeuron)
  neurons.append(associateNeuron)
  neurons.append(inputNeuron)
  
  brain = Brain(neurons)
  
  trainPairs, testPairs = taskPairs("0d3d703e")
  
  input_, output = inputOutputPair(trainPairs)
  input_ = np.array(input_)
  output = np.array(output)
  
  inputNeuron.function = lambda input_ = input_: input_
  
  connections = brain.directLearn([output], level = 3)#5)
  connection = connections[0]
  
  print(connection.toStr())
  
  input_, output = inputOutputPair(testPairs)
  input_ = np.array(input_)
  output = np.array(output)
  
  inputNeuron.function = lambda input_ = input_: input_
  
  assert(not heuristic(connection.output(), output))
  
  Вот ошибка с pytest:
  __________________________________________________ test_task0d3d703e __________________________________________________
  
  def test_task0d3d703e(): #Color mapping
  digitNeurons = []
  
  for i in range(0, 10):
  digitNeurons.append(Neuron(str(i), lambda i = i: i, [], int))
  
  mapNeuron = Neuron("map", lambda x, m: [map_(v, m) for v in x], [np.ndarray, Dict], np.ndarray)
  mappingNeuron = Neuron("mapping", lambda: Dict(), [], Dict)
  associateNeuron = Neuron("associate", lambda mapping, a, b: associate(mapping, a, b), [Dict, int, int], Dict)
  inputNeuron = Neuron("input", lambda: np.array([]), [], np.ndarray)
  
  neurons = []
  neurons += digitNeurons
  neurons.append(mapNeuron)
  neurons.append(mappingNeuron)
  neurons.append(associateNeuron)
  neurons.append(inputNeuron)
  
  brain = Brain(neurons)
  
  trainPairs, testPairs = taskPairs("0d3d703e")
  
  input_, output = inputOutputPair(trainPairs)
  input_ = np.array(input_)
  output = np.array(output)
  
  inputNeuron.function = lambda input_ = input_: input_
  
  > connections = brain.directLearn([output], level = 3)#5)
  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  
  mcve.py:625:
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
  mcve.py:517: in directLearn
  c.applyInputs(evalConnections[0][2])
  mcve.py:312: in applyInputs
  input_.applyInputs(inputs[index:index + size])
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
  
  self =
  inputs = [, ]
  
  def applyInputs(self, inputs: list):
  index = 0
  
  for i in range(0, len(self.inputs)):
  input_ = self.inputs[i]
  
  if (isinstance(input_, Connection)):
  size = len(input_.inputTypes())
  input_.applyInputs(inputs[index:index + size])
  index += size
  else:
  if (index < len(inputs)):
  if (isinstance(inputs[index], Connection)):
  assert(inputs[index].neuron.outputType == self.neuron.inputTypes[i])
  elif (isinstance(inputs[index], Neuron)):
  > assert(inputs[index].outputType == self.neuron.inputTypes[i])
  E AssertionError: assert ==
  E + where = .outputType
  
  mcve.py:319: AssertionError
  
  
  Подробнее здесь: https://stackoverflow.com/questions/798 ... ction-tree