can extract audio from kibana share method also

examples/json_to_audio.py

@@ -8,32 +8,60 @@
 import json
 import gzip
 import argparse
+import csv
+import sys
+
+
+def encoded_audio_to_ogg_file(config, encoded_audio, ogg_file_path):
+    encrypted_bytes = base64.b64decode(encoded_audio)
+    key = RSA.importKey(open(config.ssh_file).read(),
+                        passphrase=config.ssh_password)
+    cipher = PKCS1_OAEP.new(key)
+
+    decrypted_header = cipher.decrypt(
+        encrypted_bytes[:key.size_in_bytes()])
+
+    aes_key = decrypted_header[:AES.block_size]
+    iv = decrypted_header[AES.block_size:]
+
+    aes_cipher = AES.new(aes_key, AES.MODE_CBC,
+                         iv)
+    decrypted_audio = aes_cipher.decrypt(
+        encrypted_bytes[key.size_in_bytes():])
+    if "Ogg" == decrypted_audio[:3].decode():
+        print("Ogg decrypted")
+    else:
+        raise Exception("Audio not starting with Ogg")
+    with open(ogg_file_path, "wb") as out_fp:
+        out_fp.write(decrypted_audio)
 
 
 def get_sf(config):
     with gzip.open(config.document_gz, 'rt') as fp:
         document = json.load(fp)
-        encrypted_bytes = base64.b64decode(document["encrypted_audio"])
-        key = RSA.importKey(open(config.ssh_file).read(),
-                            passphrase=config.ssh_password)
-        cipher = PKCS1_OAEP.new(key)
+        encoded_audio_to_ogg_file(config, document["encrypted_audio"],
+                                  config.ogg_file)
+
 
-        decrypted_header = cipher.decrypt(
-            encrypted_bytes[:key.size_in_bytes()])
+def sanitize_file_name(s):
+    return "".join([c for c in s.replace(":", "_") if c.isalpha() or c.isdigit() or
+                    c in (' ', '_')]).rstrip()
 
-        aes_key = decrypted_header[:AES.block_size]
-        iv = decrypted_header[AES.block_size:]
 
-        aes_cipher = AES.new(aes_key, AES.MODE_CBC,
-                             iv)
-        decrypted_audio = aes_cipher.decrypt(
-            encrypted_bytes[key.size_in_bytes():])
-        if "Ogg" == decrypted_audio[:3].decode():
-            print("Ogg decrypted")
-        else:
-            raise Exception("Audio not starting with Ogg")
-        with open(config.ogg_file, "wb") as out_fp:
-            out_fp.write(decrypted_audio)
+def get_sf_from_csv(config, csv_file):
+    with open(csv_file, "r") as fp:
+        csv.field_size_limit(sys.maxsize)
+        reader = csv.reader(fp)
+        columns = next(reader)
+        date_column = columns.index("date")
+        encrypted_audio_column = columns.index("encrypted_audio")
+        for data_columns in reader:
+            encrypted_audio = data_columns[encrypted_audio_column]
+            if len(encrypted_audio) > 16:
+                ogg_path = os.path.join(os.path.dirname(csv_file),sanitize_file_name(
+                    data_columns[date_column]) + ".ogg")
+                config.ogg_file = ogg_path
+                encoded_audio_to_ogg_file(config, encrypted_audio, ogg_path)
 
 
 def main():
@@ -61,7 +89,11 @@ def main():
             args_cp.ogg_file = os.path.join(args.document_gz, document[:document.rfind(".json.gz")]+".ogg")
             get_sf(args_cp)
     else:
-        get_sf(args)
+        if args.document_gz.endswith(".json.gz"):
+            get_sf(args)
+        elif args.document_gz.endswith(".csv"):
+            # kibana share method
+            get_sf_from_csv(args, args.document_gz)
 
 
 if __name__ == "__main__":

src/noisesensor/digitalfilter.py

@@ -3,10 +3,7 @@
 from numba import float64, int32     # import the types
 import math
 
-""" A digital biquad filter is a second order recursive linear filter,
- containing two poles and two zeros. "Biquad" is an abbreviation of "bi-quadratic",
-  which refers to the fact that in the Z domain, its transfer function is the ratio of two quadratic functions
-    
+"""
 BSD 3-Clause License
 
 Copyright (c) 2023, University Gustave Eiffel

tests/test_biquad_filter.py

@@ -171,9 +171,9 @@ def test_leq_speak(self):
             raise e
 
     def test_leq_speak_g2(self):
-        fd = FilterDesign(sample_rate=32000,
-                          first_frequency_band=20,
-                          last_frequency_band=12500)
+        fd = FilterDesign(sample_rate=44100,
+                          first_frequency_band=100,
+                          last_frequency_band=16000)
         expected_leqs = []
         fd.down_sampling = fd.G2
         fd.band_division = 3
@@ -182,14 +182,14 @@ def test_leq_speak_g2(self):
                        configuration["bandpass"]]
         sc = SpectrumChannel(configuration)
         try:
-            with open(os.path.join(UNIT_TEST_PATH,
-                                   "speak_32000Hz_16bitsPCM_10s.raw"),
-                      "rb") as f:
+            with open("/home/fortin/github/NoiseCapture/sosfilter/src/test/resources/org/orbisgis/sos/speak_44100Hz_16bitsPCM_10s.raw","rb") as f:
                 if f is not None:
                     samples = numpy.frombuffer(f.read(), numpy.short)
                     samples = samples / 2**15
                     # process with acoustics library
-                    s = acoustics.Signal(samples, 32000)
+                    s = acoustics.Signal(samples, 44100)
+
+                    print(acoustics.standards.iso_tr_25417_2007.equivalent_sound_pressure_level(s.weigh(weighting='C'), 1.0))
                     frequencies, filtered_signals = s.third_octaves(
                         frequencies=center_frequency)
                     for freq, filtered_signal in list(