Changeset 6d0f203 in ammosreader


Ignore:
Timestamp:
06/30/22 16:18:57 (3 years ago)
Author:
Enrico Schwass <ennoausberlin@…>
Branches:
AmmosSource, guix
Children:
d62906b
Parents:
809cb16
Message:

apply changes from Franks session

Files:
10 edited

Legend:

Unmodified
Added
Removed

  • ammosreader/AbstractAmmosReader.py

    r809cb16 r6d0f203  
    88from ammosreader.AmmosContainer import AmmosContainer
    99from ammosreader.AmmosConstants import FrameType
     10
     11logging.basicConfig(filename='ammos.log', level=logging.DEBUG)
    1012
    1113
    … …  
    9698            return None
    9799
    98         logging.info("Current global frame header %s", current_global_frame_header)
     100        # logging.info("Current global frame header %s", current_global_frame_header)
    99101        return current_global_frame_header
    100102

  • ammosreader/AmmosAudioDataBody.py

    r809cb16 r6d0f203  
    1010        self.__samples_per_channel = samples_per_channel
    1111        self.__sample_size = sample_size
    12         self.__data = pcm_data
     12        self.__payload = pcm_data
    1313
    1414    @property
    15     def data(self):
     15    def payload(self):
    1616        """I return the raw pcm data with channels interweaved."""
    17         return self.__data
     17        return self.__payload
    1818
    1919    def pcm_for_channel(self, channel_number):
    … …  
    2121        start_offset = channel_number * self.__sample_size
    2222        step = self.__sample_size * self.__number_of_channels
    23         end = (len(self.__data) // step) * step
     23        end = (len(self.__payload) // step) * step
    2424        channel_bytes = b""
    2525        for each in range(start_offset, end, step):
    26             channel_bytes += self.__data[each:each+self.__sample_size]
     26            channel_bytes += self.__payload[each:each+self.__sample_size]
    2727        return channel_bytes
    2828
    … …  
    3232                "\nSamples per channel:" + str(self.__samples_per_channel) +
    3333                "\nSample size:" + str(self.__sample_size) +
    34                 "\nData size:" + str(len(self.__data)))
     34                "\nData size:" + str(len(self.__payload)))

  • ammosreader/AmmosContainer.py

    r809cb16 r6d0f203  
    3030
    3131    def frequencies(self):
    32         return set(list(filter(lambda frame: frame.global_frame_body.data_header.frequency, self.__global_frames)))
     32        """I return a list of unique frequencies inside this container."""
     33        return list({each.global_frame_body.data_header.frequency for each in self.__global_frames})
     34
     35    def frame_types(self):
     36        return list({each.global_frame_header.frame_type for each in self.__global_frames})
     37
     38    def frame_sizes(self):
     39        return list({each.global_frame_header.frame_length*4 for each in self.__global_frames})
     40
     41    def is_homogenic(self):
     42        return (len(self.frame_sizes()) == 1) and (len(self.frame_types()) == 1)
    3343
    3444    def __str__(self):

  • ammosreader/AmmosGlobalFrameBody.py

    r809cb16 r6d0f203  
    2929        self.__data_body = data_bytes
    3030
    31     def data_bytes_only(self):
    32 
    33         byte_string = b""
    34 
    35         for each_block in self.data_body:
    36             if not each_block:
    37                 print("Block is nil")
    38 
    39             byte_string += each_block.data
    40 
    41         return byte_string
     31    def payload(self):
     32        """I return the payload only."""
     33        return b"".join([each_block.data for each_block in self.data_body])

  • ammosreader/AmmosGlobalFrameHeader.py

    r809cb16 r6d0f203  
     1"""I provide an AMMOS global frame header."""
    12import struct
     3import logging
     4
     5logging.basicConfig(filename='ammos.log', level=logging.DEBUG)
    26
    37
    48class AmmosGlobalFrameHeader():
    5 
     9    """I implement an AMMOS global frame header."""
    610    MAGIC_WORD = "726574fb"
    711    HEADER_SIZE = 24
    … …  
    913    @classmethod
    1014    def from_bytes(cls, bytes):
     15        """I create a new AmmosGlobalFrameHeader from bytes."""
     16        assert len(bytes) == cls.HEADER_SIZE
    1117
    1218        elements = struct.unpack('<4s4s4s4s4s4s', bytes)
    1319
    1420        magic_word = elements[0].hex()
     21
     22        if magic_word != cls.MAGIC_WORD:
     23            return None
    1524
    1625        frame_length = (int.from_bytes(elements[1], byteorder='little')*4)
    … …  
    2837
    2938    def __init__(self, magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved):
    30 
     39        """I return a new instance of myself initialized with above parameters."""
    3140        if magic_word != type(self).MAGIC_WORD:
    32             print("Wrong magic word")
     41            logging.error("Wrong magic word found")
    3342            self.magic_word = magic_word
    3443        else:
    … …  
    4150
    4251    def __str__(self):
     52        """I return the string representation of myself."""
    4353        output = ("Global frame header info\n" +
    4454                  "------------------------\n" +
    … …  
    4959                  "Data header length:" + str(self.data_header_length) + "\n")
    5060        return output
    51 
    52     def size(self):
    53         return 24

  • ammosreader/AmmosIFDataBlock.py

    r809cb16 r6d0f203  
    1010
    1111    @property
    12     def data(self):
     12    def payload(self):
    1313        return self.__data

  • ammosreader/AmmosIFDataBody.py

    r809cb16 r6d0f203  
    2121
    2222    @property
    23     def data(self):
    24         return b"".join([each.data for each in self.data_blocks])
     23    def payload(self):
     24        return b"".join([each.payload for each in self.data_blocks])

  • ammosreader/AmmosIFReader.py

    r809cb16 r6d0f203  
    103103        return AmmosGlobalFrameBody(if_data_header, if_data_body)
    104104
    105     def data(self):
    106         return b"".join([each.global_frame_body.data_body.data for each in self.container.global_frames])
     105    def payload(self):
     106        return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])

  • sample_scripts/audio_reader.py

    r809cb16 r6d0f203  
    11import sys
    2 import os
    32import io
    43from pydub import AudioSegment
    54from pydub.playback import play
    6 
    7 sys.path.append('../src/')
    85
    96from ammosreader.AmmosAudioReader import AmmosAudioReader
    … …  
    2320    print("Sample rate:", dat_file.container.global_frames[0].global_frame_body.data_header.sample_rate)
    2421    print("Container size:", dat_file.container.size())
     22    print("Frequencies", dat_file.container.frequencies())
     23    print("Frame types:", dat_file.container.frame_types())
     24    print("Frame sizes:", dat_file.container.frame_sizes())
     25    print("Homogenic:", dat_file.container.is_homogenic())
    2526    pcm_data = dat_file.pcm_for_channel(0)
    2627    print("PCM data size total:", len(pcm_data))

  • sample_scripts/iqdw_reader.py

    r809cb16 r6d0f203  
    1818
    1919    dat_file.read_all_frames_left()
    20     print(len(dat_file.data()))
     20    print("Frequencies", dat_file.container.frequencies())
     21    print("Frame types:", dat_file.container.frame_types())
     22    print("Frame sizes:", dat_file.container.frame_sizes())
     23    print("Homogenic:", dat_file.container.is_homogenic())
Note: See TracChangeset for help on using the changeset viewer.