המלצה | 🚀 [פרויקט מהפכני] העברת קבצים למחשב ללא אינטרנט דרך שיחה קולית (טלפון כשר!) - הקוד המלא

[מתכנת חובב]

@אורי חשבון יפה אבל לא מדוייק (לדגום יותר מ 40 פעמים בשנייה זה כבר מתחיל להיות מסוכן לשלמות הנתונים אז כולל סיבית יתירות crc וכו' זה יוצא יותר משניה אם כי עדיין בטווח הסביר)
בכל אופן קח בחשבון שיש קווים בשתי חברות עם אותו מספר (אבל רק אחד מהם עוצר בכל תחנה אז אולי אפשר לסנן לפי זה - תלוי במאגר שיהיה)

[דוד יצחק]

כדי להפוך את הפרויקט הזה לשימושי באמת ביום-יום, אני חושב שכדאי לנו למקד את הדיון בשני כיוונים מרכזיים לשיפור:

התמקדות בדחיסת נתונים: להמשיך ולדחוף את יכולות הדחיסה לרמה הכי גבוהה שאפשר. ככל שהקובץ יהיה קטן יותר, ככה נצטרך פחות 'זמן אוויר' ופחות סיכוי לשגיאות.

מנגנון אישור (ACK) דרך מודל API: אם אנחנו רוצים לקחת את זה צעד קדימה ולחסוך את כל ה'ניפוי' (היתירות) שנדרש בגלל רעשים בשיחה קולית, הפתרון הוא יצירת אינטראקציה בין המחשב למערכת שזה אומר לפתח שלוחת API
הרעיון הוא בעצם לחסוך את שילוש הנתונים, וגם אם יש שגיאה אז מקסימום חוזרים על אותו בלוק בלבד ולא על כל הקובץ

וכנ"ל גם בקבלה
המערכת (IVR) תשלח את המידע בבלוקים.
המחשב שמקליט יבדוק כל בלוק בזמן אמת.
המחשב ידווח ל-API של המערכת אם הבלוק הגיע תקין או שגוי.
המערכת תמשיך לבלוק הבא רק אם התקבל אישור, ובמקרה של שגיאה – היא תשלח שוב רק את אותו בלוק ספציפי.

מה דעתכם?

[phpman]

@דוד-יצחק אני כתבתי על זה למעלה שיש אפשרות לעשות זאת ע״ שימוש עם מרכזיה שתבנה לבד עם אסטריסק, ואז את השולח בלוק של נתונים אחד אחרי השני והמחשב יענה עם DTMF ויאשר את הבלוקים ויבקש מחדש בלוקים שגויים כך אתה יכול לחסוך בשליחת הנתונים שוב ושוב ולהתמקד רק על הבלוקים הבעייתים

[כבוד הרב]

@דוד-יצחק אחד הדברים שאפשר לעשות כדי לדחוס מעבר לדחיסה רגילה זה להוריד את השימוש ב-UTF-8 כי זה מכפיל כל ביט.
השאלה היא אם אפשר לשלוח מהתוכנה DTMF לימות המשיח שימשיך להקלטה הבאה אם זה עבר טוב או שישמיע שוב אם זה לא עבר טוב.
את התפריט לאחר כל קובץ אפשר להגדיר עם ההקדרה הבאה במודול השמעת קבצים כמו שמופיע כאן:

ניתן להגדיר תפריט לאחר כל השמעה

after_play_tfr=tfr_more_options

ברירת מחדל של התפריט M1459 לשמיעה חוזרת הקישו 0 למעבר להשמעה הבאה הקישו 1
ניתן להגדיר את התפריט מחדש ראה כאן

י

[MNU 0]

אולי יותר טוב או יכול לעזור
https://github.com/romanz/amodem?utm_source=

[עניו כמשה]

@כבוד-הרב מדהים ביותר

[יוסף3]

@כבוד-הרב
מגניב ברמות!
מסוג הדברים שאני מצפה מהמוסד לעשות כדי להיכנס לשרתים של הגרעין האירני

[דוד יצחק]

@phpman כתב בהמלצה | [פרויקט מהפכני] העברת קבצים למחשב ללא אינטרנט דרך שיחה קולית (טלפון כשר!) - הקוד המלא:

@דוד-יצחק אני כתבתי על זה למעלה שיש אפשרות לעשות זאת ע״ שימוש עם מרכזיה שתבנה לבד עם אסטריסק, ואז את השולח בלוק של נתונים אחד אחרי השני והמחשב יענה עם DTMF ויאשר את הבלוקים ויבקש מחדש בלוקים שגויים כך אתה יכול לחסוך בשליחת הנתונים שוב ושוב ולהתמקד רק על הבלוקים הבעייתים

אוקיי אבל לא חייב ממש להקים מרכזיה אפשר גם עם שלוחת API אם כי יש לך פחות שליטה

[דוד יצחק]

בדקתי עוד כיון של הקטנת האורך של קובץ שמע וזה דווקא די הצליח
לקחתי קובץ המרתי אותו עם הסקריפט שיש בהודעה הראשונה והוא היה בעורך של 16 שעות
לקחתי את אותו קובץ דחסתי אותו בדחיסה מקסימלית עם 7ZIP (בהמלצת AI)והוא נהיה רק שעה וחצי

[מתכנת חובב]

@דוד-יצחק כל הבדיקות שעושים כאן לא מתחילות מהצד הטכני של האודיו ולכן אין הרבה עניין בזה
גם אם יהיה הישג כלשהו הוא ישבר במקרי קצה

[כבוד הרב]

מי שרוצה דוגמה חייה לאחת הגרסאות המתקדמות שלי זה הקישור:

https://drive.google.com/file/d/1WtkPXOtK0Dm0_yI*****Csmf7FbsBG8NLrkAb/view?usp=sharing

להוריד כוכביות ולא לפרסם ללא הכוכביות.
יש הקלטה מוכנה להשמעה בקו שמספרו: 0794946346.
יש להיכנס לתוכנה ללחוץ על "התחל האזנה" ואז לחייג לקו ולהיכנס לשלוחה 1 לעשות על רמקול לסתום אוזנים ואז תתחילו לראות את הטקסט שג'ימני יצר שהוא מסביר על התוכנה.
שימו לב שבקוד הזה הטקסט חולק למקטעים ולכן גם אם יתפספס קטע אחד אתם תראו את הקטע הבא אחרי כמה שניות.

הקוד בספוילר:

Spoiler

import sys
import os
import struct
import threading
import zlib
import base64
import time
import queue
import datetime
import numpy as np
import sounddevice as sd
import soundfile as sf
from scipy.io import wavfile
from scipy import signal as sig
from reedsolo import RSCodec, ReedSolomonError
import tkinter as tk
from tkinter import filedialog, messagebox, ttk, scrolledtext

# === הגדרות ליבה: STREAMER V3 ===
SAMPLE_RATE = 8000
BAUD_RATE = 50        
RS_ECC_SYMBOLS = 20   
INTERLEAVE_STEP = 10  
FREQS = [1000, 1400, 1800, 2200] # 4-FSK

# סנכרון
SYNC_DURATION = 0.4
t_sync = np.arange(int(SAMPLE_RATE * SYNC_DURATION)) / SAMPLE_RATE
SYNC_SIGNAL = sig.chirp(t_sync, f0=1000, f1=2200, t1=SYNC_DURATION, method='linear')

# גודל מנה (תווים)
CHUNK_SIZE = 120 

TEXT_EXTENSIONS = ['.txt', '.csv', '.json', '.xml', '.html', '.php', '.py', '.js', '.ini', '.log']

# === לוגיקה מתמטית (Modem Engine) ===
def interleave(bits):
    n = len(bits)
    pad = (INTERLEAVE_STEP - (n % INTERLEAVE_STEP)) % INTERLEAVE_STEP
    src = bits + [0]*pad
    dst = [0]*len(src)
    rows = len(src)//INTERLEAVE_STEP
    for i in range(len(src)):
        dst[(i%INTERLEAVE_STEP)*rows + (i//INTERLEAVE_STEP)] = src[i]
    return dst, pad

def deinterleave(bits, pad):
    n = len(bits)
    rows = n//INTERLEAVE_STEP
    dst = [0]*n
    for i in range(n):
        col = i // rows
        row = i % rows
        dst[row*INTERLEAVE_STEP + col] = bits[i]
    return dst[:-pad] if pad else dst

def text2bits(b):
    res = []
    for x in b:
        for i in range(7,-1,-1): res.append((x>>i)&1)
    return res

def bits2bytes(b):
    res = bytearray()
    for i in range(0, len(b), 8):
        byte = b[i:i+8]
        if len(byte)<8: break
        v = 0
        for bit in byte: v = (v<<1)|bit
        res.append(v)
    return bytes(res)

# === Encoder ===
def encode_packet(data_bytes, packet_id, total_packets):
    try:
        compressed = zlib.compress(data_bytes, level=9)
        header = struct.pack('>HHH', packet_id, total_packets, len(compressed))
        payload = header + compressed
        
        rsc = RSCodec(RS_ECC_SYMBOLS)
        encoded = rsc.encode(payload)
        
        bits = text2bits(encoded)
        shuffled, pad = interleave(bits)
        
        meta_header = struct.pack('>BH', pad, len(shuffled))
        meta_bits = text2bits(meta_header)
        
        final_bits = meta_bits + shuffled
        if len(final_bits) % 2 != 0: final_bits.append(0)
        
        sps = int(SAMPLE_RATE / BAUD_RATE)
        t_sym = np.arange(sps) / SAMPLE_RATE
        phase = 0
        waves = []
        
        for i in range(0, len(final_bits), 2):
            val = (final_bits[i] << 1) | final_bits[i+1]
            freq = FREQS[val]
            w = np.sin(2 * np.pi * freq * t_sym + phase)
            waves.append(w)
            phase += 2 * np.pi * freq * (sps / SAMPLE_RATE)
            phase %= 2 * np.pi
            
        silence = np.zeros(int(SAMPLE_RATE * 0.25)) 
        return np.concatenate([SYNC_SIGNAL, np.concatenate(waves), silence])
    except: return None

def generate_stream_file(filename, file_content, output_wav):
    try:
        ext = os.path.splitext(filename)[1].lower()
        is_text = ext in TEXT_EXTENSIONS
        chunks = []
        
        if is_text:
            try:
                text_str = file_content.decode('utf-8', errors='ignore')
                curr = ""
                words = text_str.replace('\n', ' \n ').split(' ') 
                for word in words:
                    if len(word.encode('utf-8')) > CHUNK_SIZE:
                        if curr: chunks.append(curr.encode('utf-8')); curr = ""
                        chunks.append(word.encode('utf-8')); continue
                    if len(curr.encode('utf-8')) + len(word.encode('utf-8')) < CHUNK_SIZE:
                        curr += word + " "
                    else:
                        chunks.append(curr.encode('utf-8')); curr = word + " "
                if curr: chunks.append(curr.encode('utf-8'))
            except:
                chunks = [file_content[i:i+CHUNK_SIZE] for i in range(0, len(file_content), CHUNK_SIZE)]
        else:
            chunks = [file_content[i:i+CHUNK_SIZE] for i in range(0, len(file_content), CHUNK_SIZE)]
            
        total = len(chunks)
        audio_parts = []
        t_wake = np.arange(int(SAMPLE_RATE * 2.0)) / SAMPLE_RATE
        wake = np.sin(2 * np.pi * 2200 * t_wake)
        audio_parts.append(wake)
        
        for i, chunk in enumerate(chunks):
            pkt = encode_packet(chunk, i+1, total)
            if pkt is not None: audio_parts.append(pkt)
                
        final_float = np.concatenate(audio_parts)
        mx = np.max(np.abs(final_float))
        final_int16 = (final_float / mx * 32767).astype(np.int16) if mx > 0 else final_float.astype(np.int16)
            
        wavfile.write(output_wav, SAMPLE_RATE, final_int16)
        return True, f"{total}"
    except Exception as e: return False, str(e)

# === Decoder ===
def decode_packet_from_stream(clean_audio):
    try:
        sps = int(SAMPLE_RATE / BAUD_RATE)
        num_syms = len(clean_audio) // sps
        if num_syms < 50: return None, 0
        
        t = np.arange(sps) / SAMPLE_RATE
        refs = [np.exp(-2j * np.pi * f * t) for f in FREQS]
        
        bits_str = ""
        for i in range(num_syms):
            chunk = clean_audio[i*sps : (i+1)*sps]
            energies = [np.abs(np.dot(chunk, r))**2 for r in refs]
            bits_str += str((np.argmax(energies)>>1)&1) + str(np.argmax(energies)&1)
            
        if len(bits_str) < 24: return None, 0
        meta_bits = [int(b) for b in bits_str[:24]]
        pad, bit_len = struct.unpack('>BH', bits2bytes(meta_bits))
        
        if bit_len > 10000 or bit_len == 0: return None, 0
        if len(bits_str) < 24 + bit_len: return None, 0 
        
        payload_bits = [int(b) for b in bits_str[24 : 24+bit_len]]
        raw_bits = deinterleave(payload_bits, pad)
        packet_bytes = bits2bytes(raw_bits)
        
        decoded = RSCodec(RS_ECC_SYMBOLS).decode(packet_bytes)[0]
        
        pkt_id, pkt_total, content_len = struct.unpack('>HHH', decoded[:6])
        content = zlib.decompress(decoded[6 : 6+content_len])
        samples_consumed = ((24 + bit_len) // 2) * sps
        
        return (pkt_id, pkt_total, content), samples_consumed
    except: return None, 0

# === GUI Professional ===
class AcousticReader:
    def __init__(self, root):
        self.root = root
        self.root.title("Acoustic Reader Pro")
        self.root.geometry("600x700")
        self.root.configure(bg="#2C3E50")
        
        self.audio_buffer = np.array([], dtype=float)
        self.is_rec = False
        self.q = queue.Queue()
        self.received_packets = {} 
        
        # Styles
        style = ttk.Style()
        style.theme_use('clam')
        style.configure("TNotebook", background="#2C3E50", borderwidth=0)
        style.configure("TNotebook.Tab", background="#34495E", foreground="white", padding=[20, 10], font=("Segoe UI", 11))
        style.map("TNotebook.Tab", background=[("selected", "#3498DB")])
        style.configure("TFrame", background="#ECF0F1")
        
        tabs = ttk.Notebook(root)
        t_decode = ttk.Frame(tabs) # קבלה ראשון
        t_encode = ttk.Frame(tabs) # יצירה שני
        
        tabs.add(t_decode, text=" 📥 קליטת נתונים (חיבור AUX) ")
        tabs.add(t_encode, text=" 📤 יצירת קובץ (למנהל) ")
        tabs.pack(fill="both", expand=True, padx=10, pady=10)
        
        self.setup_decoder(t_decode)
        self.setup_encoder(t_encode)
        
        self.root.after(100, self.process_queue)

    def setup_decoder(self, parent):
        frame = tk.Frame(parent, bg="#ECF0F1")
        frame.pack(fill="both", expand=True, padx=20, pady=20)
        
        # Header
        tk.Label(frame, text="מערכת לקליטת עלונים וקבצים", font=("Segoe UI", 18, "bold"), bg="#ECF0F1", fg="#2C3E50").pack(pady=10)
        
        # Status & Vol
        status_frame = tk.Frame(frame, bg="#BDC3C7", bd=1, relief="sunken")
        status_frame.pack(fill="x", pady=10)
        
        self.canvas_vol = tk.Canvas(status_frame, width=300, height=15, bg="#95a5a6", highlightthickness=0)
        self.canvas_vol.pack(side="left", padx=10, pady=5)
        self.vol_bar = self.canvas_vol.create_rectangle(0,0,0,15, fill="#2ECC71")
        
        self.lbl_status = tk.Label(status_frame, text="ממתין להפעלה...", font=("Segoe UI", 10), bg="#BDC3C7")
        self.lbl_status.pack(side="right", padx=10)

        # Main Button
        self.btn_rec = tk.Button(frame, text="התחל האזנה", command=self.toggle_rec, 
                                 font=("Segoe UI", 16, "bold"), bg="#27AE60", fg="white", 
                                 activebackground="#2ECC71", activeforeground="white", bd=0, padx=20, pady=10)
        self.btn_rec.pack(pady=10)
        
        # Text Area
        tk.Label(frame, text="תוכן הקובץ המתקבל:", font=("Segoe UI", 11), bg="#ECF0F1", anchor="w").pack(fill="x")
        self.txt_display = scrolledtext.ScrolledText(frame, height=15, font=("Segoe UI", 12), bd=2, relief="flat")
        self.txt_display.pack(fill="both", expand=True, pady=5)
        self.txt_display.tag_config("info", foreground="#E67E22", font=("Segoe UI", 10, "bold"))
        self.txt_display.tag_config("content", foreground="#2C3E50")
        
        # Footer buttons
        btn_frame = tk.Frame(frame, bg="#ECF0F1")
        btn_frame.pack(pady=5)
        tk.Button(btn_frame, text="נקה מסך", command=self.clear_screen, bg="#95a5a6", fg="white", bd=0, padx=10).pack(side="left", padx=5)
        tk.Button(btn_frame, text="העתק טקסט", command=self.copy_text, bg="#3498DB", fg="white", bd=0, padx=10).pack(side="left", padx=5)

    def setup_encoder(self, parent):
        frame = tk.Frame(parent, bg="#ECF0F1")
        frame.pack(fill="both", expand=True, padx=20, pady=20)
        
        tk.Label(frame, text="יצירת קובץ שמע למערכת הטלפונית", font=("Segoe UI", 16, "bold"), bg="#ECF0F1", fg="#2C3E50").pack(pady=20)
        tk.Label(frame, text="בחר קובץ טקסט/מסמך להמרה.\nהמערכת תיצור קובץ WAV מותאם לשידור.", font=("Segoe UI", 11), bg="#ECF0F1", fg="#7F8C8D").pack(pady=10)
        
        tk.Button(frame, text="בחר קובץ והמר", command=self.do_enc, 
                  font=("Segoe UI", 14), bg="#2980B9", fg="white", 
                  activebackground="#3498DB", activeforeground="white", bd=0, padx=20, pady=15).pack(pady=40)
        
        self.lbl_enc = tk.Label(frame, text="", font=("Segoe UI", 11, "bold"), bg="#ECF0F1", fg="#27AE60")
        self.lbl_enc.pack()

    def clear_screen(self):
        self.txt_display.delete('1.0', tk.END)
        self.received_packets = {}

    def copy_text(self):
        self.root.clipboard_clear()
        self.root.clipboard_append(self.txt_display.get("1.0", tk.END))
        messagebox.showinfo("הועתק", "התוכן הועתק ללוח")

    def do_enc(self):
        fn = filedialog.askopenfilename()
        if not fn: return
        sn = filedialog.asksaveasfilename(defaultextension=".wav", filetypes=[("WAV","*.wav")])
        if not sn: return
        
        self.lbl_enc.config(text="מעבד... אנא המתן", fg="blue")
        self.root.update()
        try:
            with open(fn,'rb') as f: c=f.read()
            ok, msg = generate_stream_file(fn, c, sn)
            if ok: self.lbl_enc.config(text=f"הצלחה! נוצר קובץ עם {msg} מנות.", fg="#27AE60")
            else: self.lbl_enc.config(text=f"שגיאה: {msg}", fg="#C0392B")
        except Exception as e: self.lbl_enc.config(text=f"שגיאה: {e}", fg="#C0392B")

    def toggle_rec(self):
        if not self.is_rec:
            self.is_rec = True
            self.btn_rec.config(text="⏹️ עצור הקשבה", bg="#C0392B", activebackground="#E74C3C")
            self.audio_buffer = np.array([], dtype=float)
            self.lbl_status.config(text="מקשיב לקו... (נא להשמיע)", fg="#2980B9")
            self.clear_screen()
            threading.Thread(target=self.rec_thread).start()
        else:
            self.is_rec = False
            self.btn_rec.config(text="התחל האזנה", bg="#27AE60", activebackground="#2ECC71")
            self.lbl_status.config(text="מוכן")

    def update_vol(self, indata):
        peak = np.max(np.abs(indata)) / 32768.0
        w = min(300, int(peak * 300))
        c = "#2ECC71" if w < 200 else "#E74C3C"
        self.canvas_vol.coords(self.vol_bar, 0, 0, w, 15)
        self.canvas_vol.itemconfig(self.vol_bar, fill=c)

    def rec_thread(self):
        def cb(i, f, t, s): self.q.put(i.copy())
        # שימוש בברירת מחדל של המערכת (ללא device index)
        try:
            with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, callback=cb):
                while self.is_rec: sd.sleep(100)
        except Exception as e:
            self.is_rec = False
            messagebox.showerror("Error", f"שגיאת מיקרופון: {e}")

    def process_queue(self):
        try:
            while not self.q.empty():
                chunk = self.q.get_nowait().flatten()
                self.update_vol(chunk * 32767)
                self.audio_buffer = np.concatenate((self.audio_buffer, chunk))
            
            if len(self.audio_buffer) > SAMPLE_RATE * 15:
                self.audio_buffer = self.audio_buffer[-SAMPLE_RATE*15:]
            self.try_decode()
        except: pass
        self.root.after(50, self.process_queue)

    def try_decode(self):
        if len(self.audio_buffer) < 4000: return
        buff = self.audio_buffer.copy()
        
        sos = sig.butter(4, [800, 2400], btype='bandpass', fs=SAMPLE_RATE, output='sos')
        clean = sig.sosfilt(sos, buff)
        mx = np.max(np.abs(clean))
        if mx > 0.01: clean /= mx
        
        corr = sig.correlate(clean, SYNC_SIGNAL, mode='valid')
        peaks, _ = sig.find_peaks(np.abs(corr), height=5, distance=SAMPLE_RATE)
        
        for peak_idx in peaks:
            start_data = peak_idx + len(SYNC_SIGNAL)
            if len(clean) > start_data + 500:
                audio_chunk = clean[start_data:]
                result, consumed = decode_packet_from_stream(audio_chunk)
                if result:
                    pkt_id, total, content = result
                    if pkt_id not in self.received_packets:
                        self.received_packets[pkt_id] = True
                        try:
                            text = content.decode('utf-8')
                            self.append_text(f"--- חלק {pkt_id} מתוך {total} ---\n", "info")
                            self.append_text(text + "\n", "content")
                        except:
                            self.append_text(f"--- חלק {pkt_id}/{total} (קובץ בינארי) ---\n", "info")
                    
                    cut_pos = start_data + consumed + 100
                    if cut_pos < len(self.audio_buffer): self.audio_buffer = self.audio_buffer[cut_pos:]
                    else: self.audio_buffer = np.array([], dtype=float)
                    break 

    def append_text(self, text, tag):
        self.txt_display.insert(tk.END, text, tag)
        self.txt_display.see(tk.END)

if __name__ == "__main__":
    root = tk.Tk()
    app = AcousticReader(root)
    root.mainloop()

[דוד יצחק]

@כבוד-הרב כדאי שפרט מה חדש בגרסה הזו

[כבוד הרב]

@דוד-יצחק תפתח ותראה זה שונה לגמרי וכל הלוגיקה שונתה והיא תואמת בעיקר לטקסט.

[smct]

@כבוד-הרב
לי לא נפתח התוכנה עכשיו נפתח לי

[כבוד הרב]

@smct מוזר תנסה שוב
או שאולי המחשב שלך איטי מדאי

[דוד יצחק]

@כבוד-הרב
אוקיי אני רואה ששונה הממשק
אבל אני רואה שהוא לוקח הרבה פחות אורך אודיו
מה שונה בלוגיקה?

[מתכנת חובב]

@דוד-יצחק לא בדקתי את הלוגיקה אבל הגיוני שאם זה עכשיו מותאם לטקסט אז במקום רצפים של 8 ביטים הוא יכול להסתפק ב 5 או כמה שהוא בחר לייצג

[דוד יצחק]

@כבוד-הרב זה בעצם גם אומר שזה מתאים רק לטקסט ולא מתאים לקבצים?

[כבוד הרב]

@מתכנת-חובב עכשיו שזה לטקסט זה המיר לי 2,697 תווים ל-7:10 דקות!

[מתכנת חובב]

@כבוד-הרב זה קצת דיבורים באוויר
כמה ביטים לשניה?