Автор – Сухачов Денис Павлович
This is a work, and the author’s right to a work under international law comes into force from the moment the work is created.
Нижче наведено оновлений варіант коду з наступними доповненнями:
python
Копіювати
import numpy as np
import pickle
import os
import time
import cv2
import logging
import speech_recognition as sr
from concurrent.futures import ThreadPoolExecutor, as_completed
# Налаштування логування
logging.basicConfig(level=logging.INFO, format=’%(asctime)s [%(levelname)s] %(message)s’)
# ======================= Клас Neuron =======================
class Neuron:
def __init__(self, vector=None, memory_file=’neuron_memory.pkl’,
consciousness_frequency=1.0, thought_speed=1.0, info_quality=1.0,
discreteness=1.0, streamness=1.0, density=1.0, interconnectedness=1.0):
“””
Нейрон з вектором ваг, внутрішньою пам’яттю та параметрами «свідомості».
“””
if vector is None:
self.vector = np.random.rand(3)
else:
self.vector = np.array(vector, dtype=float)
norm = np.linalg.norm(self.vector)
if norm > 0:
self.vector = self.vector / norm
self.memory = {}
self.memory_file = memory_file
self.load_memory()
self.connections = {}
self.consciousness_frequency = consciousness_frequency
self.thought_speed = thought_speed
self.info_quality = info_quality
self.discreteness = discreteness
self.streamness = streamness
self.density = density
self.interconnectedness = interconnectedness
self.logic_rules = {
“fact”: [],
“unverified”: [],
“probable”: [],
“a_priori”: []
}
logging.info(“Neuron ініціалізовано з вектором %s”, self.vector)
def add_connection(self, neuron, weight_coefficient=1.0):
self.connections[neuron] = weight_coefficient
logging.info(“Додано зв’язок з нейроном із коефіцієнтом %s”, weight_coefficient)
def update_weight(self, input_vector, learning_rate=0.1):
input_vector = np.array(input_vector, dtype=float)
self.vector += learning_rate * input_vector
norm = np.linalg.norm(self.vector)
if norm > 0:
self.vector = self.vector / norm
logging.info(“Вектор ваг оновлено до %s”, self.vector)
def store_memory(self, key, value, weight=1.0, category=”fact”):
self.memory[key] = {“value”: value, “weight”: weight, “category”: category}
if category in self.logic_rules:
self.logic_rules[category].append(key)
self.save_memory()
logging.info(“Записано пам’ять: %s -> %s”, key, value)
def save_memory(self):
with open(self.memory_file, ‘wb’) as f:
pickle.dump(self.memory, f)
logging.info(“Пам’ять збережено у файл %s”, self.memory_file)
def load_memory(self):
if os.path.exists(self.memory_file):
with open(self.memory_file, ‘rb’) as f:
self.memory = pickle.load(f)
logging.info(“Пам’ять завантажено з файлу %s”, self.memory_file)
def activate(self, input_vector):
input_vector = np.array(input_vector, dtype=float)
activation_value = np.dot(self.vector, input_vector)
for neuron, coeff in self.connections.items():
activation_value += coeff * np.dot(neuron.vector, input_vector)
modulation = self.consciousness_frequency * self.thought_speed * self.info_quality
activation_value *= modulation
logging.info(“Активація нейрона: %s”, activation_value)
return activation_value
def update_consciousness_parameters(self, consciousness_frequency=None, thought_speed=None,
info_quality=None, discreteness=None,
streamness=None, density=None, interconnectedness=None):
if consciousness_frequency is not None:
self.consciousness_frequency = consciousness_frequency
if thought_speed is not None:
self.thought_speed = thought_speed
if info_quality is not None:
self.info_quality = info_quality
if discreteness is not None:
self.discreteness = discreteness
if streamness is not None:
self.streamness = streamness
if density is not None:
self.density = density
if interconnectedness is not None:
self.interconnectedness = interconnectedness
logging.info(“Оновлено параметри свідомості: %s”, self.assess_state())
def assess_state(self):
return {
“consciousness_frequency”: self.consciousness_frequency,
“thought_speed”: self.thought_speed,
“info_quality”: self.info_quality,
“discreteness”: self.discreteness,
“streamness”: self.streamness,
“density”: self.density,
“interconnectedness”: self.interconnectedness
}
def logical_evaluation(self, input_data):
if isinstance(input_data, str) and “exp” in input_data.lower():
category = “fact”
else:
category = “probable”
key = f”evaluation_{time.time()}”
self.store_memory(key, input_data, weight=1.0, category=category)
logging.info(“Логічна оцінка: %s -> %s”, input_data, category)
return category
def associate_memories(self, key1, key2, association_strength=0.1):
if key1 in self.memory and key2 in self.memory:
combined_key = f”{key1}-{key2}”
combined_value = (self.memory[key1][“value”], self.memory[key2][“value”])
combined_weight = (self.memory[key1][“weight”] + self.memory[key2][“weight”]) * association_strength
self.memory[combined_key] = {“value”: combined_value, “weight”: combined_weight, “category”: “associated”}
self.save_memory()
logging.info(“Асоціація створена між %s та %s”, key1, key2)
# ======================= Клас Perceptron =======================
class Perceptron:
def __init__(self, neurons=None):
if neurons is None:
self.neurons = []
else:
self.neurons = neurons
self.mental_constructions = {}
logging.info(“Perceptron ініціалізовано із %s нейронами”, len(self.neurons))
def add_neuron(self, neuron):
self.neurons.append(neuron)
logging.info(“Нейрон додано до перцептрона.”)
def evaluate_network(self, input_vector):
outputs = [neuron.activate(input_vector) for neuron in self.neurons]
global_activation = sum(outputs) / len(outputs) if outputs else 0
logging.info(“Глобальна активація мережі: %s”, global_activation)
return global_activation
def build_mental_construction(self, construct_name, neuron_keys):
combined_data = {}
for neuron in self.neurons:
for key in neuron.memory:
if key in neuron_keys:
combined_data[key] = neuron.memory[key]
self.mental_constructions[construct_name] = combined_data
logging.info(“Ментальна конструкція ‘%s’ побудована.”, construct_name)
return self.mental_constructions[construct_name]
def assess_network_state(self):
states = [neuron.assess_state() for neuron in self.neurons]
logging.info(“Стан мережі: %s”, states)
return states
# ======================= Клас ImageProcessor =======================
class ImageProcessor:
def __init__(self, process_color=True):
“””
Якщо process_color=True, зображення не переводяться у відтінки сірого.
Для повноколірної обробки ми будемо обчислювати середній колір за каналами.
“””
self.process_color = process_color
def preprocess_image(self, image_path):
image = cv2.imread(image_path)
if image is None:
error_msg = “Зображення не знайдено за шляхом: ” + image_path
logging.error(error_msg)
raise ValueError(error_msg)
if self.process_color:
# Не переводимо в градації сірого – працюємо з кольоровим зображенням
return image
else:
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
return gray
def extract_color_features(self, image):
“””
Для повноколірної обробки обчислюємо середній колір (B, G, R) по всіх пікселях.
“””
mean_color = np.mean(image, axis=(0, 1))
logging.info(“Обчислено середній колір: %s”, mean_color)
return mean_color
def extract_hu_moments(self, image):
# Якщо працюємо в режимі сірого, можемо обчислювати Hu Moments
moments = cv2.moments(image)
huMoments = cv2.HuMoments(moments).flatten()
huMoments = -np.sign(huMoments) * np.log10(np.abs(huMoments) + 1e-10)
return huMoments
def process_image(self, image_path):
image = self.preprocess_image(image_path)
if self.process_color:
# Виділяємо простий колірний дескриптор – середній колір (B, G, R)
features = self.extract_color_features(image)
else:
preprocessed = image # вже в градаціях сірого
features = self.extract_hu_moments(preprocessed)
logging.info(“Оброблено зображення: отримано ознаки %s”, features)
return features
# ======================= Клас VoiceProcessor =======================
class VoiceProcessor:
def __init__(self, recognizer=None):
self.recognizer = recognizer if recognizer is not None else sr.Recognizer()
def process_audio(self, audio_path, engine=”sphinx”):
with sr.AudioFile(audio_path) as source:
audio = self.recognizer.record(source)
try:
if engine.lower() == “google”:
text = self.recognizer.recognize_google(audio, language=”uk-UA”)
else:
text = self.recognizer.recognize_sphinx(audio, language=”uk-UA”)
logging.info(“Розпізнано аудіо: %s”, text)
return text
except sr.UnknownValueError:
logging.warning(“Розпізнавання аудіо невдалося (UnknownValueError)”)
return “Невідомо”
except sr.RequestError as e:
err_msg = f”Помилка сервісу розпізнавання: {e}”
logging.error(err_msg)
return err_msg
# ======================= Клас VideoProcessor =======================
class VideoProcessor:
def __init__(self, image_processor):
self.image_processor = image_processor
self.frame_cache = {}
def process_frame(self, frame, frame_index):
if frame_index in self.frame_cache:
return self.frame_cache[frame_index]
# Якщо обробка повноколірна, використовується оригінальне зображення
features = self.image_processor.extract_color_features(frame) if self.image_processor.process_color \
else self.image_processor.extract_hu_moments(frame)
self.frame_cache[frame_index] = features
return features
def process_video(self, video_path, frame_interval=0.33):
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
error_msg = “Не вдалося відкрити відео за шляхом: ” + video_path
logging.error(error_msg)
raise ValueError(error_msg)
fps = cap.get(cv2.CAP_PROP_FPS)
skip_frames = int(round(fps * frame_interval))
features_list = []
frame_index = 0
tasks = []
executor = ThreadPoolExecutor(max_workers=4)
while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
if frame_index % skip_frames == 0:
tasks.append(executor.submit(self.process_frame, frame, frame_index))
frame_index += 1
cap.release()
for future in as_completed(tasks):
try:
features_list.append(future.result())
except Exception as exc:
logging.error(“Помилка обробки кадру: %s”, exc)
executor.shutdown()
logging.info(“Оброблено відео: отримано %s наборів ознак”, len(features_list))
return features_list
# ======================= Клас ImageGenerator =======================
class ImageGenerator:
def generate_image(self, query):
“””
Генерує зображення або діаграму за запитом.
Якщо запит містить слово “diagram”, малює просту схему,
інакше створює зображення з текстом.
“””
# Створимо порожнє зображення (500×500) з білим фоном
img = np.ones((500, 500, 3), dtype=np.uint8) * 255
if “diagram” in query.lower():
# Намалюємо просту діаграму: прямокутник, коло, лінію і текст
cv2.rectangle(img, (50, 50), (150, 150), (0, 0, 255), 3)
cv2.circle(img, (300, 300), 50, (255, 0, 0), -1)
cv2.line(img, (0, 0), (500, 500), (0, 255, 0), 2)
cv2.putText(img, “Diagram”, (180, 250), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 0), 2)
else:
cv2.putText(img, query, (50, 250), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 0), 2)
logging.info(“Згенеровано зображення за запитом: %s”, query)
return img
# ======================= Клас MemoryGraph =======================
class MemoryGraph:
def __init__(self):
self.nodes = {}
def add_node(self, node_id, data):
self.nodes[node_id] = {“data”: data, “neighbors”: {}}
logging.info(“Додано вузол у граф пам’яті: %s”, node_id)
def add_edge(self, node_id1, node_id2, weight=1.0):
if node_id1 in self.nodes and node_id2 in self.nodes:
self.nodes[node_id1][“neighbors”][node_id2] = weight
self.nodes[node_id2][“neighbors”][node_id1] = weight
logging.info(“Додано ребро між %s та %s”, node_id1, node_id2)
def get_node(self, node_id):
return self.nodes.get(node_id, None)
def search_by_data(self, query):
results = []
for node_id, info in self.nodes.items():
if query.lower() in str(info[“data”]).lower():
results.append(node_id)
logging.info(“Пошук за даними ‘%s’ повернув: %s”, query, results)
return results
# ======================= Клас CognitiveMonitor =======================
class CognitiveMonitor:
def __init__(self, perceptron):
self.perceptron = perceptron
def report(self):
states = self.perceptron.assess_network_state()
avg_consciousness = sum(state[“consciousness_frequency”] for state in states) / len(states)
logging.info(“Cognitive Monitor Report: Average Consciousness Frequency: %s”, avg_consciousness)
def adjust_parameters(self, target_frequency=1.0):
for neuron in self.perceptron.neurons:
if neuron.consciousness_frequency < target_frequency:
neuron.update_consciousness_parameters(consciousness_frequency=target_frequency)
logging.info(“Налаштовано consciousness_frequency для нейрона до %s”, target_frequency)
# ======================= Клас MultiModalProcessor =======================
class MultiModalProcessor:
def __init__(self, image_processor, voice_processor):
self.image_processor = image_processor
self.voice_processor = voice_processor
def process(self, image_path=None, audio_path=None, text_input=None):
results = {}
if image_path:
try:
image_features = self.image_processor.process_image(image_path)
results[“image_features”] = image_features
except Exception as e:
results[“image_features_error”] = str(e)
logging.error(“Помилка обробки зображення: %s”, e)
if audio_path:
voice_text = self.voice_processor.process_audio(audio_path, engine=”sphinx”)
results[“voice_text”] = voice_text
if text_input:
results[“text_input”] = text_input
logging.info(“Результати мультимодальної обробки: %s”, results)
return results
# ======================= Клас CognitiveSystem =======================
class CognitiveSystem:
def __init__(self):
self.perceptron = Perceptron()
self.memory_graph = MemoryGraph()
self.image_processor = ImageProcessor(process_color=True)
self.voice_processor = VoiceProcessor()
self.multi_modal_processor = MultiModalProcessor(self.image_processor, self.voice_processor)
self.monitor = CognitiveMonitor(self.perceptron)
self.video_processor = VideoProcessor(self.image_processor)
self.image_generator = ImageGenerator()
logging.info(“CognitiveSystem ініціалізовано.”)
def add_neuron(self, neuron):
self.perceptron.add_neuron(neuron)
def integrate_modalities(self, image_path=None, audio_path=None, text_input=None):
results = self.multi_modal_processor.process(image_path, audio_path, text_input)
if “image_features” in results:
node_id = f”img_{time.time()}”
self.memory_graph.add_node(node_id, data=[results[“image_features”]])
if “voice_text” in results:
node_id = f”voice_{time.time()}”
self.memory_graph.add_node(node_id, data=results[“voice_text”])
if “text_input” in results:
node_id = f”text_{time.time()}”
self.memory_graph.add_node(node_id, data=results[“text_input”])
return results
def integrate_video(self, video_path):
video_features = self.video_processor.process_video(video_path, frame_interval=0.33)
if video_features:
avg_features = np.mean(video_features, axis=0)
node_id = f”video_{time.time()}”
self.memory_graph.add_node(node_id, data=avg_features.tolist())
logging.info(“Агреговані ознаки відео: %s”, avg_features)
return avg_features
else:
return None
def run(self, input_vector):
activation = self.perceptron.evaluate_network(input_vector)
logging.info(“Global Activation: %s”, activation)
return activation
def monitor_system(self):
self.monitor.report()
self.monitor.adjust_parameters(target_frequency=1.0)
def generate_image(self, query):
generated = self.image_generator.generate_image(query)
node_id = f”gen_{time.time()}”
# Зберігаємо згенероване зображення (можна конвертувати у список пікселів)
self.memory_graph.add_node(node_id, data=generated.tolist())
logging.info(“Згенеровано зображення за запитом ‘%s’, збережено як %s”, query, node_id)
return generated
# ======================= Приклад використання =======================
if __name__ == “__main__”:
# Створення когнітивної системи
system = CognitiveSystem()
# Створення та додавання нейронів до системи
neuron1 = Neuron(vector=[0.1, 0.2, 0.7],
memory_file=’neuron1_memory.pkl’,
consciousness_frequency=0.9,
thought_speed=1.2,
info_quality=0.9)
neuron2 = Neuron(vector=[0.5, 0.4, 0.1],
memory_file=’neuron2_memory.pkl’,
consciousness_frequency=0.8,
thought_speed=1.0,
info_quality=1.0)
neuron1.add_connection(neuron2, weight_coefficient=0.8)
system.add_neuron(neuron1)
system.add_neuron(neuron2)
# Інтеграція мультимодальних даних (зображення, аудіо, текст)
image_path = “sample_image.jpg” # Шлях до зображення
audio_path = “sample_audio.wav” # Шлях до аудіофайлу
text_input = “Додатковий текстовий вхід”
modalities = system.integrate_modalities(image_path=image_path,
audio_path=audio_path,
text_input=text_input)
logging.info(“Мультимодальні дані: %s”, modalities)
# Інтеграція відео (кадри кожні 0.33 секунди)
video_path = “sample_video.mp4” # Шлях до відеофайлу
try:
video_features = system.integrate_video(video_path)
logging.info(“Агреговані ознаки відео: %s”, video_features)
except Exception as e:
logging.error(“Помилка обробки відео: %s”, e)
# Використання отриманих ознак як вхідного сигналу
if modalities.get(“image_features”) is not None:
input_vector = modalities[“image_features”]
elif video_features is not None:
input_vector = video_features
else:
input_vector = [0.3, 0.6, 0.1]
system.run(input_vector)
# Генерація зображення (діаграми) за запитом
generated_img = system.generate_image(“Show me a diagram”)
cv2.imwrite(“generated_diagram.jpg”, generated_img)
logging.info(“Згенероване зображення збережено як generated_diagram.jpg”)
# Виконання метакогнітивного моніторингу
system.monitor_system()
# Побудова ментальної конструкції на основі локусів пам’яті з нейронів
neuron1.store_memory(“A”, “Поняття A”, weight=0.9, category=”fact”)
neuron1.store_memory(“B”, “Поняття B”, weight=0.8, category=”a_priori”)
construction = system.perceptron.build_mental_construction(“Construction1”, [“A”, “B”])
logging.info(“Ментальна конструкція ‘Construction1’: %s”, construction)
# Вивід графу пам’яті (список вузлів)
logging.info(“Вузли графу пам’яті:”)
for node_id, node_info in system.memory_graph.nodes.items():
logging.info(“%s: %s”, node_id, node_info)
Опис системи простими словами
Цей код створює комплексну когнітивну систему, яка здатна:
- Розпізнавати зображення у повноколірному режимі.
Замість перетворення в чорно-білий формат, система обробляє зображення в кольорі, виділяючи середні значення для кожного з каналів (B, G, R). Це дозволяє «бачити» всю колірну палітру. - Розпізнавати голосовий вхід.
Модуль VoiceProcessor перетворює аудіофайли (голос) на текст, використовуючи алгоритми розпізнавання (наприклад, Sphinx чи Google). - Обробляти відео.
Модуль VideoProcessor вибирає кадри з відео кожні 0.33 секунди, обробляє їх паралельно, отримуючи ознаки (наприклад, середній колір кожного кадру). Отримані ознаки можуть бути агреговані для опису всього відео. - Зберігати та організовувати інформацію.
Модуль MemoryGraph зберігає «локуси» пам’яті (наприклад, результати обробки зображень, аудіо, відео або згенеровані зображення) і дозволяє встановлювати асоціації між ними. - Моніторити та адаптувати роботу.
CognitiveMonitor аналізує стан мережі (середній рівень «свідомості» нейронів) і за потреби коригує параметри, щоб забезпечити стабільну роботу системи. - Інтегрувати мультимодальні дані.
MultiModalProcessor об’єднує дані із різних сенсорних каналів (зображення, голос, текст) для побудови єдиної моделі знань. - Генерувати зображення та схеми за запитом.
ImageGenerator дозволяє системі «малювати» зображення або діаграми на основі текстового запиту. Наприклад, за запитом «Show me a diagram» система створить зображення з простими геометричними фігурами та текстом.
Таким чином, система може не лише приймати та аналізувати різномодальні дані, а й сама створювати візуальний контент (зображення, схеми) за потребою. Це робить її потужним інструментом для моделювання когнітивних процесів і для застосування в інтерактивних асистентах, аналізі медіа або навіть генерації візуальних пояснень.
