encoder_test.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023-06-26 16:13:57.303048: I tensorflow/tsl/cuda/cudart_stub.cc:28] Could not find cuda drivers on your machine, GPU will not be used.\n",
      "2023-06-26 16:13:57.346907: I tensorflow/tsl/cuda/cudart_stub.cc:28] Could not find cuda drivers on your machine, GPU will not be used.\n",
      "2023-06-26 16:13:57.347819: I tensorflow/core/platform/cpu_feature_guard.cc:182] This TensorFlow binary is optimized to use available CPU instructions in performance-critical operations.\n",
      "To enable the following instructions: AVX2 FMA, in other operations, rebuild TensorFlow with the appropriate compiler flags.\n"
     ]
    }
   ],
   "source": [
    "import numpy as np\n",
    "from playsound import playsound\n",
    "from matplotlib import pyplot as plt\n",
    "from scipy.fft import fft, fftfreq, rfft, rfftfreq\n",
    "from scipy.io.wavfile import write, read \n",
    "from scipy.io import wavfile\n",
    "from time import sleep\n",
    "from math import floor\n",
    "import pyaudio\n",
    "import wave\n",
    "import random\n",
    "import pandas as pd\n",
    "import tensorflow as tf\n",
    "import keras\n",
    "import warnings\n",
    "warnings.filterwarnings(\"ignore\")\n",
    "import os\n",
    "import sys\n",
    "from time import sleep\n",
    "from project import decoder, modulator, wav_manager\n",
    "\n",
    "string = \"MANKIND IS DEAD. BLOOD IS FUEL. HELL IS FULL\"\n",
    "\n",
    "def split_string(string, length):\n",
    "    return [string[i:i+length] for i in range(0, len(string), length)]\n",
    "\n",
    "def encode(text):\n",
    "    binary_code = ''.join(format(ord(char), '08b') for char in text)\n",
    "    return binary_code\n",
    "\n",
    "def decode(binary_code):\n",
    "    text = ''.join(chr(int(binary_code[i:i+8], 2)) for i in range(0, len(binary_code), 8))\n",
    "    return text\n",
    "\n",
    "print(\"Enter text\")\n",
    "text = str(input())\n",
    "code = encode(text)\n",
    "code_array = split_string(code, 40)\n",
    "sample_rate = 44100\n",
    "wava = wav_manager()\n",
    "mod = modulator()\n",
    "mod.set_text(\"11\")\n",
    "_, signal = mod.signal_modulation()\n",
    "wava.set_writing_name(\"Affirmative.wav\")\n",
    "wava.write_sound(sample_rate, signal)\n",
    "print(\"Sending signal to start recording\")\n",
    "playsound(\"Affirmative.wav\")\n",
    "sleep(1.5)\n",
    "i = 1\n",
    "dec = decoder()\n",
    "while True:\n",
    "    filename = str(i)\n",
    "    chunk = 1024\n",
    "    FORMAT = pyaudio.paInt16\n",
    "    channels = 1\n",
    "    sample_rate = 44100\n",
    "    record_seconds = 0.2\n",
    "    p = pyaudio.PyAudio()\n",
    "    stream = p.open(format=FORMAT,\n",
    "                channels=channels,\n",
    "                rate=sample_rate,\n",
    "                input=True,\n",
    "                output=True,\n",
    "                frames_per_buffer=chunk)\n",
    "    frames = []\n",
    "    print(\"Trying to catch response...\")\n",
    "    for i in range(int(sample_rate / chunk * record_seconds)):\n",
    "      data = stream.read(chunk)\n",
    "      frames.append(data)\n",
    "    stream.stop_stream()\n",
    "    stream.close()\n",
    "    p.terminate()\n",
    "    wf = wave.open(filename, \"wb\")\n",
    "    wf.setnchannels(channels)\n",
    "    wf.setsampwidth(p.get_sample_size(FORMAT))\n",
    "    wf.setframerate(sample_rate)\n",
    "    wf.writeframes(b\"\".join(frames))\n",
    "    wf.close()\n",
    "    data = list(b\"\".join(frames))\n",
    "    wava = wav_manager()\n",
    "    wava.set_recorded_name(filename)\n",
    "    data, _ = wava.get_data_sample_rate()\n",
    "    xf, yf = dec.fourier_abs(data)\n",
    "    yf = dec.normalize_data_array(dec.secure_split(yf, 2), xf, 1000, 1500)\n",
    "    amp1, amp2 = dec.freq_sum(yf[0], xf, 1000), dec.freq_sum(yf[0], xf, 1500)\n",
    "    threshold = 6\n",
    "    koef = max(amp1, amp2) / min(amp1, amp2)\n",
    "    print(amp2)\n",
    "    if koef > threshold:\n",
    "       print(\"Signal to start recording accepted\", amp1, amp2)\n",
    "       break\n",
    "    amp1, amp2 = dec.freq_sum(yf[1], xf, dec.freq0), dec.freq_sum(yf[1], xf, dec.freq1)\n",
    "    koef = max(amp1, amp2) / min(amp1, amp2)\n",
    "    print(koef, threshold)\n",
    "    if koef > threshold:\n",
    "       print(\"Signal to start recording accepted\", amp1, amp2)\n",
    "       break\n",
    "sleep(2)\n",
    "for code in code_array:\n",
    "  print(\"trying for\", code)\n",
    "  correct = False\n",
    "  while not correct:\n",
    "    response = False\n",
    "    correct = False\n",
    "    mod.set_text(code)\n",
    "    _, signal = mod.signal_modulation()\n",
    "    wava.set_writing_name(\"Encoded_signal.wav\")\n",
    "    wava.write_sound(sample_rate, signal)\n",
    "    print(\"Playing signal\")\n",
    "    sleep(0.3)\n",
    "    playsound(\"Encoded_signal.wav\")\n",
    "    sleep(0.5)\n",
    "    record_seconds = 0.4\n",
    "    while not response:\n",
    "            p = pyaudio.PyAudio()\n",
    "            stream = p.open(format=FORMAT,\n",
    "                channels=channels,\n",
    "                rate=sample_rate,\n",
    "                input=True,\n",
    "                output=True,\n",
    "                frames_per_buffer=chunk)\n",
    "            frames = []\n",
    "            print(\"Catching signal...\")\n",
    "            for i in range(int(sample_rate / chunk * record_seconds)):\n",
    "                data = stream.read(chunk)\n",
    "                frames.append(data)\n",
    "            stream.stop_stream()\n",
    "            stream.close()\n",
    "            p.terminate()\n",
    "            wf = wave.open(filename, \"wb\")\n",
    "            wf.setnchannels(channels)\n",
    "            wf.setsampwidth(p.get_sample_size(FORMAT))\n",
    "            wf.setframerate(sample_rate)\n",
    "            wf.writeframes(b\"\".join(frames))\n",
    "            wf.close()\n",
    "            data = list(b\"\".join(frames))\n",
    "            wava = wav_manager()\n",
    "            wava.set_recorded_name(filename)\n",
    "            data, _ = wava.get_data_sample_rate()\n",
    "            xf, yf = dec.fourier_abs(data)\n",
    "            yf = dec.normalize_data_array(dec.secure_split(yf, 2), xf, 1000, 1500)\n",
    "            amp1, amp2 = dec.freq_sum(yf[0], xf, 1000), dec.freq_sum(yf[0], xf, 1500)\n",
    "            koef = amp2 / amp1\n",
    "            print(koef, amp1, amp2, \"koef1\", \"amp1\")\n",
    "            if koef > threshold:\n",
    "                print(\"Signal decoded\")\n",
    "                sleep(1)\n",
    "                correct, response = True, True\n",
    "                break\n",
    "            if 1 / koef > threshold:\n",
    "                print(\"Signal decoded with error\")\n",
    "                correct, response = False, True\n",
    "                break\n",
    "            amp1, amp2 = dec.freq_sum(yf[1], xf, dec.freq0), dec.freq_sum(yf[1], xf, dec.freq1)\n",
    "            koef = amp2/ amp1\n",
    "            print(koef, amp1, amp2, \"koef2\", \"amp2\")\n",
    "            if koef > threshold:\n",
    "                print(\"Signal decoded\")\n",
    "                sleep(1)\n",
    "                correct, response = True, True\n",
    "                break\n",
    "            if 1 / koef > threshold:\n",
    "                print(\"Signal decoded with error\")\n",
    "                correct, response = False, True\n",
    "                break\n",
    "    if not correct:\n",
    "        print(\"Playing the signal again\")\n",
    "        sleep(1)\n",
    "        response = False\n",
    "  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "os.close(sys.stderr.fileno())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "string = \"AAA\"\n",
    "[string[i] for i in range(len(string))]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def encode(text):\n",
    "    binary = ' '.join(format(ord(char), '08b') for char in text)\n",
    "    return binary\n",
    "\n",
    "def decode(binary):\n",
    "    text = ''.join(chr(int(char, 2)) for char in binary.split())\n",
    "    return text\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.6"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}