Index: itignore =================================================================== --- .gitignore (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,2 +1,0 @@ -*__pycache__/ -*build/ Index: itlab-ci.yml =================================================================== --- .gitlab-ci.yml (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,34 +1,0 @@ -image: python:latest - -workflow: - rules: - - if: $CI_COMMIT_BRANCH - -pages: - stage: deploy - script: - - python3 -m venv .venv - - source .venv/bin/activate - - pip install --upgrade numpy pdoc3 bitstring - - pdoc --html ammosreader -o public - - mv public/ammosreader public/doc - artifacts: - paths: - - public/doc - rules: - - if: $CI_COMMIT_BRANCH == "guix" - -pytest: - stage: test - script: - - python3 -m venv .venv - - source .venv/bin/activate - - pip install --upgrade numpy pdoc3 pytest bitstring - - cd tests - - pytest --junitxml=report.xml - artifacts: - when: always - reports: - junit: tests/report.xml - rules: - - if: $CI_COMMIT_BRANCH == "guix" Index: CENSE =================================================================== --- LICENSE (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,202 +1,0 @@ - - Apache License - Version 2.0, January 2004 - http://www.apache.org/licenses/ - - TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION - - 1. Definitions. - - "License" shall mean the terms and conditions for use, reproduction, - and distribution as defined by Sections 1 through 9 of this document. - - "Licensor" shall mean the copyright owner or entity authorized by - the copyright owner that is granting the License. - - "Legal Entity" shall mean the union of the acting entity and all - other entities that control, are controlled by, or are under common - control with that entity. For the purposes of this definition, - "control" means (i) the power, direct or indirect, to cause the - direction or management of such entity, whether by contract or - otherwise, or (ii) ownership of fifty percent (50%) or more of the - outstanding shares, or (iii) beneficial ownership of such entity. - - "You" (or "Your") shall mean an individual or Legal Entity - exercising permissions granted by this License. - - "Source" form shall mean the preferred form for making modifications, - including but not limited to software source code, documentation - source, and configuration files. - - "Object" form shall mean any form resulting from mechanical - transformation or translation of a Source form, including but - not limited to compiled object code, generated documentation, - and conversions to other media types. - - "Work" shall mean the work of authorship, whether in Source or - Object form, made available under the License, as indicated by a - copyright notice that is included in or attached to the work - (an example is provided in the Appendix below). - - "Derivative Works" shall mean any work, whether in Source or Object - form, that is based on (or derived from) the Work and for which the - editorial revisions, annotations, elaborations, or other modifications - represent, as a whole, an original work of authorship. For the purposes - of this License, Derivative Works shall not include works that remain - separable from, or merely link (or bind by name) to the interfaces of, - the Work and Derivative Works thereof. - - "Contribution" shall mean any work of authorship, including - the original version of the Work and any modifications or additions - to that Work or Derivative Works thereof, that is intentionally - submitted to Licensor for inclusion in the Work by the copyright owner - or by an individual or Legal Entity authorized to submit on behalf of - the copyright owner. For the purposes of this definition, "submitted" - means any form of electronic, verbal, or written communication sent - to the Licensor or its representatives, including but not limited to - communication on electronic mailing lists, source code control systems, - and issue tracking systems that are managed by, or on behalf of, the - Licensor for the purpose of discussing and improving the Work, but - excluding communication that is conspicuously marked or otherwise - designated in writing by the copyright owner as "Not a Contribution." - - "Contributor" shall mean Licensor and any individual or Legal Entity - on behalf of whom a Contribution has been received by Licensor and - subsequently incorporated within the Work. - - 2. Grant of Copyright License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - copyright license to reproduce, prepare Derivative Works of, - publicly display, publicly perform, sublicense, and distribute the - Work and such Derivative Works in Source or Object form. - - 3. Grant of Patent License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - (except as stated in this section) patent license to make, have made, - use, offer to sell, sell, import, and otherwise transfer the Work, - where such license applies only to those patent claims licensable - by such Contributor that are necessarily infringed by their - Contribution(s) alone or by combination of their Contribution(s) - with the Work to which such Contribution(s) was submitted. If You - institute patent litigation against any entity (including a - cross-claim or counterclaim in a lawsuit) alleging that the Work - or a Contribution incorporated within the Work constitutes direct - or contributory patent infringement, then any patent licenses - granted to You under this License for that Work shall terminate - as of the date such litigation is filed. - - 4. Redistribution. You may reproduce and distribute copies of the - Work or Derivative Works thereof in any medium, with or without - modifications, and in Source or Object form, provided that You - meet the following conditions: - - (a) You must give any other recipients of the Work or - Derivative Works a copy of this License; and - - (b) You must cause any modified files to carry prominent notices - stating that You changed the files; and - - (c) You must retain, in the Source form of any Derivative Works - that You distribute, all copyright, patent, trademark, and - attribution notices from the Source form of the Work, - excluding those notices that do not pertain to any part of - the Derivative Works; and - - (d) If the Work includes a "NOTICE" text file as part of its - distribution, then any Derivative Works that You distribute must - include a readable copy of the attribution notices contained - within such NOTICE file, excluding those notices that do not - pertain to any part of the Derivative Works, in at least one - of the following places: within a NOTICE text file distributed - as part of the Derivative Works; within the Source form or - documentation, if provided along with the Derivative Works; or, - within a display generated by the Derivative Works, if and - wherever such third-party notices normally appear. The contents - of the NOTICE file are for informational purposes only and - do not modify the License. You may add Your own attribution - notices within Derivative Works that You distribute, alongside - or as an addendum to the NOTICE text from the Work, provided - that such additional attribution notices cannot be construed - as modifying the License. - - You may add Your own copyright statement to Your modifications and - may provide additional or different license terms and conditions - for use, reproduction, or distribution of Your modifications, or - for any such Derivative Works as a whole, provided Your use, - reproduction, and distribution of the Work otherwise complies with - the conditions stated in this License. - - 5. Submission of Contributions. Unless You explicitly state otherwise, - any Contribution intentionally submitted for inclusion in the Work - by You to the Licensor shall be under the terms and conditions of - this License, without any additional terms or conditions. - Notwithstanding the above, nothing herein shall supersede or modify - the terms of any separate license agreement you may have executed - with Licensor regarding such Contributions. - - 6. Trademarks. This License does not grant permission to use the trade - names, trademarks, service marks, or product names of the Licensor, - except as required for reasonable and customary use in describing the - origin of the Work and reproducing the content of the NOTICE file. - - 7. Disclaimer of Warranty. Unless required by applicable law or - agreed to in writing, Licensor provides the Work (and each - Contributor provides its Contributions) on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - implied, including, without limitation, any warranties or conditions - of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A - PARTICULAR PURPOSE. You are solely responsible for determining the - appropriateness of using or redistributing the Work and assume any - risks associated with Your exercise of permissions under this License. - - 8. Limitation of Liability. In no event and under no legal theory, - whether in tort (including negligence), contract, or otherwise, - unless required by applicable law (such as deliberate and grossly - negligent acts) or agreed to in writing, shall any Contributor be - liable to You for damages, including any direct, indirect, special, - incidental, or consequential damages of any character arising as a - result of this License or out of the use or inability to use the - Work (including but not limited to damages for loss of goodwill, - work stoppage, computer failure or malfunction, or any and all - other commercial damages or losses), even if such Contributor - has been advised of the possibility of such damages. - - 9. Accepting Warranty or Additional Liability. While redistributing - the Work or Derivative Works thereof, You may choose to offer, - and charge a fee for, acceptance of support, warranty, indemnity, - or other liability obligations and/or rights consistent with this - License. However, in accepting such obligations, You may act only - on Your own behalf and on Your sole responsibility, not on behalf - of any other Contributor, and only if You agree to indemnify, - defend, and hold each Contributor harmless for any liability - incurred by, or claims asserted against, such Contributor by reason - of your accepting any such warranty or additional liability. - - END OF TERMS AND CONDITIONS - - APPENDIX: How to apply the Apache License to your work. - - To apply the Apache License to your work, attach the following - boilerplate notice, with the fields enclosed by brackets "[]" - replaced with your own identifying information. (Don't include - the brackets!) The text should be enclosed in the appropriate - comment syntax for the file format. We also recommend that a - file or class name and description of purpose be included on the - same "printed page" as the copyright notice for easier - identification within third-party archives. - - Copyright [2022] [ElokaBtl 912] - - Licensed under the Apache License, Version 2.0 (the "License"); - you may not use this file except in compliance with the License. - You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, software - distributed under the License is distributed on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - See the License for the specific language governing permissions and - limitations under the License. Index: README.md =================================================================== --- README.md (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ README.md (revision 571e0cc6efc01eedb038d669ad8f41e03a20ed04) @@ -1,78 +1,92 @@ # AmmosReader -Project containing necessary File- or SocketStream-Reader to read the -various datastreams of Ammos +Project containing necessary File- or SocketStream-Reader to read the various datastreams of Ammos -# Installation +## Getting started -## Standard pip install (not tested on all platforms) +To make it easy for you to get started with GitLab, here's a list of recommended next steps. -``` shell -git clone http://gitlab.kid.local/kidzg/ammosreader.git -cd ammosreader -python3 -m pip install . +Already a pro? Just edit this README.md and make it your own. Want to make it easy? [Use the template at the bottom](#editing-this-readme)! + +## Add your files + +- [ ] [Create](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#create-a-file) or [upload](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#upload-a-file) files +- [ ] [Add files using the command line](https://docs.gitlab.com/ee/gitlab-basics/add-file.html#add-a-file-using-the-command-line) or push an existing Git repository with the following command: + +``` +cd existing_repo +git remote add origin http://gitlab.kid.local/kidzg/ammosreader.git +git branch -M main +git push -uf origin main ``` -## Using the guix functional package manager +## Integrate with your tools -First subscribe to the GUIX oai channel (see Chapter Add channel to -system) in -) +- [ ] [Set up project integrations](http://gitlab.kid.local/kidzg/ammosreader/-/settings/integrations) -Then install using the guix +## Collaborate with your team -``` shell -guix package -i python-ammosreader -``` +- [ ] [Invite team members and collaborators](https://docs.gitlab.com/ee/user/project/members/) +- [ ] [Create a new merge request](https://docs.gitlab.com/ee/user/project/merge_requests/creating_merge_requests.html) +- [ ] [Automatically close issues from merge requests](https://docs.gitlab.com/ee/user/project/issues/managing_issues.html#closing-issues-automatically) +- [ ] [Enable merge request approvals](https://docs.gitlab.com/ee/user/project/merge_requests/approvals/) +- [ ] [Automatically merge when pipeline succeeds](https://docs.gitlab.com/ee/user/project/merge_requests/merge_when_pipeline_succeeds.html) -# Usage +## Test and Deploy -There are two scripts, which demonstrate the usage of the ammosreader -package. +Use the built-in continuous integration in GitLab. -## iqdw_reader.py +- [ ] [Get started with GitLab CI/CD](https://docs.gitlab.com/ee/ci/quick_start/index.html) +- [ ] [Analyze your code for known vulnerabilities with Static Application Security Testing(SAST)](https://docs.gitlab.com/ee/user/application_security/sast/) +- [ ] [Deploy to Kubernetes, Amazon EC2, or Amazon ECS using Auto Deploy](https://docs.gitlab.com/ee/topics/autodevops/requirements.html) +- [ ] [Use pull-based deployments for improved Kubernetes management](https://docs.gitlab.com/ee/user/clusters/agent/) +- [ ] [Set up protected environments](https://docs.gitlab.com/ee/ci/environments/protected_environments.html) -``` shell -python3 iqdw_reader.py 'path_to_iqdw_file' -``` +*** -## pdw_reader.py +# Editing this README -``` shell -python3 pdw_reader.py 'path_to_pdw_file' -``` +When you're ready to make this README your own, just edit this file and use the handy template below (or feel free to structure it however you want - this is just a starting point!). Thank you to [makeareadme.com](https://www.makeareadme.com/) for this template. -To minimize depencencies of the core package other scripts -(audio_reader, ammos_viewer) are now stored in separate packages. +## Suggestions for a good README +Every project is different, so consider which of these sections apply to yours. The sections used in the template are suggestions for most open source projects. Also keep in mind that while a README can be too long and detailed, too long is better than too short. If you think your README is too long, consider utilizing another form of documentation rather than cutting out information. -# Logging +## Name +Choose a self-explaining name for your project. -If you use ammosreader in your own scripts a logger can be imported and -used as follows: +## Description +Let people know what your project can do specifically. Provide context and add a link to any reference visitors might be unfamiliar with. A list of Features or a Background subsection can also be added here. If there are alternatives to your project, this is a good place to list differentiating factors. -``` python -from ammosreader import logger -logger.warning("Put your warning here") -``` +## Badges +On some READMEs, you may see small images that convey metadata, such as whether or not all the tests are passing for the project. You can use Shields to add some to your README. Many services also have instructions for adding a badge. -## Log-File +## Visuals +Depending on what you are making, it can be a good idea to include screenshots or even a video (you'll frequently see GIFs rather than actual videos). Tools like ttygif can help, but check out Asciinema for a more sophisticated method. -The name of the logfile is ammos.log The logger uses the directory -stored in the AMMOS_LOG_DIR environment variable. If this variable is -not set, /tmp is used as default. This might change to *var/log* in -future versions running under GUIX system. If the logfile is not -writable, the standard logger config is used and the logs go directly to -console (tty). (see ammos_logging.conf file in this repository) +## Installation +Within a particular ecosystem, there may be a common way of installing things, such as using Yarn, NuGet, or Homebrew. However, consider the possibility that whoever is reading your README is a novice and would like more guidance. Listing specific steps helps remove ambiguity and gets people to using your project as quickly as possible. If it only runs in a specific context like a particular programming language version or operating system or has dependencies that have to be installed manually, also add a Requirements subsection. -## Log-Level +## Usage +Use examples liberally, and show the expected output if you can. It's helpful to have inline the smallest example of usage that you can demonstrate, while providing links to more sophisticated examples if they are too long to reasonably include in the README. -You can change the log-Level by setting the environment variable -AMMOS_LOG_LEVEL. To turn off logging completely set the level to -'CRITICAL' +## Support +Tell people where they can go to for help. It can be any combination of an issue tracker, a chat room, an email address, etc. -``` shell -export AMMOS_LOG_LEVEL=CRITICAL -``` +## Roadmap +If you have ideas for releases in the future, it is a good idea to list them in the README. -If the environment variable does not exist, the logging level from -ammos_logging.conf is used. +## Contributing +State if you are open to contributions and what your requirements are for accepting them. + +For people who want to make changes to your project, it's helpful to have some documentation on how to get started. Perhaps there is a script that they should run or some environment variables that they need to set. Make these steps explicit. These instructions could also be useful to your future self. + +You can also document commands to lint the code or run tests. These steps help to ensure high code quality and reduce the likelihood that the changes inadvertently break something. Having instructions for running tests is especially helpful if it requires external setup, such as starting a Selenium server for testing in a browser. + +## Authors and acknowledgment +Show your appreciation to those who have contributed to the project. + +## License +For open source projects, say how it is licensed. + +## Project status +If you have run out of energy or time for your project, put a note at the top of the README saying that development has slowed down or stopped completely. Someone may choose to fork your project or volunteer to step in as a maintainer or owner, allowing your project to keep going. You can also make an explicit request for maintainers. Index: mosreader/AbstractAmmosReader.py =================================================================== --- ammosreader/AbstractAmmosReader.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,158 +1,0 @@ -"""I provide a base class for specialized AmmosReaders.""" - -from abc import ABC, abstractmethod - -from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader -from ammosreader.AmmosSingleFrame import AmmosSingleFrame -from ammosreader.AmmosContainer import AmmosContainer -from ammosreader.AmmosConstants import FrameType -from ammosreader import logger - - -class AbstractAmmosReader(ABC): - """I implement a base class for specialized AmmosReaders.""" - - def __init__(self, file_name): - """ - I am the standard constructor for Ammos Readers. - - Additional information about the file can be added as key/value pairs in tags - - :param file_name: The file to read Ammos data from - :type file_name: str - """ - self.__file_name = file_name - self.__ammos_file = open(self.file_name, "rb") - self.__container = AmmosContainer(self.file_name, []) - self.__tags = {} - - @property - def file_name(self): - """I return the name of the original file.""" - return self.__file_name - - @property - def ammos_file(self): - """I return the file to read the data from.""" - return self.__ammos_file - - @property - def container(self): - """I return the container which stores the data read.""" - return self.__container - - @property - def tags(self): - """I return all the tags.""" - return self.__tags - - def add_tag(self, a_key, a_value): - """I add information to tags using a key/value pair.""" - assert a_key not in self.__tags - self.__tags[a_key] = a_value - - def rewind_to_start(self): - """I set the file pointer to the beginning of the file for the next operation.""" - self.ammos_file.seek(0) - - def read_all_frames_left(self): - """ - I read all remaining frames into my container until end of file is reached. - - :return: a container containing all frames read - :rtype: AmmosContainer - """ - frames_read = 0 - while True: - logger.info("Reading single frame %s ...", frames_read) - current_frame = self.read_next_single_frame() - if current_frame is not None: - frames_read += 1 - self.container.add_frame(current_frame) - else: - logger.debug("Frame: %s incomplete", frames_read+1) - break - logger.info("%s frames read", len(self.container.global_frames)) - return self.container - - def read_next_global_frame_header(self): - """ - I return the next global frame header read from current position in file. - - :return: the next global frame header or None if incomplete - :rtype: AmmosGlobalFrameHeader - """ - header_size = AmmosGlobalFrameHeader.HEADER_SIZE - - in_bytes = self.ammos_file.read(header_size) - logger.info("Reading next global frame header") - if ((not in_bytes) or (len(in_bytes) < header_size)): - logger.debug("Can not read all %s bytes of global frame header", header_size) - return None - - current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(in_bytes) - - if current_global_frame_header is None: - return None - - # logger.info("Current global frame header %s", current_global_frame_header) - return current_global_frame_header - - @abstractmethod - def read_next_global_frame_body(self, data_header_length): - """My descendents have to implement this.""" - pass - - def read_next_single_frame(self): - """ - I read and return a single global frame. - - :return: a single global frame - :rtype: AmmosSingleFrame - """ - global_frame_header = self.read_next_global_frame_header() - - if global_frame_header is None: - return None - - if global_frame_header is None: - logger.debug("Global frame header missing") - return None - - if global_frame_header.data_header_length is None: - logger.debug("Data header length empty") - return None - - if global_frame_header.frame_type not in list(FrameType): - logger.info("Unknown frame type %s found", global_frame_header.frame_type) - return None - - # FIXME: Refactor duplicate logic - This code stinks - - global_frame_body = None - - if_data_types = [FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT.value, - FrameType.IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT.value, - FrameType.IF_DATA_16BIT_REAL_REAL.value, - FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED.value, - FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED.value] - - if global_frame_header.frame_type == FrameType.AUDIO_DATA.value: - logger.info("Audio Datastream found") - global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length) - if global_frame_body is None: - return None - return AmmosSingleFrame(global_frame_header, global_frame_body) - - if global_frame_header.frame_type in if_data_types: - logger.info("IF Datastream found") - global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length) - if global_frame_body is None: - return None - return AmmosSingleFrame(global_frame_header, global_frame_body) - - if global_frame_body is None: - return None - - logger.info("Unsupported frame type %s found", global_frame_header.frame_type) - return None Index: mosreader/AmmosAudioDataBody.py =================================================================== --- ammosreader/AmmosAudioDataBody.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,34 +1,0 @@ -"""I provide an AMMOS data block for audio data frames.""" - - -class AmmosAudioDataBody: - """I implement an AMMOS data block for audio data frames.""" - - def __init__(self, pcm_data, number_of_channels=1, samples_per_channel=1, sample_size=1): - """I return a new AMMOS data block for audio data frames.""" - self.__number_of_channels = number_of_channels - self.__samples_per_channel = samples_per_channel - self.__sample_size = sample_size - self.__payload = pcm_data - - @property - def payload(self): - """I return the raw pcm data with channels interweaved.""" - return self.__payload - - def pcm_for_channel(self, channel_number): - """I return the raw pcm data for a given channel.""" - start_offset = channel_number * self.__sample_size - step = self.__sample_size * self.__number_of_channels - end = (len(self.__payload) // step) * step - channel_bytes = b"" - for each in range(start_offset, end, step): - channel_bytes += self.__payload[each:each+self.__sample_size] - return channel_bytes - - def __str__(self): - """I return the string representation of myself.""" - return ("Number of channels:" + str(self.__number_of_channels) + - "\nSamples per channel:" + str(self.__samples_per_channel) + - "\nSample size:" + str(self.__sample_size) + - "\nData size:" + str(len(self.__payload))) Index: mosreader/AmmosAudioDataHeader.py =================================================================== --- ammosreader/AmmosAudioDataHeader.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,50 +1,0 @@ -"""I provide an AMMOS data header for audio data frames.""" - -import struct -from ammosreader.AmmosConstants import AmmosAudioDemodType - - -class AmmosAudioDataHeader: - """I implement an AMMOS data header for audio data frames.""" - - HEADER_SIZE = 36 # 9 words - - @classmethod - def from_bytes(cls, in_bytes): - """I return an AMMOS data header from given bytes.""" - assert len(in_bytes) == cls.HEADER_SIZE - elements = struct.unpack(' bytearray: - """ - Gets the next bytes from the socket, for example headers and body data. - - Args: - byte_count (int): number of bytes to read - - Raises: - TimeoutError: Raises TimeoutError if the socket does not serve data anymore - - Returns: - bytearray: data from socket as bytearray - """ - - byte_array = [] - - while len(b''.join(byte_array)) < byte_count: - logger.info("Remaining Bytes: %s", byte_count - len(b''.join(byte_array))) - # logger.info(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}") - self.__socket.settimeout(5) - new_bytes = self.__socket.recv(byte_count - len(b''.join(byte_array)), socket.MSG_WAITALL) - - if not new_bytes: - raise TimeoutError("Socket timed out while reading data") - logger.info("Got %s bytes of %s remaining", len(new_bytes), byte_count - len(b''.join(byte_array))) - # logger.info(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} remaining") - - byte_array.append(new_bytes) - - return b''.join(byte_array) - - def __audio_data_body_to_numpy(self, audio_data_body: bytearray) -> np.ndarray: - """ - converts the audio data body to a numpy array - - Args: - audio_data_body (bytearray): audio data from audio data body - - Returns: - np.ndarray: audio data as numpy array - """ - - return np.frombuffer(audio_data_body, dtype=np.int16) - - def read_next_frame(self) -> tuple[np.ndarray, int]: - """Reads the next ammos audio frame from socket - - Raises: - TimeoutError: Raises TimeoutError if the socket does not serve data anymore - - Returns: - tuple[np.ndarray, int]: Contains the audio data and the sample rate - """ - - # get first byte of the day - self.__socket.settimeout(5) - - new_byte = self.__socket.recv(1, socket.MSG_WAITALL) - # raise Exception if socket does not return anything - if len(new_byte) < 1: - raise TimeoutError - - # read loop - while new_byte: - # - self.__magic_word_buffer.append(new_byte) - byte_array = b''.join(self.__magic_word_buffer) - - if byte_array.hex() == AmmosGlobalFrameHeader.MAGIC_WORD: - # print(byte_array.hex()) - - ammos_global_header_buffer = list(self.__magic_word_buffer) - ammos_global_header_buffer.append(self.__get_next_data(20)) - # while len(b''.join(ammos_global_header_buffer)) < 24: - # ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer)))) - - ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer)) - logger.info(ammos_global_header) - - if (ammos_global_header.data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL): - - byte_array_header = self.__get_next_data(AmmosExtendedAudioDataHeader.HEADER_SIZE) - # while len(b''.join(byte_array_header)) < 44: - # byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header)))) - - ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header) - logger.debug("%s, %s, %s", - str(ammos_extended_audio_data_header.number_of_samples or ''), - str(ammos_extended_audio_data_header.number_of_channels or ''), - str(ammos_extended_audio_data_header.sample_size or '')) - - audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples * - ammos_extended_audio_data_header.number_of_channels * - ammos_extended_audio_data_header.sample_size) - - audio_array = self.__audio_data_body_to_numpy(audio_body) - # print("44,256", len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate) - logger.debug("%s, %s", len(audio_array), len(audio_array) / ammos_extended_audio_data_header.sample_rate) - - return [audio_array, ammos_extended_audio_data_header.sample_rate] - - elif (ammos_global_header.data_header_length == AmmosAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL): - byte_array_header = self.__get_next_data(AmmosAudioDataHeader.HEADER_SIZE) - # while len(b''.join(byte_array_header)) < 36: - # byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header)))) - - ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header) - logger.debug("%s, %s, %s", - str(ammos_audio_data_header.number_of_samples or ''), - str(ammos_audio_data_header.number_of_channels or ''), - str(ammos_audio_data_header.sample_size or '')) - - audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples * - ammos_extended_audio_data_header.number_of_channels * - ammos_extended_audio_data_header.sample_size) - - audio_array = self.__audio_data_body_to_numpy(audio_body) - logger.debug("%s, %s", len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate) - - return [audio_array, ammos_audio_data_header.sample_rate] - - # get the next byte - self.__socket.settimeout(5) - - new_byte = self.__socket.recv(1, socket.MSG_WAITALL) - # raise Exception if socket does not return anything - if len(new_byte) < 1: - raise TimeoutError - - return None Index: mosreader/AmmosConstants.py =================================================================== --- ammosreader/AmmosConstants.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,56 +1,0 @@ -"""I provide several constants used in R&S software.""" - -from enum import Enum - - -class FrameType(int, Enum): - """I map numbers to human readable format specifiers.""" - - TEST_DATA = 0x00 - IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT = 0x01 - IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT = 0x02 - IF_DATA_16BIT_REAL_REAL = 0x03 - IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED = 0x04 - IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED = 0x05 - SPECTRUM_DATA_8BIT = 0x10 - SPECTRUM_DATA_16BIT = 0x11 - SPECTRUM_DATA_32BIT = 0x12 - SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x13 - SEGMENTATION_SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x14 - HF_TUNING_INDICATOR_DATA = 0x20 - HF_SCAN_CHANNEL_FOUND_DATA = 0x22 - HF_SCAN_FREQUENCY_FOUND_DATA = 0x23 - HF_SCAN_SWEEP_RESTARTED_DATA = 0x24 - HF_DEMODULATION_SYMBOL_STREAM_DATA = 0x30 - HF_DECODER_TEXT_DATA = 0x40 - HF_SPECTRUM_VISUALIZATION_DATA = 0x50 - HF_TIMEDOMAIN_VISUALIZATION_DATA = 0x51 - AUDIO_DATA = 0x100 - EMISSION_LIST_DATA = 0x110 - LEVEL_DATA = 0x120 - SYMBOL_DATA = 0x130 - INSTANTANEOUS_DATA = 0x140 - BURST_EMMISION_LIST = 0x150 - IMAGE_DATA = 0x160 - TRANSMISSION_SYSTEM_RESULT_DATA = 0x170 - PULSE_DESCRIPTION_WORD_DATA = 0x200 - PULSE_REPETITION_WORD_DATA = 0x210 - EM050_SCAN_DATA = 0x4000 - SCAN_LEVEL = 0x4001 - SCAN_TUNING = 0x4002 - SCAN_LEVEL_TUNING = 0x4003 - DDF_RESERVED_START = 0x5000 - DDF_RESERVED_END = 0x50FF - - -class AmmosAudioDemodType(int, Enum): - """I map numbers to human readable demodulation types.""" - - FM = 0 - AM = 1 - ISB = 5 - CW = 6 - USB = 7 - LSB = 8 - DIGITAL = 256 - UNKNOWN = 0xFFFFFFFF Index: mosreader/AmmosContainer.py =================================================================== --- ammosreader/AmmosContainer.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,59 +1,0 @@ -"""I provide an Ammos container.""" - - -class AmmosContainer(): - """I store the content of an R&S Ammos file in a more accessible way.""" - def __init__(self, name, frames): - self.__name = name - self.__global_frames = frames - self.__tags = [] - - @property - def name(self): - return self.__name - - @name.setter - def name(self, a_name): - self.__name = a_name - - @property - def global_frames(self): - return self.__global_frames - - def add_tag(self, tag): - self.__tags.append(tag) - - def add_frame(self, frame): - self.__global_frames.append(frame) - - def size(self): - # return sum([each.global_frame_header.frame_length for each in self.__global_frames]) - return sum(each.global_frame_header.frame_length for each in self.__global_frames) - - def frequencies(self): - """I return a list of unique frequencies inside this container.""" - return list({each.global_frame_body.data_header.frequency for each in self.__global_frames}) - - def frame_types(self): - return list({each.global_frame_header.frame_type for each in self.__global_frames}) - - def frame_sizes(self): - return [each.global_frame_header.frame_length for each in self.__global_frames] - - def unique_frame_sizes(self): - return list({each.global_frame_header.frame_length for each in self.__global_frames}) - - def is_homogenic(self): - return (len(self.unique_frame_sizes()) == 1) and (len(self.frame_types()) == 1) - - def __str__(self): - start_time = self.__global_frames[0].global_frame_body.data_header.timestamp - end_time = self.__global_frames[-1].global_frame_body.data_header.timestamp - - frq = str(self.__global_frames[0].global_frame_body.data_header.frequency) - - return ("Start time: " + str(start_time) + - "\nEnd time : " + str(end_time) + "\nFrequencies: " + frq) - - def data_only(self): - return [each.global_frame_body.data_body.data for each in self.__global_frames] Index: mosreader/AmmosExtendedAudioDataHeader.py =================================================================== --- ammosreader/AmmosExtendedAudioDataHeader.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,40 +1,0 @@ -"""I provide an Ammos extended data header for audio data frames.""" - -import struct -from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader - - -class AmmosExtendedAudioDataHeader: - - HEADER_SIZE = 44 # 11 words - - @classmethod - def from_bytes(cls, in_bytes): - """I return a new AMMOS extended data header for audio frames built from given bytes.""" - assert len(in_bytes) == cls.HEADER_SIZE - standard_header = AmmosAudioDataHeader.from_bytes(in_bytes[0:AmmosAudioDataHeader.HEADER_SIZE]) - extended_header_elements = struct.unpack('= unix_time: - raise OverflowError("Timestamp invalid") - time_of_arrival = nanoseconds - - third_word = bin(int.from_bytes(parts[1], byteorder='little')) - padding = 32-len(str(third_word)[2:]) - third_word_bit_string = "0" * padding + str(third_word)[2:] - pdw_format_identifier = int(third_word_bit_string[0:6], 2) - center_frequency = int(third_word_bit_string[5:32], 2) # Value of zero means unknown - - fourth_word = bin(int.from_bytes(parts[2], byteorder='little')) - padding = 32-len(str(fourth_word)[2:]) - fourth_word_bit_string = "0" * padding + str(fourth_word)[2:] - is_valid = bool(int(fourth_word_bit_string[0])) - is_pulse = bool(int(fourth_word_bit_string[1])) - level_unit = int(fourth_word_bit_string[2]) - signal_start_missing = bool(int(fourth_word_bit_string[3])) - signal_end_missing = bool(int(fourth_word_bit_string[4])) - pulse_width = int(fourth_word_bit_string[7:33], 2) - - fifth_word = bin(int.from_bytes(parts[3], byteorder='little')) - padding = 32-len(str(fifth_word)[2:]) - fifth_word_bit_string = "0" * padding + str(fifth_word)[2:] - frequency_shift_or_bandwidth = int(fifth_word_bit_string[0:20], 2) - pulse_level_or_pulse_field_strength = BitArray(bin=fifth_word_bit_string[20:32]).int - - sixth_word = bin(int.from_bytes(parts[4], byteorder='little')) - padding = 32-len(str(sixth_word)[2:]) - sixth_word_bit_string = "0" * padding + str(sixth_word)[2:] - region_of_interest = bool(int(sixth_word_bit_string[0])) - azimuth_confidence = int(sixth_word_bit_string[1:7], 2) - modulation = int(sixth_word_bit_string[7:12], 2) - sector = int(sixth_word_bit_string[28:32], 2) - - seventh_word = bin(int.from_bytes(parts[5], byteorder='little')) - padding = 32-len(str(seventh_word)[2:]) - seventh_word_bit_string = "0" * padding + str(seventh_word)[2:] - polarity = int(seventh_word_bit_string[0:2], 2) - df_quality = int(seventh_word_bit_string[2:9], 2) - elevation = BitArray(bin=seventh_word_bit_string[9:20]).int - azimuth = int(seventh_word_bit_string[20:32], 2) - - eighth_word = bin(int.from_bytes(parts[5], byteorder='little')) - padding = 32-len(str(eighth_word)[2:]) - eighth_word_bit_string = "0" * padding + str(eighth_word)[2:] - channel = int(eighth_word_bit_string[0:4], 2) - - return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit, - signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth, - pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation, - sector, polarity, df_quality, elevation, azimuth, channel) - - def __init__(self, time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, - level_unit, signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth, - pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation, - sector, polarity, df_quality, elevation, azimuth, channel): - r""" - I return an instance of an Pulse Data word. - - :param time_of_arrival: nanoseconds since 1970-01-01 00:00:00 - :type time_of_arrival: Integer - :param pdw_format: format code - :type pdw_format: Integer - :param center_frequency: center frequency in KHz - :type center_frequency: Integer - :param is_valid: flag to mark if pdw data body is valid - :type is_valid: Boolean - :param is_pulse: flag to mark if pdw data body contains a pulse or a continuous wave signal - :type is_pulse: Boolean - :param level_unit: 1 means dBµV - 0 means dBµV/m - :type level_unit: Integer - :param signal_start_missing: signal started before time of arrival - :type signal_start_missing: Boolean - :param signal_end_missing: signal stops after time of arrival - :type signal_end_missing: Boolean - :param pulse_width: pulse width in nanoseconds - Zero if no valid pulse detected - :type pulse_width: Integer - :param frequency_shift_or_bandwidth: Value in KHz - Value set to 1048575 means Unknown - :type frequency_shift_or_bandwidth: Integer - :param pulse_level_or_pulse_field_strength: Pulse level or Pulse Field Strength depending on level_unit \ - (-200.0...200.0) in tenth degrees / minus 2048 means no valid level detected - :type pulse_level_or_pulse_field_strength: Integer - :param region_of_interest: Marks if signal is from region of interest - :type region_of_interest: Boolean - :param azimuth_confidence: degree in tenth steps of (0-62) / 63 means confidence unknown - :type azimuth_confidence: Integer - :param modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.) - :type modulation: String - :param sector: reference antenna sector (0-15) - :type sector: Integer - :param polarity: Horizontal, Vertical, Clockwise, Counter clockwise - :type polarity: String - :param df_quality: Direction finding quality in percent (0-100) - Zero means unknown - :type df_quality: Integer - :param elevation: elevation of incoming signal (from -90 to 90 degree) in steps tenths degree \ - minus 1024 means unknown - :type elevation: Integer - :param azimuth: azimuth of incoming signal (from 0 to 3599 tenth degree) plus 4095 means unknown - :type azimuth: Integer - :param channel: detecting channel (0-16) - Zero means unknown - :type channel: Integer - :return: An instance of class PDW with attributes set according to the data of a data body - :rtype: PDW - """ - assert ((-2000 <= pulse_level_or_pulse_field_strength <= 2000) or - (pulse_level_or_pulse_field_strength == -2048) or - (pulse_level_or_pulse_field_strength is None)) - assert ((0 <= azimuth_confidence <= 63) or (azimuth_confidence is None)) - assert 0 <= sector <= 15 - assert 0 <= polarity <= 3 - assert 0 <= df_quality <= 100 - assert ((-900 <= elevation <= 900) or (elevation == -1024) or (elevation is None)) - assert ((0 <= azimuth <= 3599) or (azimuth == 4095) or (azimuth is None)) - assert 0 <= channel <= 15 - self.__time_of_arrival = time_of_arrival # - self.__pdw_format_identifier = pdw_format_identifier - self.__center_frequency = center_frequency # - self.__is_valid = is_valid # - self.__is_pulse = is_pulse # - self.__level_unit = level_unit # - self.__signal_start_missing = signal_start_missing - self.__signal_end_missing = signal_end_missing - self.__pulse_width = pulse_width # - self.__frequency_shift_or_bandwidth = frequency_shift_or_bandwidth # - self.__pulse_level_or_pulse_field_strength = pulse_level_or_pulse_field_strength # - self.__region_of_interest = region_of_interest - self.__azimuth_confidence = azimuth_confidence - self.__modulation = modulation # - self.__sector = sector - self.__polarity = polarity - self.__df_quality = df_quality # - self.__elevation = elevation # - self.__azimuth = azimuth - self.__channel = channel # - - def __str__(self): - """ - I return the string representation of myself. - - :rtype: str - """ - output = ("Time of arrival: " + str(self.time_of_arrival) + "\n" + - "PDW Format identifier: " + str(self.pdw_format_identifier) + "\n" + - "Center frequency: " + str(self.center_frequency) + " KHz\n") - - if self.is_valid: - output += "Signal: Valid\n" - else: - output += "Signal: Invalid\n" - - if self.is_pulse: - output += "Signal type: Pulse\n" - else: - output += "Signal type: Continuous wave\n" - - # FIXME: use fstrings or another better performance string concat - if self.level_unit == "dbµV": - output += "Pulse level: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV\n" - else: - output += "Pulse field strength: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV/m\n" - - output += ("Pulse width: " + str(self.pulse_width) + " nanoseconds\n" + - "Frequency shift or bandwidth: " + str(self.frequency_shift_or_bandwidth) + " KHz\n") - - if self.region_of_interest: - output += "Region of interest: Yes\n" - else: - output += "Region of interest: No\n" - - if self.azimuth_confidence == None: - output += "Azimuth confidence: Invalid\n" - else: - output += "Azimuth confidence: " + str(self.azimuth_confidence) + " degree\n" - - output += "Modulation: " + str(self.modulation) + "\n" - - if self.sector == None: - output += "Sector: Unknown\n" - else: - output += "Sector:" + str(self.sector) + "\n" - - output += "Polarity: " + str(self.polarity) + "\n" - - output += "DF quality: " + str(self.df_quality) + " %\n" - - if self.elevation == None: - output += "Elevation: Unknown\n" - else: - output += "Elevation: " + str(self.elevation) + " degree\n" - - if self.azimuth == None: - output += "Azimuth: Unknown\n" - else: - output += "Azimuth: " + str(self.azimuth) + " degree\n" - - output += "Channel: " + str(self.channel) + "\n" - - return output - - @property - def time_of_arrival(self): - return np.datetime64(self.__time_of_arrival, 'ns') - - @property - def center_frequency(self): - if self.__center_frequency == 0: - return None - return self.__center_frequency - - @property - def is_valid(self): - return self.__is_valid - - @property - def is_invalid(self): - return not self.__is_valid - - @property - def is_pulse(self): - return self.__is_pulse - - @property - def is_cw(self): - return not self.__is_pulse - - @property - def level_unit(self): - if self.__level_unit == 1: - return "dbµV" - else: - return "dbµV/m" - - @property - def signal_start_missing(self): - return self.__signal_start_missing - - @property - def signal_end_missing(self): - return self.__signal_end_missing - - @property - def pulse_width(self): - if self.__pulse_width == 0: - return None - return self.__pulse_width - - @property - def frequency_shift_or_bandwidth(self): - if self.__frequency_shift_or_bandwidth == 1048575: - return None - return self.__frequency_shift_or_bandwidth - - @property - def pulse_level_or_pulse_field_strength(self): - if self.__pulse_level_or_pulse_field_strength == -2048: - return None - return (self.__pulse_level_or_pulse_field_strength / 10) - - @property - def azimuth_confidence(self): - if self.__azimuth_confidence == 63: - return None - return (self.__azimuth_confidence / 10) - - @property - def modulation(self): - return PDW.modulations().get(self.__modulation, PDW.modulations()[0]) - - @property - def region_of_interest(self): - return self.__region_of_interest - - @property - def sector(self): - if self.__sector == 0: - return None - return self.__sector - - @property - def polarity(self): - return PDW.polarities().get(self.__polarity, PDW.polarities()[0]) - - @property - def elevation(self): - if self.__elevation == -1024: - return None - return self.__elevation / 10 - - @property - def azimuth(self): - if self.__azimuth == 4095: - return None - return self.__azimuth / 10 - - @property - def channel(self): - return self.__channel - - @property - def pdw_format_identifier(self): - return self.__pdw_format_identifier - - @property - def df_quality(self): - return self.__df_quality - - def to_datacrunch_json(self): - import uuid - return {'TOA': int(self.time_of_arrival), - 'FORMAT': self.pdw_format_identifier, - 'CENTER_FREQUENCY': self.center_frequency, - 'VALID_FLAG': self.is_valid, - 'PULSE_FLAG': self.is_pulse, - 'LU_FLAG': self.level_unit, - 'SNS_FLAG': self.signal_start_missing, - 'SNE_FLAG': self.signal_end_missing, - 'LEVEL': self.pulse_level_or_pulse_field_strength, - 'PULSE_WIDTH': self.pulse_width, - 'BANDWIDTH': self.frequency_shift_or_bandwidth, - 'IN_OUT_FLAG': self.region_of_interest, - 'DF_CONFIDENCE': self.azimuth_confidence, - 'MODULATION': self.modulation, - 'SECTOR': self.sector, - 'POL': self.polarity, - 'DF_QUALITY': self.df_quality, - 'ELEVATION': self.elevation, - 'AZIMUTH': self.azimuth, - 'CHANNEL': self.channel, - 'UUID': uuid.uuid4() - } - - def to_json(self): - return {'TIME OF ARRIVAL': self.time_of_arrival, - 'FORMATIDENTIFIER': self.pdw_format_identifier, - 'CENTER FREQUENCY': self.center_frequency, - 'VALID': self.is_valid, - 'PULSE': self.is_pulse, - 'PULSELEVEL': self.pulse_level_or_pulse_field_strength, - 'PULSEWIDTH': self.pulse_width, - 'BANDWIDTH': self.frequency_shift_or_bandwidth, - 'REGIONOFINTEREST': self.region_of_interest, - 'AZIMUTHCONFIDENCE': self.azimuth_confidence, - 'MODULATION': self.modulation, - 'SECTOR': self.sector, - 'POLARITY': self.polarity, - 'DFQUALITY': self.df_quality, - 'ELEVATION': self.elevation, - 'AZIMUTH': self.azimuth, - 'CHANNEL': self.channel - } - -if __name__ == '__main__': - pass Index: mosreader/PPDWContainer.py =================================================================== --- ammosreader/PPDWContainer.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,82 +1,0 @@ -from datetime import datetime -import numpy as np - -class PPDWContainer(): - """ - I store multiple signals imported from one or more .ppdw files - .. automethod:: __init__ - """ - - def __init__(self, name, signals=None): - - if signals is None: - self.signals = [] - else: - self.signals = signals - - self.name = name - - def __str__(self): - return "\n".join(["Number of pulses:" + str(len(self.signals)), - "Number of valid pulses:" + str(self.number_of_valid_pulses()), - "Number of invalid pulses:" + str(self.number_of_invalid_pulses()), - "Start time:" + str(self.start_time()), - "End time:" + str(self.end_time())]) - - def add(self, a_pdw): - # print(a_pdw) - self.signals.append(a_pdw) - - def as_pulse_dict(self): - pulse_dict = {} - pulse_dict[0] = self.signals[0].to_json() - for index, current_pdw in enumerate(self.signals[1:]): - pulse_dict[index+1] = current_pdw.to_json() - pulse_dict[index]["dtoa"] = (current_pdw.time_of_arrival - pulse_dict[index]["time of arrival"]).item()/1000.0 - pulse_dict[index+1]["dtoa"] = 0.0 # np.timedelta64(0, 'us') - return pulse_dict - - def number_of_valid_pulses(self): - return sum(each.is_valid() for each in self.signals) - - def number_of_invalid_pulses(self): - return sum(not each.is_valid() for each in self.signals) - - def without_invalids(self): - return PPDWContainer(self.name, [each for each in self.signals if each.is_valid()]) - - def julian_date_string(self): - ts = (self.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's') - time_tuple = datetime.utcfromtimestamp(ts).timetuple() - return str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3) - - def center_frequencies(self): - return list({each.center_frequency for each in self.signals}) - - def channels(self): - return list({each.channel for each in self.signals}) - - def modulations(self): - return list({each.modulation for each in self.signals}) - - def bandwidths(self): - return list({each.frequency_shift_or_bandwidth for each in self.signals}) - - def start_time(self): - return min(each.time_of_arrival for each in self.signals) - - def end_time(self): - return max(each.time_of_arrival for each in self.signals) - - def to_json(self): - return {'JULIANDATE': self.julian_date_string(), - 'STARTTIME': str(self.start_time()), - 'ENDTIME': str(self.end_time()), - 'CENTERFREQUENCIES': self.center_frequencies(), - 'CHANNELS': self.channels(), - 'MODULATIONS': self.modulations(), - 'BANDWIDTHS': self.bandwidths() - } - -if __name__ == '__main__': - pass Index: mosreader/PPDWReader.py =================================================================== --- ammosreader/PPDWReader.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,32 +1,0 @@ -from ammosreader.PDW import PDW -from ammosreader.PPDWContainer import PPDWContainer -from ammosreader import logger - - -class PPDWReader(): - """I read data from a file and return a PPDWContainer.""" - - def __init__(self, a_file): - - self.file = a_file - assert self.file.is_file() - self.content = self.file.read_bytes() - self.cursor = 0 - self.container = PPDWContainer(self.file.stem) - - def read_all_frames_left(self): - try: - while self.cursor <= len(self.content) - 32: - current_bytes = self.content[self.cursor:self.cursor+32] - assert len(current_bytes) == 32 - if not current_bytes: - logger.info('End of file detected') - break - if self.cursor + 32 >= len(self.content): - logger.info('Can not read all 32 bytes of next PDW. EOF') - break - self.container.add(PDW.from_bytes(current_bytes)) - self.cursor += 32 - except Exception: - return None - return self.container Index: mosreader/__init__.py =================================================================== --- ammosreader/__init__.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,25 +1,0 @@ -import logging -import logging.config -import os -from pathlib import Path - -# FIXME: change the default log path to /var/log/ammos/ammos.log when system configuration created this dir -# with the appropriate rights - -log_dir = Path(os.environ.get('AMMOS_LOG_DIR', '/tmp/')) - -log_path = log_dir / 'ammos.log' - -if not log_path.exists(): - try: - log_path.touch() - except PermissionError: - conf_file = Path(__file__).parent / 'ammos_logging.conf' - print("Conf file", conf_file) - logging.config.fileConfig(conf_file) -else: - logging.basicConfig(filename=str(log_path), encoding='utf-8', level=logging.DEBUG) - -logger = logging.getLogger(__name__) - -logger.setLevel(os.environ.get('AMMOS_LOG_LEVEL', logger.getEffectiveLevel())) Index: mosreader/ammos_logging.conf =================================================================== --- ammosreader/ammos_logging.conf (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,21 +1,0 @@ -[loggers] -keys=root - -[handlers] -keys=stream_handler - -[formatters] -keys=formatter - -[logger_root] -level=DEBUG -handlers=stream_handler - -[handler_stream_handler] -class=StreamHandler -level=DEBUG -formatter=formatter -args=(sys.stderr,) - -[formatter_formatter] -format=%(asctime)s %(name)-12s %(levelname)-8s %(message)s Index: c/ammosreader/AbstractAmmosReader.html =================================================================== --- doc/ammosreader/AbstractAmmosReader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,646 +1,0 @@ - - - - - - -ammosreader.AbstractAmmosReader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AbstractAmmosReader

-
-
-

I provide a base class for specialized AmmosReaders.

-
- -Expand source code - -
"""I provide a base class for specialized AmmosReaders."""
-
-from abc import ABC, abstractmethod
-
-from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader
-from ammosreader.AmmosSingleFrame import AmmosSingleFrame
-from ammosreader.AmmosContainer import AmmosContainer
-from ammosreader.AmmosConstants import FrameType
-from ammosreader import logger
-
-
-class AbstractAmmosReader(ABC):
-    """I implement a base class for specialized AmmosReaders."""
-
-    def __init__(self, file_name):
-        """
-        I am the standard constructor for Ammos Readers.
-
-        Additional information about the file can be added as key/value pairs in tags
-
-        :param file_name: The file to read Ammos data from
-        :type file_name: str
-        """
-        self.__file_name = file_name
-        self.__ammos_file = open(self.file_name, "rb")
-        self.__container = AmmosContainer(self.file_name, [])
-        self.__tags = {}
-
-    @property
-    def file_name(self):
-        """I return the name of the original file."""
-        return self.__file_name
-
-    @property
-    def ammos_file(self):
-        """I return the file to read the data from."""
-        return self.__ammos_file
-
-    @property
-    def container(self):
-        """I return the container which stores the data read."""
-        return self.__container
-
-    @property
-    def tags(self):
-        """I return all the tags."""
-        return self.__tags
-
-    def add_tag(self, a_key, a_value):
-        """I add information to tags using a key/value pair."""
-        assert a_key not in self.__tags
-        self.__tags[a_key] = a_value
-
-    def rewind_to_start(self):
-        """I set the file pointer to the beginning of the file for the next operation."""
-        self.ammos_file.seek(0)
-
-    def read_all_frames_left(self):
-        """
-        I read all remaining frames into my container until end of file is reached.
-
-        :return: a container containing all frames read
-        :rtype: AmmosContainer
-        """
-        frames_read = 0
-        while True:
-            logger.info("Reading single frame %s ...", frames_read)
-            current_frame = self.read_next_single_frame()
-            if current_frame is not None:
-                frames_read += 1
-                self.container.add_frame(current_frame)
-            else:
-                logger.debug("Frame: %s incomplete", frames_read+1)
-                break
-        logger.info("%s frames read", len(self.container.global_frames))
-        return self.container
-
-    def read_next_global_frame_header(self):
-        """
-        I return the next global frame header read from current position in file.
-
-        :return: the next global frame header or None if incomplete
-        :rtype: AmmosGlobalFrameHeader
-        """
-        header_size = AmmosGlobalFrameHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-        logger.info("Reading next global frame header")
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame header", header_size)
-            return None
-
-        current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(in_bytes)
-
-        if current_global_frame_header is None:
-            return None
-
-        # logger.info("Current global frame header %s", current_global_frame_header)
-        return current_global_frame_header
-
-    @abstractmethod
-    def read_next_global_frame_body(self, data_header_length):
-        """My descendents have to implement this."""
-        pass
-
-    def read_next_single_frame(self):
-        """
-        I read and return a single global frame.
-
-        :return: a single global frame
-        :rtype: AmmosSingleFrame
-        """
-        global_frame_header = self.read_next_global_frame_header()
-
-        if global_frame_header is None:
-            return None
-
-        if global_frame_header is None:
-            logger.debug("Global frame header missing")
-            return None
-
-        if global_frame_header.data_header_length is None:
-            logger.debug("Data header length empty")
-            return None
-
-        if global_frame_header.frame_type not in list(FrameType):
-            logger.info("Unknown frame type %s found", global_frame_header.frame_type)
-            return None
-
-        # FIXME: Refactor duplicate logic - This code stinks
-
-        global_frame_body = None
-
-        if_data_types = [FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT.value,
-                         FrameType.IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT.value,
-                         FrameType.IF_DATA_16BIT_REAL_REAL.value,
-                         FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED.value,
-                         FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED.value]
-
-        if global_frame_header.frame_type == FrameType.AUDIO_DATA.value:
-            logger.info("Audio Datastream found")
-            global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
-            if global_frame_body is None:
-                return None
-            return AmmosSingleFrame(global_frame_header, global_frame_body)
-
-        if global_frame_header.frame_type in if_data_types:
-            logger.info("IF Datastream found")
-            global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
-            if global_frame_body is None:
-                return None
-            return AmmosSingleFrame(global_frame_header, global_frame_body)
-
-        if global_frame_body is None:
-            return None
-
-        logger.info("Unsupported frame type %s found", global_frame_header.frame_type)
-        return None
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AbstractAmmosReader -(file_name) -
-
-

I implement a base class for specialized AmmosReaders.

-

I am the standard constructor for Ammos Readers.

-

Additional information about the file can be added as key/value pairs in tags

-

:param file_name: The file to read Ammos data from -:type file_name: str

-
- -Expand source code - -
class AbstractAmmosReader(ABC):
-    """I implement a base class for specialized AmmosReaders."""
-
-    def __init__(self, file_name):
-        """
-        I am the standard constructor for Ammos Readers.
-
-        Additional information about the file can be added as key/value pairs in tags
-
-        :param file_name: The file to read Ammos data from
-        :type file_name: str
-        """
-        self.__file_name = file_name
-        self.__ammos_file = open(self.file_name, "rb")
-        self.__container = AmmosContainer(self.file_name, [])
-        self.__tags = {}
-
-    @property
-    def file_name(self):
-        """I return the name of the original file."""
-        return self.__file_name
-
-    @property
-    def ammos_file(self):
-        """I return the file to read the data from."""
-        return self.__ammos_file
-
-    @property
-    def container(self):
-        """I return the container which stores the data read."""
-        return self.__container
-
-    @property
-    def tags(self):
-        """I return all the tags."""
-        return self.__tags
-
-    def add_tag(self, a_key, a_value):
-        """I add information to tags using a key/value pair."""
-        assert a_key not in self.__tags
-        self.__tags[a_key] = a_value
-
-    def rewind_to_start(self):
-        """I set the file pointer to the beginning of the file for the next operation."""
-        self.ammos_file.seek(0)
-
-    def read_all_frames_left(self):
-        """
-        I read all remaining frames into my container until end of file is reached.
-
-        :return: a container containing all frames read
-        :rtype: AmmosContainer
-        """
-        frames_read = 0
-        while True:
-            logger.info("Reading single frame %s ...", frames_read)
-            current_frame = self.read_next_single_frame()
-            if current_frame is not None:
-                frames_read += 1
-                self.container.add_frame(current_frame)
-            else:
-                logger.debug("Frame: %s incomplete", frames_read+1)
-                break
-        logger.info("%s frames read", len(self.container.global_frames))
-        return self.container
-
-    def read_next_global_frame_header(self):
-        """
-        I return the next global frame header read from current position in file.
-
-        :return: the next global frame header or None if incomplete
-        :rtype: AmmosGlobalFrameHeader
-        """
-        header_size = AmmosGlobalFrameHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-        logger.info("Reading next global frame header")
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame header", header_size)
-            return None
-
-        current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(in_bytes)
-
-        if current_global_frame_header is None:
-            return None
-
-        # logger.info("Current global frame header %s", current_global_frame_header)
-        return current_global_frame_header
-
-    @abstractmethod
-    def read_next_global_frame_body(self, data_header_length):
-        """My descendents have to implement this."""
-        pass
-
-    def read_next_single_frame(self):
-        """
-        I read and return a single global frame.
-
-        :return: a single global frame
-        :rtype: AmmosSingleFrame
-        """
-        global_frame_header = self.read_next_global_frame_header()
-
-        if global_frame_header is None:
-            return None
-
-        if global_frame_header is None:
-            logger.debug("Global frame header missing")
-            return None
-
-        if global_frame_header.data_header_length is None:
-            logger.debug("Data header length empty")
-            return None
-
-        if global_frame_header.frame_type not in list(FrameType):
-            logger.info("Unknown frame type %s found", global_frame_header.frame_type)
-            return None
-
-        # FIXME: Refactor duplicate logic - This code stinks
-
-        global_frame_body = None
-
-        if_data_types = [FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT.value,
-                         FrameType.IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT.value,
-                         FrameType.IF_DATA_16BIT_REAL_REAL.value,
-                         FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED.value,
-                         FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED.value]
-
-        if global_frame_header.frame_type == FrameType.AUDIO_DATA.value:
-            logger.info("Audio Datastream found")
-            global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
-            if global_frame_body is None:
-                return None
-            return AmmosSingleFrame(global_frame_header, global_frame_body)
-
-        if global_frame_header.frame_type in if_data_types:
-            logger.info("IF Datastream found")
-            global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
-            if global_frame_body is None:
-                return None
-            return AmmosSingleFrame(global_frame_header, global_frame_body)
-
-        if global_frame_body is None:
-            return None
-
-        logger.info("Unsupported frame type %s found", global_frame_header.frame_type)
-        return None
-
-

Ancestors

-
    -
  • abc.ABC
    • -
-

Subclasses

- -

Instance variables

-
-
var ammos_file
-
-

I return the file to read the data from.

-
- -Expand source code - -
@property
-def ammos_file(self):
-    """I return the file to read the data from."""
-    return self.__ammos_file
-
-
-
var container
-
-

I return the container which stores the data read.

-
- -Expand source code - -
@property
-def container(self):
-    """I return the container which stores the data read."""
-    return self.__container
-
-
-
var file_name
-
-

I return the name of the original file.

-
- -Expand source code - -
@property
-def file_name(self):
-    """I return the name of the original file."""
-    return self.__file_name
-
-
-
var tags
-
-

I return all the tags.

-
- -Expand source code - -
@property
-def tags(self):
-    """I return all the tags."""
-    return self.__tags
-
-
-
-

Methods

-
-
-def add_tag(self, a_key, a_value) -
-
-

I add information to tags using a key/value pair.

-
- -Expand source code - -
def add_tag(self, a_key, a_value):
-    """I add information to tags using a key/value pair."""
-    assert a_key not in self.__tags
-    self.__tags[a_key] = a_value
-
-
-
-def read_all_frames_left(self) -
-
-

I read all remaining frames into my container until end of file is reached.

-

:return: a container containing all frames read -:rtype: AmmosContainer

-
- -Expand source code - -
def read_all_frames_left(self):
-    """
-    I read all remaining frames into my container until end of file is reached.
-
-    :return: a container containing all frames read
-    :rtype: AmmosContainer
-    """
-    frames_read = 0
-    while True:
-        logger.info("Reading single frame %s ...", frames_read)
-        current_frame = self.read_next_single_frame()
-        if current_frame is not None:
-            frames_read += 1
-            self.container.add_frame(current_frame)
-        else:
-            logger.debug("Frame: %s incomplete", frames_read+1)
-            break
-    logger.info("%s frames read", len(self.container.global_frames))
-    return self.container
-
-
-
-def read_next_global_frame_body(self, data_header_length) -
-
-

My descendents have to implement this.

-
- -Expand source code - -
@abstractmethod
-def read_next_global_frame_body(self, data_header_length):
-    """My descendents have to implement this."""
-    pass
-
-
-
-def read_next_global_frame_header(self) -
-
-

I return the next global frame header read from current position in file.

-

:return: the next global frame header or None if incomplete -:rtype: AmmosGlobalFrameHeader

-
- -Expand source code - -
def read_next_global_frame_header(self):
-    """
-    I return the next global frame header read from current position in file.
-
-    :return: the next global frame header or None if incomplete
-    :rtype: AmmosGlobalFrameHeader
-    """
-    header_size = AmmosGlobalFrameHeader.HEADER_SIZE
-
-    in_bytes = self.ammos_file.read(header_size)
-    logger.info("Reading next global frame header")
-    if ((not in_bytes) or (len(in_bytes) < header_size)):
-        logger.debug("Can not read all %s bytes of global frame header", header_size)
-        return None
-
-    current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(in_bytes)
-
-    if current_global_frame_header is None:
-        return None
-
-    # logger.info("Current global frame header %s", current_global_frame_header)
-    return current_global_frame_header
-
-
-
-def read_next_single_frame(self) -
-
-

I read and return a single global frame.

-

:return: a single global frame -:rtype: AmmosSingleFrame

-
- -Expand source code - -
def read_next_single_frame(self):
-    """
-    I read and return a single global frame.
-
-    :return: a single global frame
-    :rtype: AmmosSingleFrame
-    """
-    global_frame_header = self.read_next_global_frame_header()
-
-    if global_frame_header is None:
-        return None
-
-    if global_frame_header is None:
-        logger.debug("Global frame header missing")
-        return None
-
-    if global_frame_header.data_header_length is None:
-        logger.debug("Data header length empty")
-        return None
-
-    if global_frame_header.frame_type not in list(FrameType):
-        logger.info("Unknown frame type %s found", global_frame_header.frame_type)
-        return None
-
-    # FIXME: Refactor duplicate logic - This code stinks
-
-    global_frame_body = None
-
-    if_data_types = [FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT.value,
-                     FrameType.IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT.value,
-                     FrameType.IF_DATA_16BIT_REAL_REAL.value,
-                     FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED.value,
-                     FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED.value]
-
-    if global_frame_header.frame_type == FrameType.AUDIO_DATA.value:
-        logger.info("Audio Datastream found")
-        global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
-        if global_frame_body is None:
-            return None
-        return AmmosSingleFrame(global_frame_header, global_frame_body)
-
-    if global_frame_header.frame_type in if_data_types:
-        logger.info("IF Datastream found")
-        global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
-        if global_frame_body is None:
-            return None
-        return AmmosSingleFrame(global_frame_header, global_frame_body)
-
-    if global_frame_body is None:
-        return None
-
-    logger.info("Unsupported frame type %s found", global_frame_header.frame_type)
-    return None
-
-
-
-def rewind_to_start(self) -
-
-

I set the file pointer to the beginning of the file for the next operation.

-
- -Expand source code - -
def rewind_to_start(self):
-    """I set the file pointer to the beginning of the file for the next operation."""
-    self.ammos_file.seek(0)
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosAudioDataBody.html =================================================================== --- doc/ammosreader/AmmosAudioDataBody.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,190 +1,0 @@ - - - - - - -ammosreader.AmmosAudioDataBody API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosAudioDataBody

-
-
-

I provide an AMMOS data block for audio data frames.

-
- -Expand source code - -
"""I provide an AMMOS data block for audio data frames."""
-
-
-class AmmosAudioDataBody:
-    """I implement an AMMOS data block for audio data frames."""
-
-    def __init__(self, pcm_data, number_of_channels=1, samples_per_channel=1, sample_size=1):
-        """I return a new AMMOS data block for audio data frames."""
-        self.__number_of_channels = number_of_channels
-        self.__samples_per_channel = samples_per_channel
-        self.__sample_size = sample_size
-        self.__payload = pcm_data
-
-    @property
-    def payload(self):
-        """I return the raw pcm data with channels interweaved."""
-        return self.__payload
-
-    def pcm_for_channel(self, channel_number):
-        """I return the raw pcm data for a given channel."""
-        start_offset = channel_number * self.__sample_size
-        step = self.__sample_size * self.__number_of_channels
-        end = (len(self.__payload) // step) * step
-        channel_bytes = b""
-        for each in range(start_offset, end, step):
-            channel_bytes += self.__payload[each:each+self.__sample_size]
-        return channel_bytes
-
-    def __str__(self):
-        """I return the string representation of myself."""
-        return ("Number of channels:" + str(self.__number_of_channels) +
-                "\nSamples per channel:" + str(self.__samples_per_channel) +
-                "\nSample size:" + str(self.__sample_size) +
-                "\nData size:" + str(len(self.__payload)))
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosAudioDataBody -(pcm_data, number_of_channels=1, samples_per_channel=1, sample_size=1) -
-
-

I implement an AMMOS data block for audio data frames.

-

I return a new AMMOS data block for audio data frames.

-
- -Expand source code - -
class AmmosAudioDataBody:
-    """I implement an AMMOS data block for audio data frames."""
-
-    def __init__(self, pcm_data, number_of_channels=1, samples_per_channel=1, sample_size=1):
-        """I return a new AMMOS data block for audio data frames."""
-        self.__number_of_channels = number_of_channels
-        self.__samples_per_channel = samples_per_channel
-        self.__sample_size = sample_size
-        self.__payload = pcm_data
-
-    @property
-    def payload(self):
-        """I return the raw pcm data with channels interweaved."""
-        return self.__payload
-
-    def pcm_for_channel(self, channel_number):
-        """I return the raw pcm data for a given channel."""
-        start_offset = channel_number * self.__sample_size
-        step = self.__sample_size * self.__number_of_channels
-        end = (len(self.__payload) // step) * step
-        channel_bytes = b""
-        for each in range(start_offset, end, step):
-            channel_bytes += self.__payload[each:each+self.__sample_size]
-        return channel_bytes
-
-    def __str__(self):
-        """I return the string representation of myself."""
-        return ("Number of channels:" + str(self.__number_of_channels) +
-                "\nSamples per channel:" + str(self.__samples_per_channel) +
-                "\nSample size:" + str(self.__sample_size) +
-                "\nData size:" + str(len(self.__payload)))
-
-

Instance variables

-
-
var payload
-
-

I return the raw pcm data with channels interweaved.

-
- -Expand source code - -
@property
-def payload(self):
-    """I return the raw pcm data with channels interweaved."""
-    return self.__payload
-
-
-
-

Methods

-
-
-def pcm_for_channel(self, channel_number) -
-
-

I return the raw pcm data for a given channel.

-
- -Expand source code - -
def pcm_for_channel(self, channel_number):
-    """I return the raw pcm data for a given channel."""
-    start_offset = channel_number * self.__sample_size
-    step = self.__sample_size * self.__number_of_channels
-    end = (len(self.__payload) // step) * step
-    channel_bytes = b""
-    for each in range(start_offset, end, step):
-        channel_bytes += self.__payload[each:each+self.__sample_size]
-    return channel_bytes
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosAudioDataHeader.html =================================================================== --- doc/ammosreader/AmmosAudioDataHeader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,216 +1,0 @@ - - - - - - -ammosreader.AmmosAudioDataHeader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosAudioDataHeader

-
-
-

I provide an AMMOS data header for audio data frames.

-
- -Expand source code - -
"""I provide an AMMOS data header for audio data frames."""
-
-import struct
-from ammosreader.AmmosConstants import AmmosAudioDemodType
-
-
-class AmmosAudioDataHeader:
-    """I implement an AMMOS data header for audio data frames."""
-
-    HEADER_SIZE = 36  # 9 words
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return an AMMOS data header from given bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-        elements = struct.unpack('<IIQIIIII', in_bytes)
-        sample_rate = elements[0]
-        status = elements[1]
-        frequency = elements[2]
-        demod_bandwidth = elements[3]
-        demod_type = AmmosAudioDemodType(elements[4])
-        number_of_samples = elements[5]
-        number_of_channels = elements[6]
-        sample_size = elements[7]
-        return AmmosAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
-                                    number_of_samples, number_of_channels, sample_size)
-
-    def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples,
-                 number_of_channels, sample_size):
-        """I create a new instance of myself using the above parameters."""
-        self.sample_rate = sample_rate
-        self.status = status
-        self.frequency = frequency
-        self.demod_bandwidth = demod_bandwidth
-        self.demod_type = AmmosAudioDemodType(demod_type)
-        self.number_of_samples = number_of_samples
-        self.number_of_channels = number_of_channels
-        self.sample_size = sample_size
-
-    def __str__(self):
-        """I return the string representation of myself."""
-        return ("\nAmmosAudioDataHeader\n" +
-                "Sample rate:" + str(self.sample_rate) + "\n" +
-                "Status:" + str(self.status) + "\n" +
-                "Frequency:" + str(self.frequency) + "\n" +
-                "Demodulation bandwidth:" + str(self.demod_bandwidth) + "\n" +
-                "Demodulation type:" + str(self.demod_type) + "\n" +
-                "Sample count:" + str(self.number_of_samples) + "\n" +
-                "Channel count:" + str(self.number_of_channels) + "\n" +
-                "Sample size:" + str(self.sample_size) + "\n")
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosAudioDataHeader -(sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples, number_of_channels, sample_size) -
-
-

I implement an AMMOS data header for audio data frames.

-

I create a new instance of myself using the above parameters.

-
- -Expand source code - -
class AmmosAudioDataHeader:
-    """I implement an AMMOS data header for audio data frames."""
-
-    HEADER_SIZE = 36  # 9 words
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return an AMMOS data header from given bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-        elements = struct.unpack('<IIQIIIII', in_bytes)
-        sample_rate = elements[0]
-        status = elements[1]
-        frequency = elements[2]
-        demod_bandwidth = elements[3]
-        demod_type = AmmosAudioDemodType(elements[4])
-        number_of_samples = elements[5]
-        number_of_channels = elements[6]
-        sample_size = elements[7]
-        return AmmosAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
-                                    number_of_samples, number_of_channels, sample_size)
-
-    def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples,
-                 number_of_channels, sample_size):
-        """I create a new instance of myself using the above parameters."""
-        self.sample_rate = sample_rate
-        self.status = status
-        self.frequency = frequency
-        self.demod_bandwidth = demod_bandwidth
-        self.demod_type = AmmosAudioDemodType(demod_type)
-        self.number_of_samples = number_of_samples
-        self.number_of_channels = number_of_channels
-        self.sample_size = sample_size
-
-    def __str__(self):
-        """I return the string representation of myself."""
-        return ("\nAmmosAudioDataHeader\n" +
-                "Sample rate:" + str(self.sample_rate) + "\n" +
-                "Status:" + str(self.status) + "\n" +
-                "Frequency:" + str(self.frequency) + "\n" +
-                "Demodulation bandwidth:" + str(self.demod_bandwidth) + "\n" +
-                "Demodulation type:" + str(self.demod_type) + "\n" +
-                "Sample count:" + str(self.number_of_samples) + "\n" +
-                "Channel count:" + str(self.number_of_channels) + "\n" +
-                "Sample size:" + str(self.sample_size) + "\n")
-
-

Class variables

-
-
var HEADER_SIZE
-
-
-
-
-

Static methods

-
-
-def from_bytes(in_bytes) -
-
-

I return an AMMOS data header from given bytes.

-
- -Expand source code - -
@classmethod
-def from_bytes(cls, in_bytes):
-    """I return an AMMOS data header from given bytes."""
-    assert len(in_bytes) == cls.HEADER_SIZE
-    elements = struct.unpack('<IIQIIIII', in_bytes)
-    sample_rate = elements[0]
-    status = elements[1]
-    frequency = elements[2]
-    demod_bandwidth = elements[3]
-    demod_type = AmmosAudioDemodType(elements[4])
-    number_of_samples = elements[5]
-    number_of_channels = elements[6]
-    sample_size = elements[7]
-    return AmmosAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
-                                number_of_samples, number_of_channels, sample_size)
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosAudioReader.html =================================================================== --- doc/ammosreader/AmmosAudioReader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,511 +1,0 @@ - - - - - - -ammosreader.AmmosAudioReader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosAudioReader

-
-
-

I provide a specialized Ammos Reader for audio data.

-
- -Expand source code - -
"""I provide a specialized Ammos Reader for audio data."""
-
-from ammosreader.AbstractAmmosReader import AbstractAmmosReader
-from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody
-from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
-from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader
-from ammosreader.AmmosAudioDataBody import AmmosAudioDataBody
-from ammosreader import logger
-
-
-class AmmosAudioReader(AbstractAmmosReader):
-    """I read the audio data embedded in an R&S AMMOS recording."""
-
-    def read_next_global_frame_body_data_header(self):
-        """
-        I return the next global frame body data header from current position in file.
-
-        :param data_header_size: the number of bytes to read
-        :type data_header_size: int
-        :return: the next Ammos Audio Data header or None if incomplete
-        :rtype: AmmosAudioDataHeader
-        """
-        header_size = AmmosAudioDataHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-
-        logger.info("\nReading global frame body standard data header\n")
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame body data header", header_size)
-            return None
-        return AmmosAudioDataHeader.from_bytes(in_bytes)
-
-    def read_next_global_frame_body_extended_data_header(self):
-        """
-        I return the next global frame body extended data header from current position in file.
-
-        :return: the next Ammos Audio Extended Data header or None if incomplete
-        :rtype: AmmosExtendedAudioDataHeader
-        """
-        header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
-            return None
-        return AmmosExtendedAudioDataHeader.from_bytes(in_bytes)
-
-    def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
-        """
-        I return the next audio data read from current position in file.
-
-        :param number_of_samples: the number of samples per channel inside data body
-        :type number_of_samples: int
-
-        :param number_of_channels: number of channels (e.g. mono, stereo or even more)
-        :type number_of_channels: int
-
-        :param sample_size: sample size in bytes (1, 2 or 4 bytes)
-        :type sample_size: int
-
-        :return: the next audio data or None if incomplete
-        :rtype: bytes
-        """
-        total = number_of_samples*number_of_channels*sample_size
-
-        byte_string = self.ammos_file.read(total)
-
-        if len(byte_string) != total:
-            logger.debug("Can not read all %s bytes of data body", total)
-            return None
-        return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)
-
-    def read_next_global_frame_body(self, data_header_length):
-        """
-        I return the next global frame body read from current position in file.
-
-        :param data_header_length: the length of the data header
-        :type data_header_length: int
-        """
-        audio_data_header = None
-
-        if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
-            logger.info("Read standard data header")
-            audio_data_header = self.read_next_global_frame_body_data_header()
-
-        logger.info("Data header length %s", data_header_length)
-        if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
-            logger.info("Read extended data header")
-            audio_data_header = self.read_next_global_frame_body_extended_data_header()
-
-        if audio_data_header is None:
-            logger.debug("Data header missing or format unknown")
-            return None
-
-        audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
-                                                         audio_data_header.number_of_channels,
-                                                         audio_data_header.sample_size)
-
-        if audio_data_body is None:
-            logger.debug("Data body missing")
-            return None
-
-        return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
-
-    def pcm_for_channel(self, a_channel):
-        """
-        I return the raw pcm audio data for a given channel.
-
-        :param a_channel: the channel I have to extract
-        :type a_channel: int
-
-        :rtype: bytes
-        """
-        return (b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel)
-                          for each in self.container.global_frames]))
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosAudioReader -(file_name) -
-
-

I read the audio data embedded in an R&S AMMOS recording.

-

I am the standard constructor for Ammos Readers.

-

Additional information about the file can be added as key/value pairs in tags

-

:param file_name: The file to read Ammos data from -:type file_name: str

-
- -Expand source code - -
class AmmosAudioReader(AbstractAmmosReader):
-    """I read the audio data embedded in an R&S AMMOS recording."""
-
-    def read_next_global_frame_body_data_header(self):
-        """
-        I return the next global frame body data header from current position in file.
-
-        :param data_header_size: the number of bytes to read
-        :type data_header_size: int
-        :return: the next Ammos Audio Data header or None if incomplete
-        :rtype: AmmosAudioDataHeader
-        """
-        header_size = AmmosAudioDataHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-
-        logger.info("\nReading global frame body standard data header\n")
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame body data header", header_size)
-            return None
-        return AmmosAudioDataHeader.from_bytes(in_bytes)
-
-    def read_next_global_frame_body_extended_data_header(self):
-        """
-        I return the next global frame body extended data header from current position in file.
-
-        :return: the next Ammos Audio Extended Data header or None if incomplete
-        :rtype: AmmosExtendedAudioDataHeader
-        """
-        header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
-            return None
-        return AmmosExtendedAudioDataHeader.from_bytes(in_bytes)
-
-    def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
-        """
-        I return the next audio data read from current position in file.
-
-        :param number_of_samples: the number of samples per channel inside data body
-        :type number_of_samples: int
-
-        :param number_of_channels: number of channels (e.g. mono, stereo or even more)
-        :type number_of_channels: int
-
-        :param sample_size: sample size in bytes (1, 2 or 4 bytes)
-        :type sample_size: int
-
-        :return: the next audio data or None if incomplete
-        :rtype: bytes
-        """
-        total = number_of_samples*number_of_channels*sample_size
-
-        byte_string = self.ammos_file.read(total)
-
-        if len(byte_string) != total:
-            logger.debug("Can not read all %s bytes of data body", total)
-            return None
-        return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)
-
-    def read_next_global_frame_body(self, data_header_length):
-        """
-        I return the next global frame body read from current position in file.
-
-        :param data_header_length: the length of the data header
-        :type data_header_length: int
-        """
-        audio_data_header = None
-
-        if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
-            logger.info("Read standard data header")
-            audio_data_header = self.read_next_global_frame_body_data_header()
-
-        logger.info("Data header length %s", data_header_length)
-        if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
-            logger.info("Read extended data header")
-            audio_data_header = self.read_next_global_frame_body_extended_data_header()
-
-        if audio_data_header is None:
-            logger.debug("Data header missing or format unknown")
-            return None
-
-        audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
-                                                         audio_data_header.number_of_channels,
-                                                         audio_data_header.sample_size)
-
-        if audio_data_body is None:
-            logger.debug("Data body missing")
-            return None
-
-        return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
-
-    def pcm_for_channel(self, a_channel):
-        """
-        I return the raw pcm audio data for a given channel.
-
-        :param a_channel: the channel I have to extract
-        :type a_channel: int
-
-        :rtype: bytes
-        """
-        return (b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel)
-                          for each in self.container.global_frames]))
-
-

Ancestors

- -

Methods

-
-
-def pcm_for_channel(self, a_channel) -
-
-

I return the raw pcm audio data for a given channel.

-

:param a_channel: the channel I have to extract -:type a_channel: int

-

:rtype: bytes

-
- -Expand source code - -
def pcm_for_channel(self, a_channel):
-    """
-    I return the raw pcm audio data for a given channel.
-
-    :param a_channel: the channel I have to extract
-    :type a_channel: int
-
-    :rtype: bytes
-    """
-    return (b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel)
-                      for each in self.container.global_frames]))
-
-
-
-def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size) -
-
-

I return the next audio data read from current position in file.

-

:param number_of_samples: the number of samples per channel inside data body -:type number_of_samples: int

-

:param number_of_channels: number of channels (e.g. mono, stereo or even more) -:type number_of_channels: int

-

:param sample_size: sample size in bytes (1, 2 or 4 bytes) -:type sample_size: int

-

:return: the next audio data or None if incomplete -:rtype: bytes

-
- -Expand source code - -
def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
-    """
-    I return the next audio data read from current position in file.
-
-    :param number_of_samples: the number of samples per channel inside data body
-    :type number_of_samples: int
-
-    :param number_of_channels: number of channels (e.g. mono, stereo or even more)
-    :type number_of_channels: int
-
-    :param sample_size: sample size in bytes (1, 2 or 4 bytes)
-    :type sample_size: int
-
-    :return: the next audio data or None if incomplete
-    :rtype: bytes
-    """
-    total = number_of_samples*number_of_channels*sample_size
-
-    byte_string = self.ammos_file.read(total)
-
-    if len(byte_string) != total:
-        logger.debug("Can not read all %s bytes of data body", total)
-        return None
-    return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)
-
-
-
-def read_next_global_frame_body(self, data_header_length) -
-
-

I return the next global frame body read from current position in file.

-

:param data_header_length: the length of the data header -:type data_header_length: int

-
- -Expand source code - -
def read_next_global_frame_body(self, data_header_length):
-    """
-    I return the next global frame body read from current position in file.
-
-    :param data_header_length: the length of the data header
-    :type data_header_length: int
-    """
-    audio_data_header = None
-
-    if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
-        logger.info("Read standard data header")
-        audio_data_header = self.read_next_global_frame_body_data_header()
-
-    logger.info("Data header length %s", data_header_length)
-    if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
-        logger.info("Read extended data header")
-        audio_data_header = self.read_next_global_frame_body_extended_data_header()
-
-    if audio_data_header is None:
-        logger.debug("Data header missing or format unknown")
-        return None
-
-    audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
-                                                     audio_data_header.number_of_channels,
-                                                     audio_data_header.sample_size)
-
-    if audio_data_body is None:
-        logger.debug("Data body missing")
-        return None
-
-    return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
-
-
-
-def read_next_global_frame_body_data_header(self) -
-
-

I return the next global frame body data header from current position in file.

-

:param data_header_size: the number of bytes to read -:type data_header_size: int -:return: the next Ammos Audio Data header or None if incomplete -:rtype: AmmosAudioDataHeader

-
- -Expand source code - -
def read_next_global_frame_body_data_header(self):
-    """
-    I return the next global frame body data header from current position in file.
-
-    :param data_header_size: the number of bytes to read
-    :type data_header_size: int
-    :return: the next Ammos Audio Data header or None if incomplete
-    :rtype: AmmosAudioDataHeader
-    """
-    header_size = AmmosAudioDataHeader.HEADER_SIZE
-
-    in_bytes = self.ammos_file.read(header_size)
-
-    logger.info("\nReading global frame body standard data header\n")
-    if ((not in_bytes) or (len(in_bytes) < header_size)):
-        logger.debug("Can not read all %s bytes of global frame body data header", header_size)
-        return None
-    return AmmosAudioDataHeader.from_bytes(in_bytes)
-
-
-
-def read_next_global_frame_body_extended_data_header(self) -
-
-

I return the next global frame body extended data header from current position in file.

-

:return: the next Ammos Audio Extended Data header or None if incomplete -:rtype: AmmosExtendedAudioDataHeader

-
- -Expand source code - -
def read_next_global_frame_body_extended_data_header(self):
-    """
-    I return the next global frame body extended data header from current position in file.
-
-    :return: the next Ammos Audio Extended Data header or None if incomplete
-    :rtype: AmmosExtendedAudioDataHeader
-    """
-    header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
-
-    in_bytes = self.ammos_file.read(header_size)
-
-    if ((not in_bytes) or (len(in_bytes) < header_size)):
-        logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
-        return None
-    return AmmosExtendedAudioDataHeader.from_bytes(in_bytes)
-
-
-
-

Inherited members

- -
-
-
-
- -
- - - Index: c/ammosreader/AmmosAudioSocketReader.html =================================================================== --- doc/ammosreader/AmmosAudioSocketReader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,501 +1,0 @@ - - - - - - -ammosreader.AmmosAudioSocketReader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosAudioSocketReader

-
-
-

I read a Ammos datastream from a socket.

-
- -Expand source code - -
"""I read a Ammos datastream from a socket."""
-
-# import select
-import socket
-from collections import deque
-import numpy as np
-from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
-from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader
-from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader
-from ammosreader.AmmosConstants import AmmosAudioDemodType
-from ammosreader import logger
-
-
-class AmmosAudioSocketReader:
-    """I read Ammos Audio data from a socket"""
-
-    def __init__(self, in_socket: socket.socket):
-        """
-        Initializes the AmmosAudioSocketReader
-
-        Args:
-            socket (socket.socket): socket to read from
-        """
-
-        # buffer for reading socket bytewise und check for the magic word
-        self.__magic_word_buffer = deque(maxlen=4)
-
-        # input socket to read from
-        self.__socket = in_socket
-
-    def __get_next_data(self, byte_count: int) -> bytearray:
-        """
-        Gets the next bytes from the socket, for example headers and body data.
-
-        Args:
-            byte_count (int): number of bytes to read
-
-        Raises:
-            TimeoutError: Raises TimeoutError if the socket does not serve data anymore
-
-        Returns:
-            bytearray: data from socket as bytearray
-        """
-
-        byte_array = []
-
-        while len(b''.join(byte_array)) < byte_count:
-            logger.info("Remaining Bytes: %s", byte_count - len(b''.join(byte_array)))
-            # logger.info(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}")
-            self.__socket.settimeout(5)
-            new_bytes = self.__socket.recv(byte_count - len(b''.join(byte_array)), socket.MSG_WAITALL)
-
-            if not new_bytes:
-                raise TimeoutError("Socket timed out while reading data")
-            logger.info("Got %s bytes of %s remaining", len(new_bytes), byte_count - len(b''.join(byte_array)))
-            # logger.info(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} remaining")
-
-            byte_array.append(new_bytes)
-
-        return b''.join(byte_array)
-
-    def __audio_data_body_to_numpy(self, audio_data_body: bytearray) -> np.ndarray:
-        """
-        converts the audio data body to a numpy array
-
-        Args:
-            audio_data_body (bytearray): audio data from audio data body
-
-        Returns:
-            np.ndarray: audio data as numpy array
-        """
-
-        return np.frombuffer(audio_data_body, dtype=np.int16)
-
-    def read_next_frame(self) -> tuple[np.ndarray, int]:
-        """Reads the next ammos audio frame from socket
-
-        Raises:
-            TimeoutError: Raises TimeoutError if the socket does not serve data anymore
-
-        Returns:
-            tuple[np.ndarray, int]: Contains the audio data and the sample rate
-        """
-
-        # get first byte of the day
-        self.__socket.settimeout(5)
-
-        new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
-        # raise Exception if socket does not return anything
-        if len(new_byte) < 1:
-            raise TimeoutError
-
-        # read loop
-        while new_byte:
-            #
-            self.__magic_word_buffer.append(new_byte)
-            byte_array = b''.join(self.__magic_word_buffer)
-
-            if byte_array.hex() == AmmosGlobalFrameHeader.MAGIC_WORD:
-                # print(byte_array.hex())
-
-                ammos_global_header_buffer = list(self.__magic_word_buffer)
-                ammos_global_header_buffer.append(self.__get_next_data(20))
-                # while len(b''.join(ammos_global_header_buffer)) < 24:
-                #    ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer))))
-
-                ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer))
-                logger.info(ammos_global_header)
-
-                if (ammos_global_header.data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
-
-                    byte_array_header = self.__get_next_data(AmmosExtendedAudioDataHeader.HEADER_SIZE)
-                    # while len(b''.join(byte_array_header)) < 44:
-                    #    byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header))))
-
-                    ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header)
-                    logger.debug(str(ammos_extended_audio_data_header.number_of_samples or ''),
-                                 str(ammos_extended_audio_data_header.number_of_channels or ''),
-                                 str(ammos_extended_audio_data_header.sample_size or ''))
-
-                    audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
-                                                      ammos_extended_audio_data_header.number_of_channels *
-                                                      ammos_extended_audio_data_header.sample_size)
-
-                    audio_array = self.__audio_data_body_to_numpy(audio_body)
-                    # print("44,256", len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
-                    logger.debug("%s, %s", len(audio_array), len(audio_array) / ammos_extended_audio_data_header.sample_rate)
-
-                    return [audio_array, ammos_extended_audio_data_header.sample_rate]
-
-                elif (ammos_global_header.data_header_length == AmmosAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
-                    byte_array_header = self.__get_next_data(AmmosAudioDataHeader.HEADER_SIZE)
-                    # while len(b''.join(byte_array_header)) < 36:
-                    #    byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header))))
-
-                    ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header)
-                    logger.debug(str(ammos_audio_data_header.number_of_samples or ''),
-                                 str(ammos_audio_data_header.number_of_channels or ''),
-                                 str(ammos_audio_data_header.sample_size or ''))
-
-                    audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
-                                                      ammos_extended_audio_data_header.number_of_channels *
-                                                      ammos_extended_audio_data_header.sample_size)
-
-                    audio_array = self.__audio_data_body_to_numpy(audio_body)
-                    logger.debug("%s, %s", len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
-
-                    return [audio_array, ammos_audio_data_header.sample_rate]
-
-            # get the next byte
-            self.__socket.settimeout(5)
-
-            new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
-            # raise Exception if socket does not return anything
-            if len(new_byte) < 1:
-                raise TimeoutError
-
-        return None
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosAudioSocketReader -(in_socket: socket.socket) -
-
-

I read Ammos Audio data from a socket

-

Initializes the AmmosAudioSocketReader

-

Args

-
-
socket : socket.socket
-
socket to read from
-
-
- -Expand source code - -
class AmmosAudioSocketReader:
-    """I read Ammos Audio data from a socket"""
-
-    def __init__(self, in_socket: socket.socket):
-        """
-        Initializes the AmmosAudioSocketReader
-
-        Args:
-            socket (socket.socket): socket to read from
-        """
-
-        # buffer for reading socket bytewise und check for the magic word
-        self.__magic_word_buffer = deque(maxlen=4)
-
-        # input socket to read from
-        self.__socket = in_socket
-
-    def __get_next_data(self, byte_count: int) -> bytearray:
-        """
-        Gets the next bytes from the socket, for example headers and body data.
-
-        Args:
-            byte_count (int): number of bytes to read
-
-        Raises:
-            TimeoutError: Raises TimeoutError if the socket does not serve data anymore
-
-        Returns:
-            bytearray: data from socket as bytearray
-        """
-
-        byte_array = []
-
-        while len(b''.join(byte_array)) < byte_count:
-            logger.info("Remaining Bytes: %s", byte_count - len(b''.join(byte_array)))
-            # logger.info(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}")
-            self.__socket.settimeout(5)
-            new_bytes = self.__socket.recv(byte_count - len(b''.join(byte_array)), socket.MSG_WAITALL)
-
-            if not new_bytes:
-                raise TimeoutError("Socket timed out while reading data")
-            logger.info("Got %s bytes of %s remaining", len(new_bytes), byte_count - len(b''.join(byte_array)))
-            # logger.info(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} remaining")
-
-            byte_array.append(new_bytes)
-
-        return b''.join(byte_array)
-
-    def __audio_data_body_to_numpy(self, audio_data_body: bytearray) -> np.ndarray:
-        """
-        converts the audio data body to a numpy array
-
-        Args:
-            audio_data_body (bytearray): audio data from audio data body
-
-        Returns:
-            np.ndarray: audio data as numpy array
-        """
-
-        return np.frombuffer(audio_data_body, dtype=np.int16)
-
-    def read_next_frame(self) -> tuple[np.ndarray, int]:
-        """Reads the next ammos audio frame from socket
-
-        Raises:
-            TimeoutError: Raises TimeoutError if the socket does not serve data anymore
-
-        Returns:
-            tuple[np.ndarray, int]: Contains the audio data and the sample rate
-        """
-
-        # get first byte of the day
-        self.__socket.settimeout(5)
-
-        new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
-        # raise Exception if socket does not return anything
-        if len(new_byte) < 1:
-            raise TimeoutError
-
-        # read loop
-        while new_byte:
-            #
-            self.__magic_word_buffer.append(new_byte)
-            byte_array = b''.join(self.__magic_word_buffer)
-
-            if byte_array.hex() == AmmosGlobalFrameHeader.MAGIC_WORD:
-                # print(byte_array.hex())
-
-                ammos_global_header_buffer = list(self.__magic_word_buffer)
-                ammos_global_header_buffer.append(self.__get_next_data(20))
-                # while len(b''.join(ammos_global_header_buffer)) < 24:
-                #    ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer))))
-
-                ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer))
-                logger.info(ammos_global_header)
-
-                if (ammos_global_header.data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
-
-                    byte_array_header = self.__get_next_data(AmmosExtendedAudioDataHeader.HEADER_SIZE)
-                    # while len(b''.join(byte_array_header)) < 44:
-                    #    byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header))))
-
-                    ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header)
-                    logger.debug(str(ammos_extended_audio_data_header.number_of_samples or ''),
-                                 str(ammos_extended_audio_data_header.number_of_channels or ''),
-                                 str(ammos_extended_audio_data_header.sample_size or ''))
-
-                    audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
-                                                      ammos_extended_audio_data_header.number_of_channels *
-                                                      ammos_extended_audio_data_header.sample_size)
-
-                    audio_array = self.__audio_data_body_to_numpy(audio_body)
-                    # print("44,256", len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
-                    logger.debug("%s, %s", len(audio_array), len(audio_array) / ammos_extended_audio_data_header.sample_rate)
-
-                    return [audio_array, ammos_extended_audio_data_header.sample_rate]
-
-                elif (ammos_global_header.data_header_length == AmmosAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
-                    byte_array_header = self.__get_next_data(AmmosAudioDataHeader.HEADER_SIZE)
-                    # while len(b''.join(byte_array_header)) < 36:
-                    #    byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header))))
-
-                    ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header)
-                    logger.debug(str(ammos_audio_data_header.number_of_samples or ''),
-                                 str(ammos_audio_data_header.number_of_channels or ''),
-                                 str(ammos_audio_data_header.sample_size or ''))
-
-                    audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
-                                                      ammos_extended_audio_data_header.number_of_channels *
-                                                      ammos_extended_audio_data_header.sample_size)
-
-                    audio_array = self.__audio_data_body_to_numpy(audio_body)
-                    logger.debug("%s, %s", len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
-
-                    return [audio_array, ammos_audio_data_header.sample_rate]
-
-            # get the next byte
-            self.__socket.settimeout(5)
-
-            new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
-            # raise Exception if socket does not return anything
-            if len(new_byte) < 1:
-                raise TimeoutError
-
-        return None
-
-

Methods

-
-
-def read_next_frame(self) ‑> tuple[numpy.ndarray, int] -
-
-

Reads the next ammos audio frame from socket

-

Raises

-
-
TimeoutError
-
Raises TimeoutError if the socket does not serve data anymore
-
-

Returns

-
-
tuple[np.ndarray, int]
-
Contains the audio data and the sample rate
-
-
- -Expand source code - -
def read_next_frame(self) -> tuple[np.ndarray, int]:
-    """Reads the next ammos audio frame from socket
-
-    Raises:
-        TimeoutError: Raises TimeoutError if the socket does not serve data anymore
-
-    Returns:
-        tuple[np.ndarray, int]: Contains the audio data and the sample rate
-    """
-
-    # get first byte of the day
-    self.__socket.settimeout(5)
-
-    new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
-    # raise Exception if socket does not return anything
-    if len(new_byte) < 1:
-        raise TimeoutError
-
-    # read loop
-    while new_byte:
-        #
-        self.__magic_word_buffer.append(new_byte)
-        byte_array = b''.join(self.__magic_word_buffer)
-
-        if byte_array.hex() == AmmosGlobalFrameHeader.MAGIC_WORD:
-            # print(byte_array.hex())
-
-            ammos_global_header_buffer = list(self.__magic_word_buffer)
-            ammos_global_header_buffer.append(self.__get_next_data(20))
-            # while len(b''.join(ammos_global_header_buffer)) < 24:
-            #    ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer))))
-
-            ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer))
-            logger.info(ammos_global_header)
-
-            if (ammos_global_header.data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
-
-                byte_array_header = self.__get_next_data(AmmosExtendedAudioDataHeader.HEADER_SIZE)
-                # while len(b''.join(byte_array_header)) < 44:
-                #    byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header))))
-
-                ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header)
-                logger.debug(str(ammos_extended_audio_data_header.number_of_samples or ''),
-                             str(ammos_extended_audio_data_header.number_of_channels or ''),
-                             str(ammos_extended_audio_data_header.sample_size or ''))
-
-                audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
-                                                  ammos_extended_audio_data_header.number_of_channels *
-                                                  ammos_extended_audio_data_header.sample_size)
-
-                audio_array = self.__audio_data_body_to_numpy(audio_body)
-                # print("44,256", len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
-                logger.debug("%s, %s", len(audio_array), len(audio_array) / ammos_extended_audio_data_header.sample_rate)
-
-                return [audio_array, ammos_extended_audio_data_header.sample_rate]
-
-            elif (ammos_global_header.data_header_length == AmmosAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
-                byte_array_header = self.__get_next_data(AmmosAudioDataHeader.HEADER_SIZE)
-                # while len(b''.join(byte_array_header)) < 36:
-                #    byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header))))
-
-                ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header)
-                logger.debug(str(ammos_audio_data_header.number_of_samples or ''),
-                             str(ammos_audio_data_header.number_of_channels or ''),
-                             str(ammos_audio_data_header.sample_size or ''))
-
-                audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
-                                                  ammos_extended_audio_data_header.number_of_channels *
-                                                  ammos_extended_audio_data_header.sample_size)
-
-                audio_array = self.__audio_data_body_to_numpy(audio_body)
-                logger.debug("%s, %s", len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
-
-                return [audio_array, ammos_audio_data_header.sample_rate]
-
-        # get the next byte
-        self.__socket.settimeout(5)
-
-        new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
-        # raise Exception if socket does not return anything
-        if len(new_byte) < 1:
-            raise TimeoutError
-
-    return None
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosConstants.html =================================================================== --- doc/ammosreader/AmmosConstants.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,436 +1,0 @@ - - - - - - -ammosreader.AmmosConstants API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosConstants

-
-
-

I provide several constants used in R&S software.

-
- -Expand source code - -
"""I provide several constants used in R&S software."""
-
-from enum import Enum
-
-
-class FrameType(int, Enum):
-    """I map numbers to human readable format specifiers."""
-
-    TEST_DATA = 0x00
-    IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT = 0x01
-    IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT = 0x02
-    IF_DATA_16BIT_REAL_REAL = 0x03
-    IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED = 0x04
-    IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED = 0x05
-    SPECTRUM_DATA_8BIT = 0x10
-    SPECTRUM_DATA_16BIT = 0x11
-    SPECTRUM_DATA_32BIT = 0x12
-    SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x13
-    SEGMENTATION_SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x14
-    HF_TUNING_INDICATOR_DATA = 0x20
-    HF_SCAN_CHANNEL_FOUND_DATA = 0x22
-    HF_SCAN_FREQUENCY_FOUND_DATA = 0x23
-    HF_SCAN_SWEEP_RESTARTED_DATA = 0x24
-    HF_DEMODULATION_SYMBOL_STREAM_DATA = 0x30
-    HF_DECODER_TEXT_DATA = 0x40
-    HF_SPECTRUM_VISUALIZATION_DATA = 0x50
-    HF_TIMEDOMAIN_VISUALIZATION_DATA = 0x51
-    AUDIO_DATA = 0x100
-    EMISSION_LIST_DATA = 0x110
-    LEVEL_DATA = 0x120
-    SYMBOL_DATA = 0x130
-    INSTANTANEOUS_DATA = 0x140
-    BURST_EMMISION_LIST = 0x150
-    IMAGE_DATA = 0x160
-    TRANSMISSION_SYSTEM_RESULT_DATA = 0x170
-    PULSE_DESCRIPTION_WORD_DATA = 0x200
-    PULSE_REPETITION_WORD_DATA = 0x210
-    EM050_SCAN_DATA = 0x4000
-    SCAN_LEVEL = 0x4001
-    SCAN_TUNING = 0x4002
-    SCAN_LEVEL_TUNING = 0x4003
-    DDF_RESERVED_START = 0x5000
-    DDF_RESERVED_END = 0x50FF
-
-
-class AmmosAudioDemodType(int, Enum):
-    """I map numbers to human readable demodulation types."""
-
-    FM = 0
-    AM = 1
-    ISB = 5
-    CW = 6
-    USB = 7
-    LSB = 8
-    DIGITAL = 256
-    UNKNOWN = 0xFFFFFFFF
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosAudioDemodType -(value, names=None, *, module=None, qualname=None, type=None, start=1) -
-
-

I map numbers to human readable demodulation types.

-
- -Expand source code - -
class AmmosAudioDemodType(int, Enum):
-    """I map numbers to human readable demodulation types."""
-
-    FM = 0
-    AM = 1
-    ISB = 5
-    CW = 6
-    USB = 7
-    LSB = 8
-    DIGITAL = 256
-    UNKNOWN = 0xFFFFFFFF
-
-

Ancestors

-
    -
  • builtins.int
    • -
  • enum.Enum
    • -
-

Class variables

-
-
var AM
-
-
-
-
var CW
-
-
-
-
var DIGITAL
-
-
-
-
var FM
-
-
-
-
var ISB
-
-
-
-
var LSB
-
-
-
-
var UNKNOWN
-
-
-
-
var USB
-
-
-
-
-
-
-class FrameType -(value, names=None, *, module=None, qualname=None, type=None, start=1) -
-
-

I map numbers to human readable format specifiers.

-
- -Expand source code - -
class FrameType(int, Enum):
-    """I map numbers to human readable format specifiers."""
-
-    TEST_DATA = 0x00
-    IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT = 0x01
-    IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT = 0x02
-    IF_DATA_16BIT_REAL_REAL = 0x03
-    IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED = 0x04
-    IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED = 0x05
-    SPECTRUM_DATA_8BIT = 0x10
-    SPECTRUM_DATA_16BIT = 0x11
-    SPECTRUM_DATA_32BIT = 0x12
-    SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x13
-    SEGMENTATION_SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x14
-    HF_TUNING_INDICATOR_DATA = 0x20
-    HF_SCAN_CHANNEL_FOUND_DATA = 0x22
-    HF_SCAN_FREQUENCY_FOUND_DATA = 0x23
-    HF_SCAN_SWEEP_RESTARTED_DATA = 0x24
-    HF_DEMODULATION_SYMBOL_STREAM_DATA = 0x30
-    HF_DECODER_TEXT_DATA = 0x40
-    HF_SPECTRUM_VISUALIZATION_DATA = 0x50
-    HF_TIMEDOMAIN_VISUALIZATION_DATA = 0x51
-    AUDIO_DATA = 0x100
-    EMISSION_LIST_DATA = 0x110
-    LEVEL_DATA = 0x120
-    SYMBOL_DATA = 0x130
-    INSTANTANEOUS_DATA = 0x140
-    BURST_EMMISION_LIST = 0x150
-    IMAGE_DATA = 0x160
-    TRANSMISSION_SYSTEM_RESULT_DATA = 0x170
-    PULSE_DESCRIPTION_WORD_DATA = 0x200
-    PULSE_REPETITION_WORD_DATA = 0x210
-    EM050_SCAN_DATA = 0x4000
-    SCAN_LEVEL = 0x4001
-    SCAN_TUNING = 0x4002
-    SCAN_LEVEL_TUNING = 0x4003
-    DDF_RESERVED_START = 0x5000
-    DDF_RESERVED_END = 0x50FF
-
-

Ancestors

-
    -
  • builtins.int
    • -
  • enum.Enum
    • -
-

Class variables

-
-
var AUDIO_DATA
-
-
-
-
var BURST_EMMISION_LIST
-
-
-
-
var DDF_RESERVED_END
-
-
-
-
var DDF_RESERVED_START
-
-
-
-
var EM050_SCAN_DATA
-
-
-
-
var EMISSION_LIST_DATA
-
-
-
-
var HF_DECODER_TEXT_DATA
-
-
-
-
var HF_DEMODULATION_SYMBOL_STREAM_DATA
-
-
-
-
var HF_SCAN_CHANNEL_FOUND_DATA
-
-
-
-
var HF_SCAN_FREQUENCY_FOUND_DATA
-
-
-
-
var HF_SCAN_SWEEP_RESTARTED_DATA
-
-
-
-
var HF_SPECTRUM_VISUALIZATION_DATA
-
-
-
-
var HF_TIMEDOMAIN_VISUALIZATION_DATA
-
-
-
-
var HF_TUNING_INDICATOR_DATA
-
-
-
-
var IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT
-
-
-
-
var IF_DATA_16BIT_REAL_REAL
-
-
-
-
var IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT
-
-
-
-
var IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED
-
-
-
-
var IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED
-
-
-
-
var IMAGE_DATA
-
-
-
-
var INSTANTANEOUS_DATA
-
-
-
-
var LEVEL_DATA
-
-
-
-
var PULSE_DESCRIPTION_WORD_DATA
-
-
-
-
var PULSE_REPETITION_WORD_DATA
-
-
-
-
var SCAN_LEVEL
-
-
-
-
var SCAN_LEVEL_TUNING
-
-
-
-
var SCAN_TUNING
-
-
-
-
var SEGMENTATION_SPECTRUM_DATA_32BIT_FLOATINGPOINT
-
-
-
-
var SPECTRUM_DATA_16BIT
-
-
-
-
var SPECTRUM_DATA_32BIT
-
-
-
-
var SPECTRUM_DATA_32BIT_FLOATINGPOINT
-
-
-
-
var SPECTRUM_DATA_8BIT
-
-
-
-
var SYMBOL_DATA
-
-
-
-
var TEST_DATA
-
-
-
-
var TRANSMISSION_SYSTEM_RESULT_DATA
-
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosContainer.html =================================================================== --- doc/ammosreader/AmmosContainer.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,358 +1,0 @@ - - - - - - -ammosreader.AmmosContainer API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosContainer

-
-
-

I provide an Ammos container.

-
- -Expand source code - -
"""I provide an Ammos container."""
-
-
-class AmmosContainer():
-    """I store the content of an R&S Ammos file in a more accessible way."""
-    def __init__(self, name, frames):
-        self.__name = name
-        self.__global_frames = frames
-        self.__tags = []
-
-    @property
-    def name(self):
-        return self.__name
-
-    @name.setter
-    def name(self, a_name):
-        self.__name = a_name
-
-    @property
-    def global_frames(self):
-        return self.__global_frames
-
-    def add_tag(self, tag):
-        self.__tags.append(tag)
-
-    def add_frame(self, frame):
-        self.__global_frames.append(frame)
-
-    def size(self):
-        # return sum([each.global_frame_header.frame_length for each in self.__global_frames])
-        return sum(each.global_frame_header.frame_length for each in self.__global_frames)
-
-    def frequencies(self):
-        """I return a list of unique frequencies inside this container."""
-        return list({each.global_frame_body.data_header.frequency for each in self.__global_frames})
-
-    def frame_types(self):
-        return list({each.global_frame_header.frame_type for each in self.__global_frames})
-
-    def frame_sizes(self):
-        return [each.global_frame_header.frame_length for each in self.__global_frames]
-
-    def unique_frame_sizes(self):
-        return list({each.global_frame_header.frame_length for each in self.__global_frames})
-
-    def is_homogenic(self):
-        return (len(self.unique_frame_sizes()) == 1) and (len(self.frame_types()) == 1)
-
-    def __str__(self):
-        start_time = self.__global_frames[0].global_frame_body.data_header.timestamp
-        end_time = self.__global_frames[-1].global_frame_body.data_header.timestamp
-
-        frq = str(self.__global_frames[0].global_frame_body.data_header.frequency)
-
-        return ("Start time: " + str(start_time) +
-                "\nEnd time  : " + str(end_time) + "\nFrequencies: " + frq)
-
-    def data_only(self):
-        return [each.global_frame_body.data_body.data for each in self.__global_frames]
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosContainer -(name, frames) -
-
-

I store the content of an R&S Ammos file in a more accessible way.

-
- -Expand source code - -
class AmmosContainer():
-    """I store the content of an R&S Ammos file in a more accessible way."""
-    def __init__(self, name, frames):
-        self.__name = name
-        self.__global_frames = frames
-        self.__tags = []
-
-    @property
-    def name(self):
-        return self.__name
-
-    @name.setter
-    def name(self, a_name):
-        self.__name = a_name
-
-    @property
-    def global_frames(self):
-        return self.__global_frames
-
-    def add_tag(self, tag):
-        self.__tags.append(tag)
-
-    def add_frame(self, frame):
-        self.__global_frames.append(frame)
-
-    def size(self):
-        # return sum([each.global_frame_header.frame_length for each in self.__global_frames])
-        return sum(each.global_frame_header.frame_length for each in self.__global_frames)
-
-    def frequencies(self):
-        """I return a list of unique frequencies inside this container."""
-        return list({each.global_frame_body.data_header.frequency for each in self.__global_frames})
-
-    def frame_types(self):
-        return list({each.global_frame_header.frame_type for each in self.__global_frames})
-
-    def frame_sizes(self):
-        return [each.global_frame_header.frame_length for each in self.__global_frames]
-
-    def unique_frame_sizes(self):
-        return list({each.global_frame_header.frame_length for each in self.__global_frames})
-
-    def is_homogenic(self):
-        return (len(self.unique_frame_sizes()) == 1) and (len(self.frame_types()) == 1)
-
-    def __str__(self):
-        start_time = self.__global_frames[0].global_frame_body.data_header.timestamp
-        end_time = self.__global_frames[-1].global_frame_body.data_header.timestamp
-
-        frq = str(self.__global_frames[0].global_frame_body.data_header.frequency)
-
-        return ("Start time: " + str(start_time) +
-                "\nEnd time  : " + str(end_time) + "\nFrequencies: " + frq)
-
-    def data_only(self):
-        return [each.global_frame_body.data_body.data for each in self.__global_frames]
-
-

Instance variables

-
-
var global_frames
-
-
-
- -Expand source code - -
@property
-def global_frames(self):
-    return self.__global_frames
-
-
-
var name
-
-
-
- -Expand source code - -
@property
-def name(self):
-    return self.__name
-
-
-
-

Methods

-
-
-def add_frame(self, frame) -
-
-
-
- -Expand source code - -
def add_frame(self, frame):
-    self.__global_frames.append(frame)
-
-
-
-def add_tag(self, tag) -
-
-
-
- -Expand source code - -
def add_tag(self, tag):
-    self.__tags.append(tag)
-
-
-
-def data_only(self) -
-
-
-
- -Expand source code - -
def data_only(self):
-    return [each.global_frame_body.data_body.data for each in self.__global_frames]
-
-
-
-def frame_sizes(self) -
-
-
-
- -Expand source code - -
def frame_sizes(self):
-    return [each.global_frame_header.frame_length for each in self.__global_frames]
-
-
-
-def frame_types(self) -
-
-
-
- -Expand source code - -
def frame_types(self):
-    return list({each.global_frame_header.frame_type for each in self.__global_frames})
-
-
-
-def frequencies(self) -
-
-

I return a list of unique frequencies inside this container.

-
- -Expand source code - -
def frequencies(self):
-    """I return a list of unique frequencies inside this container."""
-    return list({each.global_frame_body.data_header.frequency for each in self.__global_frames})
-
-
-
-def is_homogenic(self) -
-
-
-
- -Expand source code - -
def is_homogenic(self):
-    return (len(self.unique_frame_sizes()) == 1) and (len(self.frame_types()) == 1)
-
-
-
-def size(self) -
-
-
-
- -Expand source code - -
def size(self):
-    # return sum([each.global_frame_header.frame_length for each in self.__global_frames])
-    return sum(each.global_frame_header.frame_length for each in self.__global_frames)
-
-
-
-def unique_frame_sizes(self) -
-
-
-
- -Expand source code - -
def unique_frame_sizes(self):
-    return list({each.global_frame_header.frame_length for each in self.__global_frames})
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosExtendedAudioDataHeader.html =================================================================== --- doc/ammosreader/AmmosExtendedAudioDataHeader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,197 +1,0 @@ - - - - - - -ammosreader.AmmosExtendedAudioDataHeader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosExtendedAudioDataHeader

-
-
-

I provide an Ammos extended data header for audio data frames.

-
- -Expand source code - -
"""I provide an Ammos extended data header for audio data frames."""
-
-import struct
-from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
-
-
-class AmmosExtendedAudioDataHeader:
-
-    HEADER_SIZE = 44  # 11 words
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return a new AMMOS extended data header for audio frames built from given bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-        standard_header = AmmosAudioDataHeader.from_bytes(in_bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
-        extended_header_elements = struct.unpack('<Q', in_bytes[AmmosAudioDataHeader.HEADER_SIZE:])
-        timestamp = extended_header_elements[0]
-        sample_rate = standard_header.sample_rate
-        status = standard_header.status
-        frequency = standard_header.frequency
-        demod_bandwidth = standard_header.demod_bandwidth
-        demod_type = standard_header.demod_type
-        number_of_samples = standard_header.number_of_samples
-        number_of_channels = standard_header.number_of_channels
-        sample_size = standard_header.sample_size
-        return AmmosExtendedAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
-                                            number_of_samples, number_of_channels, sample_size, timestamp)
-
-    def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type,
-                 number_of_samples, number_of_channels, sample_size, timestamp):
-        """I return a new AMMOS extended data header for audio frames built from given parameters."""
-        self.sample_rate = sample_rate
-        self.status = status
-        self.frequency = frequency
-        self.demod_bandwidth = demod_bandwidth
-        self.demod_type = demod_type
-        self.number_of_samples = number_of_samples
-        self.number_of_channels = number_of_channels
-        self.sample_size = sample_size
-        self.timestamp = timestamp
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosExtendedAudioDataHeader -(sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples, number_of_channels, sample_size, timestamp) -
-
-

I return a new AMMOS extended data header for audio frames built from given parameters.

-
- -Expand source code - -
class AmmosExtendedAudioDataHeader:
-
-    HEADER_SIZE = 44  # 11 words
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return a new AMMOS extended data header for audio frames built from given bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-        standard_header = AmmosAudioDataHeader.from_bytes(in_bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
-        extended_header_elements = struct.unpack('<Q', in_bytes[AmmosAudioDataHeader.HEADER_SIZE:])
-        timestamp = extended_header_elements[0]
-        sample_rate = standard_header.sample_rate
-        status = standard_header.status
-        frequency = standard_header.frequency
-        demod_bandwidth = standard_header.demod_bandwidth
-        demod_type = standard_header.demod_type
-        number_of_samples = standard_header.number_of_samples
-        number_of_channels = standard_header.number_of_channels
-        sample_size = standard_header.sample_size
-        return AmmosExtendedAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
-                                            number_of_samples, number_of_channels, sample_size, timestamp)
-
-    def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type,
-                 number_of_samples, number_of_channels, sample_size, timestamp):
-        """I return a new AMMOS extended data header for audio frames built from given parameters."""
-        self.sample_rate = sample_rate
-        self.status = status
-        self.frequency = frequency
-        self.demod_bandwidth = demod_bandwidth
-        self.demod_type = demod_type
-        self.number_of_samples = number_of_samples
-        self.number_of_channels = number_of_channels
-        self.sample_size = sample_size
-        self.timestamp = timestamp
-
-

Class variables

-
-
var HEADER_SIZE
-
-
-
-
-

Static methods

-
-
-def from_bytes(in_bytes) -
-
-

I return a new AMMOS extended data header for audio frames built from given bytes.

-
- -Expand source code - -
@classmethod
-def from_bytes(cls, in_bytes):
-    """I return a new AMMOS extended data header for audio frames built from given bytes."""
-    assert len(in_bytes) == cls.HEADER_SIZE
-    standard_header = AmmosAudioDataHeader.from_bytes(in_bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
-    extended_header_elements = struct.unpack('<Q', in_bytes[AmmosAudioDataHeader.HEADER_SIZE:])
-    timestamp = extended_header_elements[0]
-    sample_rate = standard_header.sample_rate
-    status = standard_header.status
-    frequency = standard_header.frequency
-    demod_bandwidth = standard_header.demod_bandwidth
-    demod_type = standard_header.demod_type
-    number_of_samples = standard_header.number_of_samples
-    number_of_channels = standard_header.number_of_channels
-    sample_size = standard_header.sample_size
-    return AmmosExtendedAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
-                                        number_of_samples, number_of_channels, sample_size, timestamp)
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosExtendedIFDataHeader.html =================================================================== --- doc/ammosreader/AmmosExtendedIFDataHeader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,268 +1,0 @@ - - - - - - -ammosreader.AmmosExtendedIFDataHeader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosExtendedIFDataHeader

-
-
-
- -Expand source code - -
import struct
-import numpy as np
-from ammosreader.AmmosIFDataHeader import AmmosIFDataHeader
-
-
-class AmmosExtendedIFDataHeader():
-    """I implement an Ammos extended data header for IF data frames."""
-
-    HEADER_SIZE = 76
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return an AMMOS extended data header from given bytes."""
-        standard_header = AmmosIFDataHeader.from_bytes(in_bytes[0:AmmosIFDataHeader.HEADER_SIZE])
-        extended_header_elements = struct.unpack('<QQI', in_bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
-        block_count = standard_header.block_count
-        block_length = standard_header.block_length
-        timestamp = standard_header.timestamp
-        status = standard_header.status
-        source_id = standard_header.source_id
-        source_state = standard_header.source_state
-        frequency = standard_header.frequency
-        bandwidth = standard_header.bandwidth
-        sample_rate = standard_header.sample_rate
-        interpolation = standard_header.interpolation
-        decimation = standard_header.decimation
-        voltage_ref = standard_header.voltage_ref
-        stream_start = np.datetime64(int(extended_header_elements[0]), 'ns')
-        sample_counter = extended_header_elements[1]
-        antenna_correction = extended_header_elements[2]
-        size = len(in_bytes)
-        return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
-                                         source_state, frequency, bandwidth, sample_rate, interpolation,
-                                         decimation, voltage_ref, stream_start, sample_counter,
-                                         antenna_correction)
-
-    def __init__(self, size, block_count, block_length, timestamp, status, source_id, source_state, frequency,
-                 bandwidth, sample_rate, interpolation, decimation, voltage_ref, stream_start, sample_counter,
-                 antenna_correction):
-        """I create a new instance of myself using the above parameters."""
-        self.size = size
-        self.block_count = block_count
-        self.block_length = block_length
-        self.timestamp = timestamp
-        self.status = status
-        self.source_id = source_id
-        self.source_state = source_state
-        self.frequency = frequency
-        self.bandwidth = bandwidth
-        self.sample_rate = sample_rate
-        self.interpolation = interpolation
-        self.decimation = decimation
-        self.voltage_ref = voltage_ref
-        self.stream_start = stream_start
-        self.sample_counter = sample_counter
-        self.antenna_correction = antenna_correction
-
-    def __str__(self):
-        output = (
-            "\nGlobal frame body data header\n" +
-            "-----------------------------\n" +
-            "Block count:" + str(self.block_count) + "\n" +
-            "Block length:" + str(self.block_length) + "\n" +
-            "Time stamp:" + str(self.timestamp) + "\n" +
-            "Frequency:" + str(self.frequency) + "\n" +
-            "Bandwidth:" + str(self.bandwidth) + "\n" +
-            "Sample rate:" + str(self.sample_rate) + "\n" +
-            "Stream start:" + str(self.stream_start) + "\n" +
-            "Sample counter:" + str(self.sample_counter) + "\n" +
-            "Antenna correction:" + str(self.antenna_correction) + "\n")
-        return output
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosExtendedIFDataHeader -(size, block_count, block_length, timestamp, status, source_id, source_state, frequency, bandwidth, sample_rate, interpolation, decimation, voltage_ref, stream_start, sample_counter, antenna_correction) -
-
-

I implement an Ammos extended data header for IF data frames.

-

I create a new instance of myself using the above parameters.

-
- -Expand source code - -
class AmmosExtendedIFDataHeader():
-    """I implement an Ammos extended data header for IF data frames."""
-
-    HEADER_SIZE = 76
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return an AMMOS extended data header from given bytes."""
-        standard_header = AmmosIFDataHeader.from_bytes(in_bytes[0:AmmosIFDataHeader.HEADER_SIZE])
-        extended_header_elements = struct.unpack('<QQI', in_bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
-        block_count = standard_header.block_count
-        block_length = standard_header.block_length
-        timestamp = standard_header.timestamp
-        status = standard_header.status
-        source_id = standard_header.source_id
-        source_state = standard_header.source_state
-        frequency = standard_header.frequency
-        bandwidth = standard_header.bandwidth
-        sample_rate = standard_header.sample_rate
-        interpolation = standard_header.interpolation
-        decimation = standard_header.decimation
-        voltage_ref = standard_header.voltage_ref
-        stream_start = np.datetime64(int(extended_header_elements[0]), 'ns')
-        sample_counter = extended_header_elements[1]
-        antenna_correction = extended_header_elements[2]
-        size = len(in_bytes)
-        return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
-                                         source_state, frequency, bandwidth, sample_rate, interpolation,
-                                         decimation, voltage_ref, stream_start, sample_counter,
-                                         antenna_correction)
-
-    def __init__(self, size, block_count, block_length, timestamp, status, source_id, source_state, frequency,
-                 bandwidth, sample_rate, interpolation, decimation, voltage_ref, stream_start, sample_counter,
-                 antenna_correction):
-        """I create a new instance of myself using the above parameters."""
-        self.size = size
-        self.block_count = block_count
-        self.block_length = block_length
-        self.timestamp = timestamp
-        self.status = status
-        self.source_id = source_id
-        self.source_state = source_state
-        self.frequency = frequency
-        self.bandwidth = bandwidth
-        self.sample_rate = sample_rate
-        self.interpolation = interpolation
-        self.decimation = decimation
-        self.voltage_ref = voltage_ref
-        self.stream_start = stream_start
-        self.sample_counter = sample_counter
-        self.antenna_correction = antenna_correction
-
-    def __str__(self):
-        output = (
-            "\nGlobal frame body data header\n" +
-            "-----------------------------\n" +
-            "Block count:" + str(self.block_count) + "\n" +
-            "Block length:" + str(self.block_length) + "\n" +
-            "Time stamp:" + str(self.timestamp) + "\n" +
-            "Frequency:" + str(self.frequency) + "\n" +
-            "Bandwidth:" + str(self.bandwidth) + "\n" +
-            "Sample rate:" + str(self.sample_rate) + "\n" +
-            "Stream start:" + str(self.stream_start) + "\n" +
-            "Sample counter:" + str(self.sample_counter) + "\n" +
-            "Antenna correction:" + str(self.antenna_correction) + "\n")
-        return output
-
-

Class variables

-
-
var HEADER_SIZE
-
-
-
-
-

Static methods

-
-
-def from_bytes(in_bytes) -
-
-

I return an AMMOS extended data header from given bytes.

-
- -Expand source code - -
@classmethod
-def from_bytes(cls, in_bytes):
-    """I return an AMMOS extended data header from given bytes."""
-    standard_header = AmmosIFDataHeader.from_bytes(in_bytes[0:AmmosIFDataHeader.HEADER_SIZE])
-    extended_header_elements = struct.unpack('<QQI', in_bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
-    block_count = standard_header.block_count
-    block_length = standard_header.block_length
-    timestamp = standard_header.timestamp
-    status = standard_header.status
-    source_id = standard_header.source_id
-    source_state = standard_header.source_state
-    frequency = standard_header.frequency
-    bandwidth = standard_header.bandwidth
-    sample_rate = standard_header.sample_rate
-    interpolation = standard_header.interpolation
-    decimation = standard_header.decimation
-    voltage_ref = standard_header.voltage_ref
-    stream_start = np.datetime64(int(extended_header_elements[0]), 'ns')
-    sample_counter = extended_header_elements[1]
-    antenna_correction = extended_header_elements[2]
-    size = len(in_bytes)
-    return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
-                                     source_state, frequency, bandwidth, sample_rate, interpolation,
-                                     decimation, voltage_ref, stream_start, sample_counter,
-                                     antenna_correction)
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosGlobalFrameBody.html =================================================================== --- doc/ammosreader/AmmosGlobalFrameBody.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,197 +1,0 @@ - - - - - - -ammosreader.AmmosGlobalFrameBody API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosGlobalFrameBody

-
-
-

I provide an AMMOS global frame body.

-
- -Expand source code - -
"""I provide an AMMOS global frame body."""
-
-class AmmosGlobalFrameBody():
-    """
-    I implement an AMMOS global frame body.
-
-    AMMOS frames can store data of various types.
-    The data is described in the data header.
-    The raw data is then stored in the data body.
-    """
-    def __init__(self, data_header, data_body):
-        self.__data_header = data_header
-        self.__data_body = data_body
-
-    @property
-    def data_header(self):
-        return self.__data_header
-
-    @data_header.setter
-    def data_header(self, a_data_header):
-        self.__data_header = a_data_header
-
-    @property
-    def data_body(self):
-        return self.__data_body
-
-    @data_body.setter
-    def data_body(self, data_bytes):
-        self.__data_body = data_bytes
-
-    def payload(self):
-        """I return the payload only."""
-        return b"".join([each_block.data for each_block in self.data_body])
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosGlobalFrameBody -(data_header, data_body) -
-
-

I implement an AMMOS global frame body.

-

AMMOS frames can store data of various types. -The data is described in the data header. -The raw data is then stored in the data body.

-
- -Expand source code - -
class AmmosGlobalFrameBody():
-    """
-    I implement an AMMOS global frame body.
-
-    AMMOS frames can store data of various types.
-    The data is described in the data header.
-    The raw data is then stored in the data body.
-    """
-    def __init__(self, data_header, data_body):
-        self.__data_header = data_header
-        self.__data_body = data_body
-
-    @property
-    def data_header(self):
-        return self.__data_header
-
-    @data_header.setter
-    def data_header(self, a_data_header):
-        self.__data_header = a_data_header
-
-    @property
-    def data_body(self):
-        return self.__data_body
-
-    @data_body.setter
-    def data_body(self, data_bytes):
-        self.__data_body = data_bytes
-
-    def payload(self):
-        """I return the payload only."""
-        return b"".join([each_block.data for each_block in self.data_body])
-
-

Instance variables

-
-
var data_body
-
-
-
- -Expand source code - -
@property
-def data_body(self):
-    return self.__data_body
-
-
-
var data_header
-
-
-
- -Expand source code - -
@property
-def data_header(self):
-    return self.__data_header
-
-
-
-

Methods

-
-
-def payload(self) -
-
-

I return the payload only.

-
- -Expand source code - -
def payload(self):
-    """I return the payload only."""
-    return b"".join([each_block.data for each_block in self.data_body])
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosGlobalFrameHeader.html =================================================================== --- doc/ammosreader/AmmosGlobalFrameHeader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,251 +1,0 @@ - - - - - - -ammosreader.AmmosGlobalFrameHeader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosGlobalFrameHeader

-
-
-

I provide an AMMOS global frame header.

-
- -Expand source code - -
"""I provide an AMMOS global frame header."""
-import struct
-from ammosreader import logger
-
-class AmmosGlobalFrameHeader:
-    """I implement an AMMOS global frame header."""
-
-    MAGIC_WORD = "726574fb"
-    HEADER_SIZE = 24
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I create a new AmmosGlobalFrameHeader from bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-
-        elements = struct.unpack('<4s4s4s4s4s4s', in_bytes)
-
-        magic_word = elements[0].hex()
-
-        logger.debug("Header created")
-        if magic_word != cls.MAGIC_WORD:
-            return None
-
-        frame_length = (int.from_bytes(elements[1], byteorder='little')*4)
-
-        running_frame_number = (int.from_bytes(elements[2], byteorder='little'))
-
-        frame_type = (int.from_bytes(elements[3], byteorder='little'))
-
-        data_header_length = 4 * int.from_bytes(elements[4], byteorder='little')
-
-        reserved = elements[5]
-
-        return AmmosGlobalFrameHeader(magic_word, frame_length, running_frame_number,
-                                      frame_type, data_header_length, reserved)
-
-    def __init__(self, magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved):
-        """I return a new instance of myself initialized with above parameters."""
-        if magic_word != type(self).MAGIC_WORD:
-            logger.error("Wrong magic word found")
-            self.magic_word = magic_word
-        else:
-            self.magic_word = magic_word
-            self.frame_length = frame_length
-            self.running_frame_number = running_frame_number
-            self.frame_type = frame_type
-            self.data_header_length = data_header_length
-            self.reserved = reserved
-
-    def __str__(self):
-        """I return the string representation of myself."""
-        output = ("Global frame header info\n" +
-                  "------------------------\n" +
-                  "Magic word:" + str(self.magic_word) + "\n" +
-                  "Frame length:" + str(self.frame_length) + "\n" +
-                  "Running frame:" + str(self.running_frame_number) + "\n" +
-                  "Frame Type:" + str(self.frame_type) + "\n" +
-                  "Data header length:" + str(self.data_header_length) + "\n")
-        return output
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosGlobalFrameHeader -(magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved) -
-
-

I implement an AMMOS global frame header.

-

I return a new instance of myself initialized with above parameters.

-
- -Expand source code - -
class AmmosGlobalFrameHeader:
-    """I implement an AMMOS global frame header."""
-
-    MAGIC_WORD = "726574fb"
-    HEADER_SIZE = 24
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I create a new AmmosGlobalFrameHeader from bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-
-        elements = struct.unpack('<4s4s4s4s4s4s', in_bytes)
-
-        magic_word = elements[0].hex()
-
-        logger.debug("Header created")
-        if magic_word != cls.MAGIC_WORD:
-            return None
-
-        frame_length = (int.from_bytes(elements[1], byteorder='little')*4)
-
-        running_frame_number = (int.from_bytes(elements[2], byteorder='little'))
-
-        frame_type = (int.from_bytes(elements[3], byteorder='little'))
-
-        data_header_length = 4 * int.from_bytes(elements[4], byteorder='little')
-
-        reserved = elements[5]
-
-        return AmmosGlobalFrameHeader(magic_word, frame_length, running_frame_number,
-                                      frame_type, data_header_length, reserved)
-
-    def __init__(self, magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved):
-        """I return a new instance of myself initialized with above parameters."""
-        if magic_word != type(self).MAGIC_WORD:
-            logger.error("Wrong magic word found")
-            self.magic_word = magic_word
-        else:
-            self.magic_word = magic_word
-            self.frame_length = frame_length
-            self.running_frame_number = running_frame_number
-            self.frame_type = frame_type
-            self.data_header_length = data_header_length
-            self.reserved = reserved
-
-    def __str__(self):
-        """I return the string representation of myself."""
-        output = ("Global frame header info\n" +
-                  "------------------------\n" +
-                  "Magic word:" + str(self.magic_word) + "\n" +
-                  "Frame length:" + str(self.frame_length) + "\n" +
-                  "Running frame:" + str(self.running_frame_number) + "\n" +
-                  "Frame Type:" + str(self.frame_type) + "\n" +
-                  "Data header length:" + str(self.data_header_length) + "\n")
-        return output
-
-

Class variables

-
-
var HEADER_SIZE
-
-
-
-
var MAGIC_WORD
-
-
-
-
-

Static methods

-
-
-def from_bytes(in_bytes) -
-
-

I create a new AmmosGlobalFrameHeader from bytes.

-
- -Expand source code - -
@classmethod
-def from_bytes(cls, in_bytes):
-    """I create a new AmmosGlobalFrameHeader from bytes."""
-    assert len(in_bytes) == cls.HEADER_SIZE
-
-    elements = struct.unpack('<4s4s4s4s4s4s', in_bytes)
-
-    magic_word = elements[0].hex()
-
-    logger.debug("Header created")
-    if magic_word != cls.MAGIC_WORD:
-        return None
-
-    frame_length = (int.from_bytes(elements[1], byteorder='little')*4)
-
-    running_frame_number = (int.from_bytes(elements[2], byteorder='little'))
-
-    frame_type = (int.from_bytes(elements[3], byteorder='little'))
-
-    data_header_length = 4 * int.from_bytes(elements[4], byteorder='little')
-
-    reserved = elements[5]
-
-    return AmmosGlobalFrameHeader(magic_word, frame_length, running_frame_number,
-                                  frame_type, data_header_length, reserved)
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosIFDataBlock.html =================================================================== --- doc/ammosreader/AmmosIFDataBlock.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,137 +1,0 @@ - - - - - - -ammosreader.AmmosIFDataBlock API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosIFDataBlock

-
-
-
- -Expand source code - -
class AmmosIFDataBlock:
-
-    def __init__(self, header, data):
-        self.__header = header
-        self.__data = data
-
-    @property
-    def header(self):
-        return self.__header
-
-    @property
-    def payload(self):
-        return self.__data
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosIFDataBlock -(header, data) -
-
-
-
- -Expand source code - -
class AmmosIFDataBlock:
-
-    def __init__(self, header, data):
-        self.__header = header
-        self.__data = data
-
-    @property
-    def header(self):
-        return self.__header
-
-    @property
-    def payload(self):
-        return self.__data
-
-

Instance variables

-
-
var header
-
-
-
- -Expand source code - -
@property
-def header(self):
-    return self.__header
-
-
-
var payload
-
-
-
- -Expand source code - -
@property
-def payload(self):
-    return self.__data
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosIFDataBlockHeader.html =================================================================== --- doc/ammosreader/AmmosIFDataBlockHeader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,325 +1,0 @@ - - - - - - -ammosreader.AmmosIFDataBlockHeader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosIFDataBlockHeader

-
-
-
- -Expand source code - -
import struct
-
-
-class AmmosIFDataBlockHeader:
-    """I implement an AMMOS IF data block header."""
-
-    HEADER_SIZE = 4
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return an AMMOS IF data block header built from bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-        elements = struct.unpack('<ccH', in_bytes)
-        header = cls()
-        first_entry = int.from_bytes(elements[0], byteorder='little')
-        header.invalidity = ((first_entry & 1) == 1)
-        header.blanking = ((first_entry & 2) == 1)
-        header.user_data = ((first_entry & 252))
-        header.reserved = int.from_bytes(elements[1], byteorder='little')
-        header.reciprocal_gain = elements[2]
-        return header
-
-    def __init__(self):
-        """I return an AMMOS IF data block header initialized with sane default values."""
-        self.__invalidity = True
-        self.__blanking = False
-        self.__user_data = 0
-        self.__reserved = 0
-        self.__reciprocal_gain = 0
-
-    @property
-    def invalidity(self):
-        return self.__invalidity
-
-    @invalidity.setter
-    def invalidity(self, a_boolean):
-        self.__invalidity = a_boolean
-
-    @property
-    def blanking(self):
-        return self.__blanking
-
-    @blanking.setter
-    def blanking(self, a_boolean):
-        self.__blanking = a_boolean
-
-    @property
-    def user_data(self):
-        return self.__user_data
-
-    @user_data.setter
-    def user_data(self, some_data):
-        self.__user_data = some_data
-
-    @property
-    def reserved(self):
-        return self.__reserved
-
-    @reserved.setter
-    def reserved(self, some_data):
-        assert some_data == 0
-        self.__reserved = some_data
-
-    @property
-    def reciprocal_gain(self):
-        return self.__reciprocal_gain
-
-    @reciprocal_gain.setter
-    def reciprocal_gain(self, a_gain_value):
-        assert 0 <= a_gain_value < pow(2, 16)
-        self.__reciprocal_gain = a_gain_value
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosIFDataBlockHeader -
-
-

I implement an AMMOS IF data block header.

-

I return an AMMOS IF data block header initialized with sane default values.

-
- -Expand source code - -
class AmmosIFDataBlockHeader:
-    """I implement an AMMOS IF data block header."""
-
-    HEADER_SIZE = 4
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return an AMMOS IF data block header built from bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-        elements = struct.unpack('<ccH', in_bytes)
-        header = cls()
-        first_entry = int.from_bytes(elements[0], byteorder='little')
-        header.invalidity = ((first_entry & 1) == 1)
-        header.blanking = ((first_entry & 2) == 1)
-        header.user_data = ((first_entry & 252))
-        header.reserved = int.from_bytes(elements[1], byteorder='little')
-        header.reciprocal_gain = elements[2]
-        return header
-
-    def __init__(self):
-        """I return an AMMOS IF data block header initialized with sane default values."""
-        self.__invalidity = True
-        self.__blanking = False
-        self.__user_data = 0
-        self.__reserved = 0
-        self.__reciprocal_gain = 0
-
-    @property
-    def invalidity(self):
-        return self.__invalidity
-
-    @invalidity.setter
-    def invalidity(self, a_boolean):
-        self.__invalidity = a_boolean
-
-    @property
-    def blanking(self):
-        return self.__blanking
-
-    @blanking.setter
-    def blanking(self, a_boolean):
-        self.__blanking = a_boolean
-
-    @property
-    def user_data(self):
-        return self.__user_data
-
-    @user_data.setter
-    def user_data(self, some_data):
-        self.__user_data = some_data
-
-    @property
-    def reserved(self):
-        return self.__reserved
-
-    @reserved.setter
-    def reserved(self, some_data):
-        assert some_data == 0
-        self.__reserved = some_data
-
-    @property
-    def reciprocal_gain(self):
-        return self.__reciprocal_gain
-
-    @reciprocal_gain.setter
-    def reciprocal_gain(self, a_gain_value):
-        assert 0 <= a_gain_value < pow(2, 16)
-        self.__reciprocal_gain = a_gain_value
-
-

Class variables

-
-
var HEADER_SIZE
-
-
-
-
-

Static methods

-
-
-def from_bytes(in_bytes) -
-
-

I return an AMMOS IF data block header built from bytes.

-
- -Expand source code - -
@classmethod
-def from_bytes(cls, in_bytes):
-    """I return an AMMOS IF data block header built from bytes."""
-    assert len(in_bytes) == cls.HEADER_SIZE
-    elements = struct.unpack('<ccH', in_bytes)
-    header = cls()
-    first_entry = int.from_bytes(elements[0], byteorder='little')
-    header.invalidity = ((first_entry & 1) == 1)
-    header.blanking = ((first_entry & 2) == 1)
-    header.user_data = ((first_entry & 252))
-    header.reserved = int.from_bytes(elements[1], byteorder='little')
-    header.reciprocal_gain = elements[2]
-    return header
-
-
-
-

Instance variables

-
-
var blanking
-
-
-
- -Expand source code - -
@property
-def blanking(self):
-    return self.__blanking
-
-
-
var invalidity
-
-
-
- -Expand source code - -
@property
-def invalidity(self):
-    return self.__invalidity
-
-
-
var reciprocal_gain
-
-
-
- -Expand source code - -
@property
-def reciprocal_gain(self):
-    return self.__reciprocal_gain
-
-
-
var reserved
-
-
-
- -Expand source code - -
@property
-def reserved(self):
-    return self.__reserved
-
-
-
var user_data
-
-
-
- -Expand source code - -
@property
-def user_data(self):
-    return self.__user_data
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosIFDataBody.html =================================================================== --- doc/ammosreader/AmmosIFDataBody.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,172 +1,0 @@ - - - - - - -ammosreader.AmmosIFDataBody API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosIFDataBody

-
-
-

I provide an AMMOS data body for IF data.

-

I manage multiple IF data blocks

-
- -Expand source code - -
"""
-I provide an AMMOS data body for IF data.
-
-I manage multiple IF data blocks
-"""
-
-class AmmosIFDataBody:
-    """I implement an AMMOS data body for IF data."""
-    def __init__(self, data_blocks=None):
-        if data_blocks is None:
-            self.__data_blocks = []
-        else:
-            self.__data_blocks = data_blocks
-
-    @property
-    def data_blocks(self):
-        return self.__data_blocks
-
-    def add_data_block(self, a_data_block):
-        self.__data_blocks.append(a_data_block)
-
-    @property
-    def payload(self):
-        return b"".join([each.payload for each in self.data_blocks])
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosIFDataBody -(data_blocks=None) -
-
-

I implement an AMMOS data body for IF data.

-
- -Expand source code - -
class AmmosIFDataBody:
-    """I implement an AMMOS data body for IF data."""
-    def __init__(self, data_blocks=None):
-        if data_blocks is None:
-            self.__data_blocks = []
-        else:
-            self.__data_blocks = data_blocks
-
-    @property
-    def data_blocks(self):
-        return self.__data_blocks
-
-    def add_data_block(self, a_data_block):
-        self.__data_blocks.append(a_data_block)
-
-    @property
-    def payload(self):
-        return b"".join([each.payload for each in self.data_blocks])
-
-

Instance variables

-
-
var data_blocks
-
-
-
- -Expand source code - -
@property
-def data_blocks(self):
-    return self.__data_blocks
-
-
-
var payload
-
-
-
- -Expand source code - -
@property
-def payload(self):
-    return b"".join([each.payload for each in self.data_blocks])
-
-
-
-

Methods

-
-
-def add_data_block(self, a_data_block) -
-
-
-
- -Expand source code - -
def add_data_block(self, a_data_block):
-    self.__data_blocks.append(a_data_block)
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosIFDataHeader.html =================================================================== --- doc/ammosreader/AmmosIFDataHeader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,240 +1,0 @@ - - - - - - -ammosreader.AmmosIFDataHeader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosIFDataHeader

-
-
-

I provide a Ammos data header for IF data frames.

-
- -Expand source code - -
"""I provide a Ammos data header for IF data frames."""
-
-import struct
-import numpy as np
-
-
-class AmmosIFDataHeader():
-    """I implement an Ammos data header for IF data frames."""
-
-    HEADER_SIZE = 56
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return an AMMOS data header from given bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-        elements = struct.unpack('<IIQIIIQIIIIi', in_bytes)
-        block_count = elements[0]
-        block_length = int(elements[1])*4
-        timestamp = np.datetime64(int(elements[2])*1000, 'ns')
-        status = elements[3]
-        source_id = elements[4]
-        source_state = elements[5]
-        frequency = elements[6]
-        bandwidth = elements[7]
-        sample_rate = elements[8]
-        interpolation = elements[9]
-        decimation = elements[10]
-        voltage_ref = elements[11]
-
-        return AmmosIFDataHeader(block_count, block_length, timestamp, status, source_id,
-                                 source_state, frequency, bandwidth, sample_rate,
-                                 interpolation, decimation, voltage_ref)
-
-    def __init__(self, block_count, block_length, timestamp, status, source_id, source_state, frequency,
-                 bandwidth, sample_rate, interpolation, decimation, voltage_ref):
-        """I create a new instance of myself using the above parameters."""
-        self.block_count = block_count
-        self.block_length = block_length
-        self.timestamp = timestamp
-        self.status = status
-        self.source_id = source_id
-        self.source_state = source_state
-        self.frequency = frequency
-        self.bandwidth = bandwidth
-        self.sample_rate = sample_rate
-        self.interpolation = interpolation
-        self.decimation = decimation
-        self.voltage_ref = voltage_ref
-
-    def __str__(self):
-        output = ("\nGlobal frame body data header\n" +
-                  "-----------------------------\n" +
-                  "Block count:" + str(self.block_count) + "\n" +
-                  "Block length:" + str(self.block_length) + "\n" +
-                  "Time stamp:" + str(self.timestamp) + "\n" +
-                  "Frequency:" + str(self.frequency) + "\n" +
-                  "Bandwidth:" + str(self.bandwidth) + "\n" +
-                  "Sample rate:" + str(self.sample_rate) + "\n")
-        return output
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosIFDataHeader -(block_count, block_length, timestamp, status, source_id, source_state, frequency, bandwidth, sample_rate, interpolation, decimation, voltage_ref) -
-
-

I implement an Ammos data header for IF data frames.

-

I create a new instance of myself using the above parameters.

-
- -Expand source code - -
class AmmosIFDataHeader():
-    """I implement an Ammos data header for IF data frames."""
-
-    HEADER_SIZE = 56
-
-    @classmethod
-    def from_bytes(cls, in_bytes):
-        """I return an AMMOS data header from given bytes."""
-        assert len(in_bytes) == cls.HEADER_SIZE
-        elements = struct.unpack('<IIQIIIQIIIIi', in_bytes)
-        block_count = elements[0]
-        block_length = int(elements[1])*4
-        timestamp = np.datetime64(int(elements[2])*1000, 'ns')
-        status = elements[3]
-        source_id = elements[4]
-        source_state = elements[5]
-        frequency = elements[6]
-        bandwidth = elements[7]
-        sample_rate = elements[8]
-        interpolation = elements[9]
-        decimation = elements[10]
-        voltage_ref = elements[11]
-
-        return AmmosIFDataHeader(block_count, block_length, timestamp, status, source_id,
-                                 source_state, frequency, bandwidth, sample_rate,
-                                 interpolation, decimation, voltage_ref)
-
-    def __init__(self, block_count, block_length, timestamp, status, source_id, source_state, frequency,
-                 bandwidth, sample_rate, interpolation, decimation, voltage_ref):
-        """I create a new instance of myself using the above parameters."""
-        self.block_count = block_count
-        self.block_length = block_length
-        self.timestamp = timestamp
-        self.status = status
-        self.source_id = source_id
-        self.source_state = source_state
-        self.frequency = frequency
-        self.bandwidth = bandwidth
-        self.sample_rate = sample_rate
-        self.interpolation = interpolation
-        self.decimation = decimation
-        self.voltage_ref = voltage_ref
-
-    def __str__(self):
-        output = ("\nGlobal frame body data header\n" +
-                  "-----------------------------\n" +
-                  "Block count:" + str(self.block_count) + "\n" +
-                  "Block length:" + str(self.block_length) + "\n" +
-                  "Time stamp:" + str(self.timestamp) + "\n" +
-                  "Frequency:" + str(self.frequency) + "\n" +
-                  "Bandwidth:" + str(self.bandwidth) + "\n" +
-                  "Sample rate:" + str(self.sample_rate) + "\n")
-        return output
-
-

Class variables

-
-
var HEADER_SIZE
-
-
-
-
-

Static methods

-
-
-def from_bytes(in_bytes) -
-
-

I return an AMMOS data header from given bytes.

-
- -Expand source code - -
@classmethod
-def from_bytes(cls, in_bytes):
-    """I return an AMMOS data header from given bytes."""
-    assert len(in_bytes) == cls.HEADER_SIZE
-    elements = struct.unpack('<IIQIIIQIIIIi', in_bytes)
-    block_count = elements[0]
-    block_length = int(elements[1])*4
-    timestamp = np.datetime64(int(elements[2])*1000, 'ns')
-    status = elements[3]
-    source_id = elements[4]
-    source_state = elements[5]
-    frequency = elements[6]
-    bandwidth = elements[7]
-    sample_rate = elements[8]
-    interpolation = elements[9]
-    decimation = elements[10]
-    voltage_ref = elements[11]
-
-    return AmmosIFDataHeader(block_count, block_length, timestamp, status, source_id,
-                             source_state, frequency, bandwidth, sample_rate,
-                             interpolation, decimation, voltage_ref)
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/AmmosIFReader.html =================================================================== --- doc/ammosreader/AmmosIFReader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,460 +1,0 @@ - - - - - - -ammosreader.AmmosIFReader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosIFReader

-
-
-

I provide a specialized Ammos Reader for IF data.

-
- -Expand source code - -
"""I provide a specialized Ammos Reader for IF data."""
-
-from ammosreader.AbstractAmmosReader import AbstractAmmosReader
-from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody
-from ammosreader.AmmosIFDataHeader import AmmosIFDataHeader
-from ammosreader.AmmosExtendedIFDataHeader import AmmosExtendedIFDataHeader
-from ammosreader.AmmosIFDataBody import AmmosIFDataBody
-from ammosreader.AmmosIFDataBlock import AmmosIFDataBlock
-from ammosreader.AmmosIFDataBlockHeader import AmmosIFDataBlockHeader
-from ammosreader import logger
-
-
-class AmmosIFReader(AbstractAmmosReader):
-    """I read the IF data embedded in an R&S AMMOS recording."""
-
-    def read_next_global_frame_body_data_header(self):
-        """I read the next data header of a global frame body from current position in file."""
-        header_size = AmmosIFDataHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-
-        logger.info("\nReading global frame body standard data header\n")
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame body data header", header_size)
-            return None
-        return AmmosIFDataHeader.from_bytes(in_bytes)
-
-    def read_next_global_frame_body_extended_data_header(self):
-        """
-        I return the next global frame body extended data header from current position in file.
-
-        :return: the next Ammos Extended IF data header or None if incomplete
-        :rtype: AmmosExtendedIFDataHeader
-        """
-        header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-
-        logger.info("\nReading global frame body extended data header\n")
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
-            return None
-        return AmmosExtendedIFDataHeader.from_bytes(in_bytes)
-
-    def read_next_if_data_body(self, number_of_data_blocks, data_length):
-        """
-        I return the next data body read from current position in file.
-
-        :param number_of_data_blocks: the number of data blocks inside the body
-        :type number_of_data_blocks: int
-
-        :param data_length: the length of the raw data inside a single block
-        :type data_length: int
-        """
-        header_size = AmmosIFDataBlockHeader.HEADER_SIZE
-
-        data_body = AmmosIFDataBody()
-
-        block_length = header_size + data_length
-
-        total = number_of_data_blocks*block_length
-
-        byte_string = self.ammos_file.read(block_length)
-
-        if len(byte_string) != total:
-            logger.debug("Can not read all %s bytes of data body", total)
-            return None
-
-        for i in range(0, number_of_data_blocks):
-            result = byte_string[i*block_length:(i*block_length+block_length)]
-            data_body.add_data_block(AmmosIFDataBlock(AmmosIFDataBlockHeader.from_bytes(result[0:header_size]),
-                                                      result[header_size:]))
-
-        return data_body
-
-    def read_next_global_frame_body(self, data_header_length):
-        """
-        I return the next global frame body read from current position in file.
-
-        :param data_header_length: the length of the data header
-        :type data_header_length: int
-        """
-        if_data_header = None
-
-        if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
-            if_data_header = self.read_next_global_frame_body_data_header()
-        if data_header_length == AmmosExtendedIFDataHeader.HEADER_SIZE:
-            if_data_header = self.read_next_global_frame_body_extended_data_header()
-
-        if if_data_header is None:
-            logger.debug("Data header missing")
-            return None
-
-        if_data_body = self.read_next_if_data_body(if_data_header.block_count, if_data_header.block_length)
-
-        if if_data_body is None:
-            logger.debug("Data body missing")
-            return None
-
-        return AmmosGlobalFrameBody(if_data_header, if_data_body)
-
-    def payload(self):
-        """I return just the pure date (payload) from my container."""
-        return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosIFReader -(file_name) -
-
-

I read the IF data embedded in an R&S AMMOS recording.

-

I am the standard constructor for Ammos Readers.

-

Additional information about the file can be added as key/value pairs in tags

-

:param file_name: The file to read Ammos data from -:type file_name: str

-
- -Expand source code - -
class AmmosIFReader(AbstractAmmosReader):
-    """I read the IF data embedded in an R&S AMMOS recording."""
-
-    def read_next_global_frame_body_data_header(self):
-        """I read the next data header of a global frame body from current position in file."""
-        header_size = AmmosIFDataHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-
-        logger.info("\nReading global frame body standard data header\n")
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame body data header", header_size)
-            return None
-        return AmmosIFDataHeader.from_bytes(in_bytes)
-
-    def read_next_global_frame_body_extended_data_header(self):
-        """
-        I return the next global frame body extended data header from current position in file.
-
-        :return: the next Ammos Extended IF data header or None if incomplete
-        :rtype: AmmosExtendedIFDataHeader
-        """
-        header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
-
-        in_bytes = self.ammos_file.read(header_size)
-
-        logger.info("\nReading global frame body extended data header\n")
-        if ((not in_bytes) or (len(in_bytes) < header_size)):
-            logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
-            return None
-        return AmmosExtendedIFDataHeader.from_bytes(in_bytes)
-
-    def read_next_if_data_body(self, number_of_data_blocks, data_length):
-        """
-        I return the next data body read from current position in file.
-
-        :param number_of_data_blocks: the number of data blocks inside the body
-        :type number_of_data_blocks: int
-
-        :param data_length: the length of the raw data inside a single block
-        :type data_length: int
-        """
-        header_size = AmmosIFDataBlockHeader.HEADER_SIZE
-
-        data_body = AmmosIFDataBody()
-
-        block_length = header_size + data_length
-
-        total = number_of_data_blocks*block_length
-
-        byte_string = self.ammos_file.read(block_length)
-
-        if len(byte_string) != total:
-            logger.debug("Can not read all %s bytes of data body", total)
-            return None
-
-        for i in range(0, number_of_data_blocks):
-            result = byte_string[i*block_length:(i*block_length+block_length)]
-            data_body.add_data_block(AmmosIFDataBlock(AmmosIFDataBlockHeader.from_bytes(result[0:header_size]),
-                                                      result[header_size:]))
-
-        return data_body
-
-    def read_next_global_frame_body(self, data_header_length):
-        """
-        I return the next global frame body read from current position in file.
-
-        :param data_header_length: the length of the data header
-        :type data_header_length: int
-        """
-        if_data_header = None
-
-        if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
-            if_data_header = self.read_next_global_frame_body_data_header()
-        if data_header_length == AmmosExtendedIFDataHeader.HEADER_SIZE:
-            if_data_header = self.read_next_global_frame_body_extended_data_header()
-
-        if if_data_header is None:
-            logger.debug("Data header missing")
-            return None
-
-        if_data_body = self.read_next_if_data_body(if_data_header.block_count, if_data_header.block_length)
-
-        if if_data_body is None:
-            logger.debug("Data body missing")
-            return None
-
-        return AmmosGlobalFrameBody(if_data_header, if_data_body)
-
-    def payload(self):
-        """I return just the pure date (payload) from my container."""
-        return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])
-
-

Ancestors

- -

Methods

-
-
-def payload(self) -
-
-

I return just the pure date (payload) from my container.

-
- -Expand source code - -
def payload(self):
-    """I return just the pure date (payload) from my container."""
-    return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])
-
-
-
-def read_next_global_frame_body(self, data_header_length) -
-
-

I return the next global frame body read from current position in file.

-

:param data_header_length: the length of the data header -:type data_header_length: int

-
- -Expand source code - -
def read_next_global_frame_body(self, data_header_length):
-    """
-    I return the next global frame body read from current position in file.
-
-    :param data_header_length: the length of the data header
-    :type data_header_length: int
-    """
-    if_data_header = None
-
-    if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
-        if_data_header = self.read_next_global_frame_body_data_header()
-    if data_header_length == AmmosExtendedIFDataHeader.HEADER_SIZE:
-        if_data_header = self.read_next_global_frame_body_extended_data_header()
-
-    if if_data_header is None:
-        logger.debug("Data header missing")
-        return None
-
-    if_data_body = self.read_next_if_data_body(if_data_header.block_count, if_data_header.block_length)
-
-    if if_data_body is None:
-        logger.debug("Data body missing")
-        return None
-
-    return AmmosGlobalFrameBody(if_data_header, if_data_body)
-
-
-
-def read_next_global_frame_body_data_header(self) -
-
-

I read the next data header of a global frame body from current position in file.

-
- -Expand source code - -
def read_next_global_frame_body_data_header(self):
-    """I read the next data header of a global frame body from current position in file."""
-    header_size = AmmosIFDataHeader.HEADER_SIZE
-
-    in_bytes = self.ammos_file.read(header_size)
-
-    logger.info("\nReading global frame body standard data header\n")
-    if ((not in_bytes) or (len(in_bytes) < header_size)):
-        logger.debug("Can not read all %s bytes of global frame body data header", header_size)
-        return None
-    return AmmosIFDataHeader.from_bytes(in_bytes)
-
-
-
-def read_next_global_frame_body_extended_data_header(self) -
-
-

I return the next global frame body extended data header from current position in file.

-

:return: the next Ammos Extended IF data header or None if incomplete -:rtype: AmmosExtendedIFDataHeader

-
- -Expand source code - -
def read_next_global_frame_body_extended_data_header(self):
-    """
-    I return the next global frame body extended data header from current position in file.
-
-    :return: the next Ammos Extended IF data header or None if incomplete
-    :rtype: AmmosExtendedIFDataHeader
-    """
-    header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
-
-    in_bytes = self.ammos_file.read(header_size)
-
-    logger.info("\nReading global frame body extended data header\n")
-    if ((not in_bytes) or (len(in_bytes) < header_size)):
-        logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
-        return None
-    return AmmosExtendedIFDataHeader.from_bytes(in_bytes)
-
-
-
-def read_next_if_data_body(self, number_of_data_blocks, data_length) -
-
-

I return the next data body read from current position in file.

-

:param number_of_data_blocks: the number of data blocks inside the body -:type number_of_data_blocks: int

-

:param data_length: the length of the raw data inside a single block -:type data_length: int

-
- -Expand source code - -
def read_next_if_data_body(self, number_of_data_blocks, data_length):
-    """
-    I return the next data body read from current position in file.
-
-    :param number_of_data_blocks: the number of data blocks inside the body
-    :type number_of_data_blocks: int
-
-    :param data_length: the length of the raw data inside a single block
-    :type data_length: int
-    """
-    header_size = AmmosIFDataBlockHeader.HEADER_SIZE
-
-    data_body = AmmosIFDataBody()
-
-    block_length = header_size + data_length
-
-    total = number_of_data_blocks*block_length
-
-    byte_string = self.ammos_file.read(block_length)
-
-    if len(byte_string) != total:
-        logger.debug("Can not read all %s bytes of data body", total)
-        return None
-
-    for i in range(0, number_of_data_blocks):
-        result = byte_string[i*block_length:(i*block_length+block_length)]
-        data_body.add_data_block(AmmosIFDataBlock(AmmosIFDataBlockHeader.from_bytes(result[0:header_size]),
-                                                  result[header_size:]))
-
-    return data_body
-
-
-
-

Inherited members

- -
-
-
-
- -
- - - Index: c/ammosreader/AmmosSingleFrame.html =================================================================== --- doc/ammosreader/AmmosSingleFrame.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,206 +1,0 @@ - - - - - - -ammosreader.AmmosSingleFrame API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.AmmosSingleFrame

-
-
-
- -Expand source code - -
class AmmosSingleFrame():
-
-    def __init__(self, global_frame_header, global_frame_body):
-        self.__global_frame_header = global_frame_header
-        self.__global_frame_body = global_frame_body
-
-    @property
-    def global_frame_header(self):
-        return self.__global_frame_header
-
-    @property
-    def global_frame_body(self):
-        return self.__global_frame_body
-
-    def data(self):
-        return self.global_frame_body.data_bytes_only()
-
-    def size_correct(self):
-        return (self.global_frame_header.frame_length == (24 + self.global_frame_header.data_header_length +
-                                                          (self.global_frame_body.data_header.block_count *
-                                                           (self.global_frame_body.data_header.block_length + 4))))
-    def __str__(self):
-        output = (
-            "Global frame header\n" +
-            "-------------------\n" +
-            "Frame type:" + str(self.global_frame_header.frame_type) + "\n" +
-            "Frame count:" + str(self.global_frame_header.running_frame_number) + "\n" +
-            "Data header length:" + str(self.global_frame_header.data_header_length) + " bytes\n" +
-            "Frame length:" + str(self.global_frame_header.frame_length) + " bytes\n"
-        )
-        return output
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class AmmosSingleFrame -(global_frame_header, global_frame_body) -
-
-
-
- -Expand source code - -
class AmmosSingleFrame():
-
-    def __init__(self, global_frame_header, global_frame_body):
-        self.__global_frame_header = global_frame_header
-        self.__global_frame_body = global_frame_body
-
-    @property
-    def global_frame_header(self):
-        return self.__global_frame_header
-
-    @property
-    def global_frame_body(self):
-        return self.__global_frame_body
-
-    def data(self):
-        return self.global_frame_body.data_bytes_only()
-
-    def size_correct(self):
-        return (self.global_frame_header.frame_length == (24 + self.global_frame_header.data_header_length +
-                                                          (self.global_frame_body.data_header.block_count *
-                                                           (self.global_frame_body.data_header.block_length + 4))))
-    def __str__(self):
-        output = (
-            "Global frame header\n" +
-            "-------------------\n" +
-            "Frame type:" + str(self.global_frame_header.frame_type) + "\n" +
-            "Frame count:" + str(self.global_frame_header.running_frame_number) + "\n" +
-            "Data header length:" + str(self.global_frame_header.data_header_length) + " bytes\n" +
-            "Frame length:" + str(self.global_frame_header.frame_length) + " bytes\n"
-        )
-        return output
-
-

Instance variables

-
-
var global_frame_body
-
-
-
- -Expand source code - -
@property
-def global_frame_body(self):
-    return self.__global_frame_body
-
-
-
var global_frame_header
-
-
-
- -Expand source code - -
@property
-def global_frame_header(self):
-    return self.__global_frame_header
-
-
-
-

Methods

-
-
-def data(self) -
-
-
-
- -Expand source code - -
def data(self):
-    return self.global_frame_body.data_bytes_only()
-
-
-
-def size_correct(self) -
-
-
-
- -Expand source code - -
def size_correct(self):
-    return (self.global_frame_header.frame_length == (24 + self.global_frame_header.data_header_length +
-                                                      (self.global_frame_body.data_header.block_count *
-                                                       (self.global_frame_body.data_header.block_length + 4))))
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/IQDWTXBlock.html =================================================================== --- doc/ammosreader/IQDWTXBlock.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,126 +1,0 @@ - - - - - - -ammosreader.IQDWTXBlock API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.IQDWTXBlock

-
-
-
- -Expand source code - -
from ammosreader.PDW import PDW
-
-
-class IQDWTXBLOCK():
-
-    @classmethod
-    def from_bytes(cls, byte_string):
-        pdw = PDW.from_bytes(byte_string[0:32])
-        return IQDWTXBLOCK(pdw, byte_string[32:])
-
-    def __init__(self, pdw, iq):
-
-        self.pdw = pdw
-        self.iq = iq
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class IQDWTXBLOCK -(pdw, iq) -
-
-
-
- -Expand source code - -
class IQDWTXBLOCK():
-
-    @classmethod
-    def from_bytes(cls, byte_string):
-        pdw = PDW.from_bytes(byte_string[0:32])
-        return IQDWTXBLOCK(pdw, byte_string[32:])
-
-    def __init__(self, pdw, iq):
-
-        self.pdw = pdw
-        self.iq = iq
-
-

Static methods

-
-
-def from_bytes(byte_string) -
-
-
-
- -Expand source code - -
@classmethod
-def from_bytes(cls, byte_string):
-    pdw = PDW.from_bytes(byte_string[0:32])
-    return IQDWTXBLOCK(pdw, byte_string[32:])
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/PDW.html =================================================================== --- doc/ammosreader/PDW.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,780 +1,0 @@ - - - - - - -ammosreader.PDW API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.PDW

-
-
-

I store the information of a single PDW block.

-
- -Expand source code - -
"""I store the information of a single PDW block."""
-
-import struct
-import math
-import numpy as np
-from ammosreader import logger
-
-class PDW():
-    """
-    I store information from a single ppdw data block.
-
-    .. automethod:: __init__
-    """
-
-    @classmethod
-    def from_bytes(cls, byte_string):
-        """
-        I create an instance of class PDW from data body (8 * 32 bits).
-
-        :param byte_string: a byte string containing a single data body read from a ppdw file
-        :type byte_string: byte string
-
-        :return: an instance of class PDW with attributes set according to the data of a data body
-        :rtype: PDW
-        """
-
-        logger.info("from bytes")
-        if (len(byte_string) != 32):
-            logger.error("Byte count invalid")
-            raise TypeError("Byte count invalid")
-
-        parts = struct.unpack('<Q4s4s4s4s4s4s', byte_string)
-
-        nanoseconds = (parts[0])
-        unix_time = np.datetime64('now', 'ns').astype(int)
-        if nanoseconds >= unix_time:
-            raise OverflowError("Timestamp invalid")
-        time_of_arrival = np.datetime64(nanoseconds, 'ns')
-
-        third_entry = bin(int.from_bytes(parts[1], byteorder='little'))
-        padding = 32-len(str(third_entry)[2:])
-        third_entry_bit_string = "0" * padding + str(third_entry)[2:]
-        pdw_format_identifier = int(third_entry_bit_string[0:6], 2)
-        center_frequency = int(third_entry_bit_string[5:32], 2)
-
-        fourth_entry = bin(int.from_bytes(parts[2], byteorder='little'))
-        padding = 32-len(str(fourth_entry)[2:])
-        fourth_entry_bit_string = "0" * padding + str(fourth_entry)[2:]
-        is_valid = bool(int(fourth_entry_bit_string[0]))
-        is_pulse = bool(int(fourth_entry_bit_string[1]))
-        level_unit = int(fourth_entry_bit_string[2])
-        signal_start_missing = bool(int(fourth_entry_bit_string[3]))
-        signal_end_missing = bool(int(fourth_entry_bit_string[4]))
-        pulse_width = int(fourth_entry_bit_string[7:33], 2)
-
-        fifth_entry = bin(int.from_bytes(parts[3], byteorder='little'))
-        padding = 32-len(str(fifth_entry)[2:])
-        fifth_entry_bit_string = "0" * padding + str(fifth_entry)[2:]
-        frequency_shift_or_bandwidth = int(fifth_entry_bit_string[0:20], 2)
-        # FIXME: You have to scale me to the range from -200.0 to 200.0 in 0.1 steps
-        pulse_level_or_pulse_field_strength = math.ceil(int(fifth_entry_bit_string[20:32], 2)) / 10
-
-        sixth_entry = bin(int.from_bytes(parts[4], byteorder='little'))
-        padding = 32-len(str(sixth_entry)[2:])
-        sixth_entry_bit_string = "0" * padding + str(sixth_entry)[2:]
-        region_of_interest = bool(int(sixth_entry_bit_string[0]))
-        # FIXME: You have to scale me to a range from 0.0 to 6.2 in steps of 0.1 - 6.3 means unknown
-        azimuth_confidence = math.ceil(int(sixth_entry_bit_string[1:7], 2)) / 10
-        modulations = {0: 'Unknown', 1: 'Unmodulated', 2: 'FM', 3: 'LFM', 4: 'PSK-2', 5: 'PSK-3', 6: 'PSK-4',
-                       7: 'PSK-m', 8: 'NLFM', 9: 'SFM', 10: 'TFM', 11: 'Pulse too short'}
-        modulation = modulations.get(int(sixth_entry_bit_string[7:12], 2), 0)
-        sector = int(sixth_entry_bit_string[28:32], 2)
-
-        seventh_entry = bin(int.from_bytes(parts[5], byteorder='little'))
-        padding = 32-len(str(seventh_entry)[2:])
-        seventh_entry_bit_string = "0" * padding + str(seventh_entry)[2:]
-        polarities = {0: 'Horizontal/Unknown', 1: 'Vertical', 2: 'Counter clockwise', 3: 'Clockwise'}
-        polarity = polarities[int(seventh_entry_bit_string[0:2], 2)]
-        df_quality = int(seventh_entry_bit_string[2:9], 2)
-        # FIXME: You have to scale me from -90 to 90 in 0.1 degree steps
-        elevation = int(seventh_entry_bit_string[9:20], 2)
-        # FIXME: You have to check me for a range from 0.0 to 359.9 in steps of 0.1
-        azimuth = 0.1 * (int(seventh_entry_bit_string[20:32], 2))
-
-        eighth_entry = bin(int.from_bytes(parts[5], byteorder='little'))
-        padding = 32-len(str(eighth_entry)[2:])
-        eighth_entry_bit_string = "0" * padding + str(eighth_entry)[2:]
-        channel = int(eighth_entry_bit_string[0:4], 2)
-
-        return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit,
-                   signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
-                   pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
-                   sector, polarity, df_quality, elevation, azimuth, channel)
-
-    def __init__(self, time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse,
-                 level_unit, signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
-                 pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
-                 sector, polarity, df_quality, elevation, azimuth, channel):
-        r"""
-        I return an instance of an Pulse Data word.
-
-        :param time_of_arrival: nanoseconds since 1970-01-01 00:00:00
-        :type time_of_arrival: Integer
-        :param pdw_format: format code
-        :type pdw_format: Integer
-        :param center_frequency: center frequency in KHz
-        :type center_frequency: Integer
-        :param is_valid: flag to mark if pdw data body is valid
-        :type is_valid: Boolean
-        :param is_pulse: flag to mark if pdw data body contains a pulse or a continuous wave signal
-        :type is_pulse: Boolean
-        :param level_unit: 0 means dBµV - 1 means dBµV/m
-        :type level_unit: Integer
-        :param signal_start_missing: signal started before time of arrival
-        :type signal_start_missing: Boolean
-        :param signal_end_missing: signal stops after time of arrival
-        :type signal_end_missing: Boolean
-        :param pulse_width: pulse width in nanoseconds - Zero if no valid pulse detected
-        :type pulse_width: Integer
-        :param frequency_shift_or_bandwidth: Value in KHz - Value set to 1048575 means Unknown
-        :type frequency_shift_or_bandwidth: Integer
-        :param pulse_level_or_pulse_field_strength: Pulse level or Pulse Field Strength depending on level_unit \
-         (-200.0...200.0) in 0.1 steps / minus 204.8 means no valid level detected
-        :type pulse_level_or_pulse_field_strength: Float
-        :param region_of_interest: Marks if signal is from region of interest
-        :type region_of_interest: Boolean
-        :param azimuth_confidence: degree in steps of 0.1 (0.0-6.2) / 6.3 means confidence unknown
-        :type azimuth_confidence: Float
-        :param modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.)
-        :type modulation: String
-        :param sector: reference antenna sector (0-15)
-        :type sector: Integer
-        :param polarity: Horizontal, Vertical, Clockwise, Counter clockwise
-        :type polarity: String
-        :param df_quality: Direction finding quality in percent (0-100) - Zero means unknown
-        :type df_quality: Integer
-        :param elevation: elevation of incoming signal (from -90 to 90 degree) in steps of 0.1 degree \
-        minus 102.4 means unknown
-        :type elevation: Float
-        :param azimuth: azimuth of incoming signal (from 0 to 359.9 degree) in steps of 0.1 degree \
-        plus 409.5 means unknown
-        :type azimuth: Float
-        :param channel: detecting channel (0-16) - Zero means unknown
-        :type channel: Integer
-        :return: An instance of class PDW with attributes set according to the data of a data body
-        :rtype: PDW
-        """
-        self.time_of_arrival = time_of_arrival #
-        self.pdw_format_identifier = pdw_format_identifier
-        self.center_frequency = center_frequency #
-        self.__is_valid = is_valid #
-        self.is_pulse = is_pulse #
-        self.level_unit = level_unit #
-        self.signal_start_missing = signal_start_missing
-        self.signal_end_missing = signal_end_missing
-        self.pulse_width = pulse_width #
-        self.frequency_shift_or_bandwidth = frequency_shift_or_bandwidth #
-        self.pulse_level_or_pulse_field_strength = pulse_level_or_pulse_field_strength #
-        self.region_of_interest = region_of_interest
-        self.azimuth_confidence = azimuth_confidence
-        self.modulation = modulation #
-        self.sector = sector
-        self.polarity = polarity
-        self.df_quality = df_quality #
-        self.elevation = elevation #
-        self.azimuth = azimuth
-        self.channel = channel #
-
-    def __str__(self):
-        """
-        I return the string representation of myself.
-
-        :rtype: str
-        """
-        output = ("Time of arrival: " + str(self.time_of_arrival) + "\n" +
-                  "PDW Format identifier: " + str(self.pdw_format_identifier) + "\n" +
-                  "Center frequency: " + str(self.center_frequency) + " KHz\n")
-
-        if self.__is_valid:
-            output += "Signal: Valid\n"
-        else:
-            output += "Signal: Invalid\n"
-
-        if self.is_pulse:
-            output += "Signal type: Pulse\n"
-        else:
-            output += "Signal type: Continuous wave\n"
-
-        if self.level_unit == 1:
-            output += "Pulse level: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV\n"
-        else:
-            output += "Pulse field strength: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV/meter\n"
-
-        output += ("Pulse width: " + str(self.pulse_width) + " nanoseconds\n" +
-                   "Frequency shift or bandwidth: " + str(self.frequency_shift_or_bandwidth) + " KHz\n")
-
-        if self.region_of_interest:
-            output += "Region of interest: Yes\n"
-        else:
-            output += "Region of interest: No\n"
-
-        if self.azimuth_confidence == 6.3:
-            output += "Azimuth confidence: Invalid\n"
-        else:
-            output += "Azimuth confidence: " + str(self.azimuth_confidence) + " degree\n"
-
-        output += "Modulation: " + str(self.modulation) + "\n"
-
-        if self.sector == 0:
-            output += "Sector: Unknown\n"
-        else:
-            output += "Sector:" + str(self.sector) + "\n"
-
-        output += "Polarity: " + str(self.polarity) + "\n"
-
-        output += "DF quality: " + str(self.df_quality) + " %\n"
-
-        if self.elevation == 1024:
-            output += "Elevation: Unknown\n"
-        else:
-            output += "Elevation: " + str(self.elevation) + " degree\n"
-
-        if self.azimuth == 409.5:
-            output += "Azimuth: Unknown\n"
-        else:
-            output += "Azimuth: " + str(self.azimuth) + " degree\n"
-
-        output += "Channel: " + str(self.channel) + "\n"
-
-        return output
-
-    def is_valid(self):
-        return self.__is_valid
-
-    def to_json(self):
-        return {'time of arrival': self.time_of_arrival,
-                'FORMATIDENTIFIER': self.pdw_format_identifier,
-                'center frequency': self.center_frequency,
-                'VALID': self.__is_valid,
-                'PULSE': self.is_pulse,
-                'PULSELEVEL': self.pulse_level_or_pulse_field_strength,
-                'PULSEWIDTH': self.pulse_width,
-                'BANDWIDTH': self.frequency_shift_or_bandwidth,
-                'REGIONOFINTEREST': self.region_of_interest,
-                'AZIMUTHCONFIDENCE': self.azimuth_confidence,
-                'MODULATION': self.modulation,
-                'SECTOR': self.sector,
-                'POLARITY': self.polarity,
-                'DFQUALITY': self.df_quality,
-                'ELEVATION': self.elevation,
-                'AZIMUTH': self.azimuth,
-                'CHANNEL': self.channel
-                }
-
-if __name__ == '__main__':
-    pass
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class PDW -(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit, signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth, pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation, sector, polarity, df_quality, elevation, azimuth, channel) -
-
-

I store information from a single ppdw data block.

-
-

Automethod: init

-
-

I return an instance of an Pulse Data word.

-

:param time_of_arrival: nanoseconds since 1970-01-01 00:00:00 -:type time_of_arrival: Integer -:param pdw_format: format code -:type pdw_format: Integer -:param center_frequency: center frequency in KHz -:type center_frequency: Integer -:param is_valid: flag to mark if pdw data body is valid -:type is_valid: Boolean -:param is_pulse: flag to mark if pdw data body contains a pulse or a continuous wave signal -:type is_pulse: Boolean -:param level_unit: 0 means dBµV - 1 means dBµV/m -:type level_unit: Integer -:param signal_start_missing: signal started before time of arrival -:type signal_start_missing: Boolean -:param signal_end_missing: signal stops after time of arrival -:type signal_end_missing: Boolean -:param pulse_width: pulse width in nanoseconds - Zero if no valid pulse detected -:type pulse_width: Integer -:param frequency_shift_or_bandwidth: Value in KHz - Value set to 1048575 means Unknown -:type frequency_shift_or_bandwidth: Integer -:param pulse_level_or_pulse_field_strength: Pulse level or Pulse Field Strength depending on level_unit \ -(-200.0…200.0) in 0.1 steps / minus 204.8 means no valid level detected -:type pulse_level_or_pulse_field_strength: Float -:param region_of_interest: Marks if signal is from region of interest -:type region_of_interest: Boolean -:param azimuth_confidence: degree in steps of 0.1 (0.0-6.2) / 6.3 means confidence unknown -:type azimuth_confidence: Float -:param modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.) -:type modulation: String -:param sector: reference antenna sector (0-15) -:type sector: Integer -:param polarity: Horizontal, Vertical, Clockwise, Counter clockwise -:type polarity: String -:param df_quality: Direction finding quality in percent (0-100) - Zero means unknown -:type df_quality: Integer -:param elevation: elevation of incoming signal (from -90 to 90 degree) in steps of 0.1 degree \ -minus 102.4 means unknown -:type elevation: Float -:param azimuth: azimuth of incoming signal (from 0 to 359.9 degree) in steps of 0.1 degree \ -plus 409.5 means unknown -:type azimuth: Float -:param channel: detecting channel (0-16) - Zero means unknown -:type channel: Integer -:return: An instance of class PDW with attributes set according to the data of a data body -:rtype: PDW

-
- -Expand source code - -
class PDW():
-    """
-    I store information from a single ppdw data block.
-
-    .. automethod:: __init__
-    """
-
-    @classmethod
-    def from_bytes(cls, byte_string):
-        """
-        I create an instance of class PDW from data body (8 * 32 bits).
-
-        :param byte_string: a byte string containing a single data body read from a ppdw file
-        :type byte_string: byte string
-
-        :return: an instance of class PDW with attributes set according to the data of a data body
-        :rtype: PDW
-        """
-
-        logger.info("from bytes")
-        if (len(byte_string) != 32):
-            logger.error("Byte count invalid")
-            raise TypeError("Byte count invalid")
-
-        parts = struct.unpack('<Q4s4s4s4s4s4s', byte_string)
-
-        nanoseconds = (parts[0])
-        unix_time = np.datetime64('now', 'ns').astype(int)
-        if nanoseconds >= unix_time:
-            raise OverflowError("Timestamp invalid")
-        time_of_arrival = np.datetime64(nanoseconds, 'ns')
-
-        third_entry = bin(int.from_bytes(parts[1], byteorder='little'))
-        padding = 32-len(str(third_entry)[2:])
-        third_entry_bit_string = "0" * padding + str(third_entry)[2:]
-        pdw_format_identifier = int(third_entry_bit_string[0:6], 2)
-        center_frequency = int(third_entry_bit_string[5:32], 2)
-
-        fourth_entry = bin(int.from_bytes(parts[2], byteorder='little'))
-        padding = 32-len(str(fourth_entry)[2:])
-        fourth_entry_bit_string = "0" * padding + str(fourth_entry)[2:]
-        is_valid = bool(int(fourth_entry_bit_string[0]))
-        is_pulse = bool(int(fourth_entry_bit_string[1]))
-        level_unit = int(fourth_entry_bit_string[2])
-        signal_start_missing = bool(int(fourth_entry_bit_string[3]))
-        signal_end_missing = bool(int(fourth_entry_bit_string[4]))
-        pulse_width = int(fourth_entry_bit_string[7:33], 2)
-
-        fifth_entry = bin(int.from_bytes(parts[3], byteorder='little'))
-        padding = 32-len(str(fifth_entry)[2:])
-        fifth_entry_bit_string = "0" * padding + str(fifth_entry)[2:]
-        frequency_shift_or_bandwidth = int(fifth_entry_bit_string[0:20], 2)
-        # FIXME: You have to scale me to the range from -200.0 to 200.0 in 0.1 steps
-        pulse_level_or_pulse_field_strength = math.ceil(int(fifth_entry_bit_string[20:32], 2)) / 10
-
-        sixth_entry = bin(int.from_bytes(parts[4], byteorder='little'))
-        padding = 32-len(str(sixth_entry)[2:])
-        sixth_entry_bit_string = "0" * padding + str(sixth_entry)[2:]
-        region_of_interest = bool(int(sixth_entry_bit_string[0]))
-        # FIXME: You have to scale me to a range from 0.0 to 6.2 in steps of 0.1 - 6.3 means unknown
-        azimuth_confidence = math.ceil(int(sixth_entry_bit_string[1:7], 2)) / 10
-        modulations = {0: 'Unknown', 1: 'Unmodulated', 2: 'FM', 3: 'LFM', 4: 'PSK-2', 5: 'PSK-3', 6: 'PSK-4',
-                       7: 'PSK-m', 8: 'NLFM', 9: 'SFM', 10: 'TFM', 11: 'Pulse too short'}
-        modulation = modulations.get(int(sixth_entry_bit_string[7:12], 2), 0)
-        sector = int(sixth_entry_bit_string[28:32], 2)
-
-        seventh_entry = bin(int.from_bytes(parts[5], byteorder='little'))
-        padding = 32-len(str(seventh_entry)[2:])
-        seventh_entry_bit_string = "0" * padding + str(seventh_entry)[2:]
-        polarities = {0: 'Horizontal/Unknown', 1: 'Vertical', 2: 'Counter clockwise', 3: 'Clockwise'}
-        polarity = polarities[int(seventh_entry_bit_string[0:2], 2)]
-        df_quality = int(seventh_entry_bit_string[2:9], 2)
-        # FIXME: You have to scale me from -90 to 90 in 0.1 degree steps
-        elevation = int(seventh_entry_bit_string[9:20], 2)
-        # FIXME: You have to check me for a range from 0.0 to 359.9 in steps of 0.1
-        azimuth = 0.1 * (int(seventh_entry_bit_string[20:32], 2))
-
-        eighth_entry = bin(int.from_bytes(parts[5], byteorder='little'))
-        padding = 32-len(str(eighth_entry)[2:])
-        eighth_entry_bit_string = "0" * padding + str(eighth_entry)[2:]
-        channel = int(eighth_entry_bit_string[0:4], 2)
-
-        return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit,
-                   signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
-                   pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
-                   sector, polarity, df_quality, elevation, azimuth, channel)
-
-    def __init__(self, time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse,
-                 level_unit, signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
-                 pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
-                 sector, polarity, df_quality, elevation, azimuth, channel):
-        r"""
-        I return an instance of an Pulse Data word.
-
-        :param time_of_arrival: nanoseconds since 1970-01-01 00:00:00
-        :type time_of_arrival: Integer
-        :param pdw_format: format code
-        :type pdw_format: Integer
-        :param center_frequency: center frequency in KHz
-        :type center_frequency: Integer
-        :param is_valid: flag to mark if pdw data body is valid
-        :type is_valid: Boolean
-        :param is_pulse: flag to mark if pdw data body contains a pulse or a continuous wave signal
-        :type is_pulse: Boolean
-        :param level_unit: 0 means dBµV - 1 means dBµV/m
-        :type level_unit: Integer
-        :param signal_start_missing: signal started before time of arrival
-        :type signal_start_missing: Boolean
-        :param signal_end_missing: signal stops after time of arrival
-        :type signal_end_missing: Boolean
-        :param pulse_width: pulse width in nanoseconds - Zero if no valid pulse detected
-        :type pulse_width: Integer
-        :param frequency_shift_or_bandwidth: Value in KHz - Value set to 1048575 means Unknown
-        :type frequency_shift_or_bandwidth: Integer
-        :param pulse_level_or_pulse_field_strength: Pulse level or Pulse Field Strength depending on level_unit \
-         (-200.0...200.0) in 0.1 steps / minus 204.8 means no valid level detected
-        :type pulse_level_or_pulse_field_strength: Float
-        :param region_of_interest: Marks if signal is from region of interest
-        :type region_of_interest: Boolean
-        :param azimuth_confidence: degree in steps of 0.1 (0.0-6.2) / 6.3 means confidence unknown
-        :type azimuth_confidence: Float
-        :param modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.)
-        :type modulation: String
-        :param sector: reference antenna sector (0-15)
-        :type sector: Integer
-        :param polarity: Horizontal, Vertical, Clockwise, Counter clockwise
-        :type polarity: String
-        :param df_quality: Direction finding quality in percent (0-100) - Zero means unknown
-        :type df_quality: Integer
-        :param elevation: elevation of incoming signal (from -90 to 90 degree) in steps of 0.1 degree \
-        minus 102.4 means unknown
-        :type elevation: Float
-        :param azimuth: azimuth of incoming signal (from 0 to 359.9 degree) in steps of 0.1 degree \
-        plus 409.5 means unknown
-        :type azimuth: Float
-        :param channel: detecting channel (0-16) - Zero means unknown
-        :type channel: Integer
-        :return: An instance of class PDW with attributes set according to the data of a data body
-        :rtype: PDW
-        """
-        self.time_of_arrival = time_of_arrival #
-        self.pdw_format_identifier = pdw_format_identifier
-        self.center_frequency = center_frequency #
-        self.__is_valid = is_valid #
-        self.is_pulse = is_pulse #
-        self.level_unit = level_unit #
-        self.signal_start_missing = signal_start_missing
-        self.signal_end_missing = signal_end_missing
-        self.pulse_width = pulse_width #
-        self.frequency_shift_or_bandwidth = frequency_shift_or_bandwidth #
-        self.pulse_level_or_pulse_field_strength = pulse_level_or_pulse_field_strength #
-        self.region_of_interest = region_of_interest
-        self.azimuth_confidence = azimuth_confidence
-        self.modulation = modulation #
-        self.sector = sector
-        self.polarity = polarity
-        self.df_quality = df_quality #
-        self.elevation = elevation #
-        self.azimuth = azimuth
-        self.channel = channel #
-
-    def __str__(self):
-        """
-        I return the string representation of myself.
-
-        :rtype: str
-        """
-        output = ("Time of arrival: " + str(self.time_of_arrival) + "\n" +
-                  "PDW Format identifier: " + str(self.pdw_format_identifier) + "\n" +
-                  "Center frequency: " + str(self.center_frequency) + " KHz\n")
-
-        if self.__is_valid:
-            output += "Signal: Valid\n"
-        else:
-            output += "Signal: Invalid\n"
-
-        if self.is_pulse:
-            output += "Signal type: Pulse\n"
-        else:
-            output += "Signal type: Continuous wave\n"
-
-        if self.level_unit == 1:
-            output += "Pulse level: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV\n"
-        else:
-            output += "Pulse field strength: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV/meter\n"
-
-        output += ("Pulse width: " + str(self.pulse_width) + " nanoseconds\n" +
-                   "Frequency shift or bandwidth: " + str(self.frequency_shift_or_bandwidth) + " KHz\n")
-
-        if self.region_of_interest:
-            output += "Region of interest: Yes\n"
-        else:
-            output += "Region of interest: No\n"
-
-        if self.azimuth_confidence == 6.3:
-            output += "Azimuth confidence: Invalid\n"
-        else:
-            output += "Azimuth confidence: " + str(self.azimuth_confidence) + " degree\n"
-
-        output += "Modulation: " + str(self.modulation) + "\n"
-
-        if self.sector == 0:
-            output += "Sector: Unknown\n"
-        else:
-            output += "Sector:" + str(self.sector) + "\n"
-
-        output += "Polarity: " + str(self.polarity) + "\n"
-
-        output += "DF quality: " + str(self.df_quality) + " %\n"
-
-        if self.elevation == 1024:
-            output += "Elevation: Unknown\n"
-        else:
-            output += "Elevation: " + str(self.elevation) + " degree\n"
-
-        if self.azimuth == 409.5:
-            output += "Azimuth: Unknown\n"
-        else:
-            output += "Azimuth: " + str(self.azimuth) + " degree\n"
-
-        output += "Channel: " + str(self.channel) + "\n"
-
-        return output
-
-    def is_valid(self):
-        return self.__is_valid
-
-    def to_json(self):
-        return {'time of arrival': self.time_of_arrival,
-                'FORMATIDENTIFIER': self.pdw_format_identifier,
-                'center frequency': self.center_frequency,
-                'VALID': self.__is_valid,
-                'PULSE': self.is_pulse,
-                'PULSELEVEL': self.pulse_level_or_pulse_field_strength,
-                'PULSEWIDTH': self.pulse_width,
-                'BANDWIDTH': self.frequency_shift_or_bandwidth,
-                'REGIONOFINTEREST': self.region_of_interest,
-                'AZIMUTHCONFIDENCE': self.azimuth_confidence,
-                'MODULATION': self.modulation,
-                'SECTOR': self.sector,
-                'POLARITY': self.polarity,
-                'DFQUALITY': self.df_quality,
-                'ELEVATION': self.elevation,
-                'AZIMUTH': self.azimuth,
-                'CHANNEL': self.channel
-                }
-
-

Static methods

-
-
-def from_bytes(byte_string) -
-
-

I create an instance of class PDW from data body (8 * 32 bits).

-

:param byte_string: a byte string containing a single data body read from a ppdw file -:type byte_string: byte string

-

:return: an instance of class PDW with attributes set according to the data of a data body -:rtype: PDW

-
- -Expand source code - -
@classmethod
-def from_bytes(cls, byte_string):
-    """
-    I create an instance of class PDW from data body (8 * 32 bits).
-
-    :param byte_string: a byte string containing a single data body read from a ppdw file
-    :type byte_string: byte string
-
-    :return: an instance of class PDW with attributes set according to the data of a data body
-    :rtype: PDW
-    """
-
-    logger.info("from bytes")
-    if (len(byte_string) != 32):
-        logger.error("Byte count invalid")
-        raise TypeError("Byte count invalid")
-
-    parts = struct.unpack('<Q4s4s4s4s4s4s', byte_string)
-
-    nanoseconds = (parts[0])
-    unix_time = np.datetime64('now', 'ns').astype(int)
-    if nanoseconds >= unix_time:
-        raise OverflowError("Timestamp invalid")
-    time_of_arrival = np.datetime64(nanoseconds, 'ns')
-
-    third_entry = bin(int.from_bytes(parts[1], byteorder='little'))
-    padding = 32-len(str(third_entry)[2:])
-    third_entry_bit_string = "0" * padding + str(third_entry)[2:]
-    pdw_format_identifier = int(third_entry_bit_string[0:6], 2)
-    center_frequency = int(third_entry_bit_string[5:32], 2)
-
-    fourth_entry = bin(int.from_bytes(parts[2], byteorder='little'))
-    padding = 32-len(str(fourth_entry)[2:])
-    fourth_entry_bit_string = "0" * padding + str(fourth_entry)[2:]
-    is_valid = bool(int(fourth_entry_bit_string[0]))
-    is_pulse = bool(int(fourth_entry_bit_string[1]))
-    level_unit = int(fourth_entry_bit_string[2])
-    signal_start_missing = bool(int(fourth_entry_bit_string[3]))
-    signal_end_missing = bool(int(fourth_entry_bit_string[4]))
-    pulse_width = int(fourth_entry_bit_string[7:33], 2)
-
-    fifth_entry = bin(int.from_bytes(parts[3], byteorder='little'))
-    padding = 32-len(str(fifth_entry)[2:])
-    fifth_entry_bit_string = "0" * padding + str(fifth_entry)[2:]
-    frequency_shift_or_bandwidth = int(fifth_entry_bit_string[0:20], 2)
-    # FIXME: You have to scale me to the range from -200.0 to 200.0 in 0.1 steps
-    pulse_level_or_pulse_field_strength = math.ceil(int(fifth_entry_bit_string[20:32], 2)) / 10
-
-    sixth_entry = bin(int.from_bytes(parts[4], byteorder='little'))
-    padding = 32-len(str(sixth_entry)[2:])
-    sixth_entry_bit_string = "0" * padding + str(sixth_entry)[2:]
-    region_of_interest = bool(int(sixth_entry_bit_string[0]))
-    # FIXME: You have to scale me to a range from 0.0 to 6.2 in steps of 0.1 - 6.3 means unknown
-    azimuth_confidence = math.ceil(int(sixth_entry_bit_string[1:7], 2)) / 10
-    modulations = {0: 'Unknown', 1: 'Unmodulated', 2: 'FM', 3: 'LFM', 4: 'PSK-2', 5: 'PSK-3', 6: 'PSK-4',
-                   7: 'PSK-m', 8: 'NLFM', 9: 'SFM', 10: 'TFM', 11: 'Pulse too short'}
-    modulation = modulations.get(int(sixth_entry_bit_string[7:12], 2), 0)
-    sector = int(sixth_entry_bit_string[28:32], 2)
-
-    seventh_entry = bin(int.from_bytes(parts[5], byteorder='little'))
-    padding = 32-len(str(seventh_entry)[2:])
-    seventh_entry_bit_string = "0" * padding + str(seventh_entry)[2:]
-    polarities = {0: 'Horizontal/Unknown', 1: 'Vertical', 2: 'Counter clockwise', 3: 'Clockwise'}
-    polarity = polarities[int(seventh_entry_bit_string[0:2], 2)]
-    df_quality = int(seventh_entry_bit_string[2:9], 2)
-    # FIXME: You have to scale me from -90 to 90 in 0.1 degree steps
-    elevation = int(seventh_entry_bit_string[9:20], 2)
-    # FIXME: You have to check me for a range from 0.0 to 359.9 in steps of 0.1
-    azimuth = 0.1 * (int(seventh_entry_bit_string[20:32], 2))
-
-    eighth_entry = bin(int.from_bytes(parts[5], byteorder='little'))
-    padding = 32-len(str(eighth_entry)[2:])
-    eighth_entry_bit_string = "0" * padding + str(eighth_entry)[2:]
-    channel = int(eighth_entry_bit_string[0:4], 2)
-
-    return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit,
-               signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
-               pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
-               sector, polarity, df_quality, elevation, azimuth, channel)
-
-
-
-

Methods

-
-
-def is_valid(self) -
-
-
-
- -Expand source code - -
def is_valid(self):
-    return self.__is_valid
-
-
-
-def to_json(self) -
-
-
-
- -Expand source code - -
def to_json(self):
-    return {'time of arrival': self.time_of_arrival,
-            'FORMATIDENTIFIER': self.pdw_format_identifier,
-            'center frequency': self.center_frequency,
-            'VALID': self.__is_valid,
-            'PULSE': self.is_pulse,
-            'PULSELEVEL': self.pulse_level_or_pulse_field_strength,
-            'PULSEWIDTH': self.pulse_width,
-            'BANDWIDTH': self.frequency_shift_or_bandwidth,
-            'REGIONOFINTEREST': self.region_of_interest,
-            'AZIMUTHCONFIDENCE': self.azimuth_confidence,
-            'MODULATION': self.modulation,
-            'SECTOR': self.sector,
-            'POLARITY': self.polarity,
-            'DFQUALITY': self.df_quality,
-            'ELEVATION': self.elevation,
-            'AZIMUTH': self.azimuth,
-            'CHANNEL': self.channel
-            }
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/PPDWContainer.html =================================================================== --- doc/ammosreader/PPDWContainer.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,444 +1,0 @@ - - - - - - -ammosreader.PPDWContainer API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.PPDWContainer

-
-
-
- -Expand source code - -
from datetime import datetime
-import numpy as np
-
-class PPDWContainer():
-    """
-    I store multiple signals imported from one or more .ppdw files
-    .. automethod:: __init__
-    """
-
-    def __init__(self, name, signals=None):
-
-        if signals is None:
-            self.signals = []
-        else:
-            self.signals = signals
-
-        self.name = name
-
-    def __str__(self):
-        return "\n".join(["Number of pulses:" + str(len(self.signals)),
-                          "Number of valid pulses:" + str(self.number_of_valid_pulses()),
-                          "Number of invalid pulses:" + str(self.number_of_invalid_pulses()),
-                          "Start time:" + str(self.start_time()),
-                          "End time:" + str(self.end_time())])
-
-    def add(self, a_pdw):
-        # print(a_pdw)
-        self.signals.append(a_pdw)
-
-    def as_pulse_dict(self):
-        pulse_dict = {}
-        pulse_dict[0] = self.signals[0].to_json()
-        for index, current_pdw in enumerate(self.signals[1:]):
-            pulse_dict[index+1] = current_pdw.to_json()
-            pulse_dict[index]["dtoa"] = (current_pdw.time_of_arrival - pulse_dict[index]["time of arrival"]).item()/1000.0
-        pulse_dict[index+1]["dtoa"] = 0.0  # np.timedelta64(0, 'us')
-        return pulse_dict
-
-    def number_of_valid_pulses(self):
-        return sum(each.is_valid() for each in self.signals)
-
-    def number_of_invalid_pulses(self):
-        return sum(not each.is_valid() for each in self.signals)
-
-    def without_invalids(self):
-        return PPDWContainer(self.name, [each for each in self.signals if each.is_valid()])
-
-    def julian_date_string(self):
-        ts = (self.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's')
-        time_tuple = datetime.utcfromtimestamp(ts).timetuple()
-        return str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3)
-
-    def center_frequencies(self):
-        return list({each.center_frequency for each in self.signals})
-
-    def channels(self):
-        return list({each.channel for each in self.signals})
-
-    def modulations(self):
-        return list({each.modulation for each in self.signals})
-
-    def bandwidths(self):
-        return list({each.frequency_shift_or_bandwidth for each in self.signals})
-
-    def start_time(self):
-        return min(each.time_of_arrival for each in self.signals)
-
-    def end_time(self):
-        return max(each.time_of_arrival for each in self.signals)
-
-    def to_json(self):
-        return {'JULIANDATE': self.julian_date_string(),
-                'STARTTIME': str(self.start_time()),
-                'ENDTIME': str(self.end_time()),
-                'CENTERFREQUENCIES': self.center_frequencies(),
-                'CHANNELS': self.channels(),
-                'MODULATIONS': self.modulations(),
-                'BANDWIDTHS': self.bandwidths()
-                }
-
-if __name__ == '__main__':
-    pass
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class PPDWContainer -(name, signals=None) -
-
-

I store multiple signals imported from one or more .ppdw files

-
-

Automethod: init

-
-
- -Expand source code - -
class PPDWContainer():
-    """
-    I store multiple signals imported from one or more .ppdw files
-    .. automethod:: __init__
-    """
-
-    def __init__(self, name, signals=None):
-
-        if signals is None:
-            self.signals = []
-        else:
-            self.signals = signals
-
-        self.name = name
-
-    def __str__(self):
-        return "\n".join(["Number of pulses:" + str(len(self.signals)),
-                          "Number of valid pulses:" + str(self.number_of_valid_pulses()),
-                          "Number of invalid pulses:" + str(self.number_of_invalid_pulses()),
-                          "Start time:" + str(self.start_time()),
-                          "End time:" + str(self.end_time())])
-
-    def add(self, a_pdw):
-        # print(a_pdw)
-        self.signals.append(a_pdw)
-
-    def as_pulse_dict(self):
-        pulse_dict = {}
-        pulse_dict[0] = self.signals[0].to_json()
-        for index, current_pdw in enumerate(self.signals[1:]):
-            pulse_dict[index+1] = current_pdw.to_json()
-            pulse_dict[index]["dtoa"] = (current_pdw.time_of_arrival - pulse_dict[index]["time of arrival"]).item()/1000.0
-        pulse_dict[index+1]["dtoa"] = 0.0  # np.timedelta64(0, 'us')
-        return pulse_dict
-
-    def number_of_valid_pulses(self):
-        return sum(each.is_valid() for each in self.signals)
-
-    def number_of_invalid_pulses(self):
-        return sum(not each.is_valid() for each in self.signals)
-
-    def without_invalids(self):
-        return PPDWContainer(self.name, [each for each in self.signals if each.is_valid()])
-
-    def julian_date_string(self):
-        ts = (self.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's')
-        time_tuple = datetime.utcfromtimestamp(ts).timetuple()
-        return str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3)
-
-    def center_frequencies(self):
-        return list({each.center_frequency for each in self.signals})
-
-    def channels(self):
-        return list({each.channel for each in self.signals})
-
-    def modulations(self):
-        return list({each.modulation for each in self.signals})
-
-    def bandwidths(self):
-        return list({each.frequency_shift_or_bandwidth for each in self.signals})
-
-    def start_time(self):
-        return min(each.time_of_arrival for each in self.signals)
-
-    def end_time(self):
-        return max(each.time_of_arrival for each in self.signals)
-
-    def to_json(self):
-        return {'JULIANDATE': self.julian_date_string(),
-                'STARTTIME': str(self.start_time()),
-                'ENDTIME': str(self.end_time()),
-                'CENTERFREQUENCIES': self.center_frequencies(),
-                'CHANNELS': self.channels(),
-                'MODULATIONS': self.modulations(),
-                'BANDWIDTHS': self.bandwidths()
-                }
-
-

Methods

-
-
-def add(self, a_pdw) -
-
-
-
- -Expand source code - -
def add(self, a_pdw):
-    # print(a_pdw)
-    self.signals.append(a_pdw)
-
-
-
-def as_pulse_dict(self) -
-
-
-
- -Expand source code - -
def as_pulse_dict(self):
-    pulse_dict = {}
-    pulse_dict[0] = self.signals[0].to_json()
-    for index, current_pdw in enumerate(self.signals[1:]):
-        pulse_dict[index+1] = current_pdw.to_json()
-        pulse_dict[index]["dtoa"] = (current_pdw.time_of_arrival - pulse_dict[index]["time of arrival"]).item()/1000.0
-    pulse_dict[index+1]["dtoa"] = 0.0  # np.timedelta64(0, 'us')
-    return pulse_dict
-
-
-
-def bandwidths(self) -
-
-
-
- -Expand source code - -
def bandwidths(self):
-    return list({each.frequency_shift_or_bandwidth for each in self.signals})
-
-
-
-def center_frequencies(self) -
-
-
-
- -Expand source code - -
def center_frequencies(self):
-    return list({each.center_frequency for each in self.signals})
-
-
-
-def channels(self) -
-
-
-
- -Expand source code - -
def channels(self):
-    return list({each.channel for each in self.signals})
-
-
-
-def end_time(self) -
-
-
-
- -Expand source code - -
def end_time(self):
-    return max(each.time_of_arrival for each in self.signals)
-
-
-
-def julian_date_string(self) -
-
-
-
- -Expand source code - -
def julian_date_string(self):
-    ts = (self.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's')
-    time_tuple = datetime.utcfromtimestamp(ts).timetuple()
-    return str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3)
-
-
-
-def modulations(self) -
-
-
-
- -Expand source code - -
def modulations(self):
-    return list({each.modulation for each in self.signals})
-
-
-
-def number_of_invalid_pulses(self) -
-
-
-
- -Expand source code - -
def number_of_invalid_pulses(self):
-    return sum(not each.is_valid() for each in self.signals)
-
-
-
-def number_of_valid_pulses(self) -
-
-
-
- -Expand source code - -
def number_of_valid_pulses(self):
-    return sum(each.is_valid() for each in self.signals)
-
-
-
-def start_time(self) -
-
-
-
- -Expand source code - -
def start_time(self):
-    return min(each.time_of_arrival for each in self.signals)
-
-
-
-def to_json(self) -
-
-
-
- -Expand source code - -
def to_json(self):
-    return {'JULIANDATE': self.julian_date_string(),
-            'STARTTIME': str(self.start_time()),
-            'ENDTIME': str(self.end_time()),
-            'CENTERFREQUENCIES': self.center_frequencies(),
-            'CHANNELS': self.channels(),
-            'MODULATIONS': self.modulations(),
-            'BANDWIDTHS': self.bandwidths()
-            }
-
-
-
-def without_invalids(self) -
-
-
-
- -Expand source code - -
def without_invalids(self):
-    return PPDWContainer(self.name, [each for each in self.signals if each.is_valid()])
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/PPDWReader.html =================================================================== --- doc/ammosreader/PPDWReader.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,172 +1,0 @@ - - - - - - -ammosreader.PPDWReader API documentation - - - - - - - - - - - -
-
-
-

Module ammosreader.PPDWReader

-
-
-
- -Expand source code - -
from ammosreader.PDW import PDW
-from ammosreader.PPDWContainer import PPDWContainer
-from ammosreader import logger
-
-
-class PPDWReader():
-    """I read data from a file and return a PPDWContainer."""
-
-    def __init__(self, a_file):
-
-        self.file = a_file
-        assert self.file.is_file()
-        self.content = self.file.read_bytes()
-        self.cursor = 0
-        self.container = PPDWContainer(self.file.stem)
-
-    def read_all_frames_left(self):
-        try:
-            while self.cursor <= len(self.content) - 32:
-                current_bytes = self.content[self.cursor:self.cursor+32]
-                assert len(current_bytes) == 32
-                if not current_bytes:
-                    logger.info('End of file detected')
-                    break
-                if self.cursor + 32 >= len(self.content):
-                    logger.info('Can not read all 32 bytes of next PDW. EOF')
-                    break
-                self.container.add(PDW.from_bytes(current_bytes))
-                self.cursor += 32
-        except Exception:
-            return None
-        return self.container
-
-
-
-
-
-
-
-
-
-

Classes

-
-
-class PPDWReader -(a_file) -
-
-

I read data from a file and return a PPDWContainer.

-
- -Expand source code - -
class PPDWReader():
-    """I read data from a file and return a PPDWContainer."""
-
-    def __init__(self, a_file):
-
-        self.file = a_file
-        assert self.file.is_file()
-        self.content = self.file.read_bytes()
-        self.cursor = 0
-        self.container = PPDWContainer(self.file.stem)
-
-    def read_all_frames_left(self):
-        try:
-            while self.cursor <= len(self.content) - 32:
-                current_bytes = self.content[self.cursor:self.cursor+32]
-                assert len(current_bytes) == 32
-                if not current_bytes:
-                    logger.info('End of file detected')
-                    break
-                if self.cursor + 32 >= len(self.content):
-                    logger.info('Can not read all 32 bytes of next PDW. EOF')
-                    break
-                self.container.add(PDW.from_bytes(current_bytes))
-                self.cursor += 32
-        except Exception:
-            return None
-        return self.container
-
-

Methods

-
-
-def read_all_frames_left(self) -
-
-
-
- -Expand source code - -
def read_all_frames_left(self):
-    try:
-        while self.cursor <= len(self.content) - 32:
-            current_bytes = self.content[self.cursor:self.cursor+32]
-            assert len(current_bytes) == 32
-            if not current_bytes:
-                logger.info('End of file detected')
-                break
-            if self.cursor + 32 >= len(self.content):
-                logger.info('Can not read all 32 bytes of next PDW. EOF')
-                break
-            self.container.add(PDW.from_bytes(current_bytes))
-            self.cursor += 32
-    except Exception:
-        return None
-    return self.container
-
-
-
-
-
-
-
- -
- - - Index: c/ammosreader/index.html =================================================================== --- doc/ammosreader/index.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,188 +1,0 @@ - - - - - - -ammosreader API documentation - - - - - - - - - - - -
-
-
-

Package ammosreader

-
-
-
- -Expand source code - -
import logging
-import logging.config
-import os
-from pathlib import Path
-
-# FIXME: change the default log path to /var/log/ammos/ammos.log when system configuration created this dir
-# with the appropriate rights
-
-log_dir = Path(os.environ.get('AMMOS_LOG_DIR', '/tmp/'))
-
-log_path = log_dir / 'ammos.log'
-
-if not log_path.exists():
-    try:
-        log_path.touch()
-    except PermissionError:
-        conf_file = Path(__file__).parent / 'ammos_logging.conf'
-        print("Conf file", conf_file)
-        logging.config.fileConfig(conf_file)
-else:
-    logging.basicConfig(filename=str(log_path), encoding='utf-8', level=logging.DEBUG)
-
-logger = logging.getLogger(__name__)
-
-
-
-

Sub-modules

-
-
ammosreader.AbstractAmmosReader
-
-

I provide a base class for specialized AmmosReaders.

-
-
ammosreader.AmmosAudioDataBody
-
-

I provide an AMMOS data block for audio data frames.

-
-
ammosreader.AmmosAudioDataHeader
-
-

I provide an AMMOS data header for audio data frames.

-
-
ammosreader.AmmosAudioReader
-
-

I provide a specialized Ammos Reader for audio data.

-
-
ammosreader.AmmosAudioSocketReader
-
-

I read a Ammos datastream from a socket.

-
-
ammosreader.AmmosConstants
-
-

I provide several constants used in R&S software.

-
-
ammosreader.AmmosContainer
-
-

I provide an Ammos container.

-
-
ammosreader.AmmosExtendedAudioDataHeader
-
-

I provide an Ammos extended data header for audio data frames.

-
-
ammosreader.AmmosExtendedIFDataHeader
-
-
-
-
ammosreader.AmmosGlobalFrameBody
-
-

I provide an AMMOS global frame body.

-
-
ammosreader.AmmosGlobalFrameHeader
-
-

I provide an AMMOS global frame header.

-
-
ammosreader.AmmosIFDataBlock
-
-
-
-
ammosreader.AmmosIFDataBlockHeader
-
-
-
-
ammosreader.AmmosIFDataBody
-
-

I provide an AMMOS data body for IF data …

-
-
ammosreader.AmmosIFDataHeader
-
-

I provide a Ammos data header for IF data frames.

-
-
ammosreader.AmmosIFReader
-
-

I provide a specialized Ammos Reader for IF data.

-
-
ammosreader.AmmosSingleFrame
-
-
-
-
ammosreader.IQDWTXBlock
-
-
-
-
ammosreader.PDW
-
-

I store the information of a single PDW block.

-
-
ammosreader.PPDWContainer
-
-
-
-
ammosreader.PPDWReader
-
-
-
-
-
-
-
-
-
-
-
-
- -
- - - Index: g/FrameTypes.org =================================================================== --- org/FrameTypes.org (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,112 +1,0 @@ -/-------------------------------------- -//--- 010 Editor v2.0.2 Binary Template -// -// File: -// Author: -// Revision: -// Purpose: -//-------------------------------------- -enum typDATASTREAM_TYPES -{ - /* Type for test data */ - ekFRH_DATASTREAM__TEST_DATA = 0x00, /**< CURRENTLY NOT USED */ - - /* Type for IF data 32Bit Real- and 32Bit Imaginary-part, fixed point, not rescaled */ - ekFRH_DATASTREAM__IFDATA_32RE_32IM_FIX = 0x01, /**< IF data : struct < typIFD_IFDATA_FRAME */ - - /* Type for IF data 16Bit Real- and 16Bit Imaginary-part, fixed point, not rescaled */ - ekFRH_DATASTREAM__IFDATA_16RE_16IM_FIX = 0x02, /**< IF data : struct < typIFD_IFDATA_FRAME */ - - /* Type for IF data 16Bit Real-part, two samples for each word, fixed point, not rescaled */ - ekFRH_DATASTREAM__IFDATA_16RE_16RE_FIX = 0x03, /**< IF data : struct < typIFD_IFDATA_FRAME */ - - /* Type for IF data 32Bit Real- and 32Bit Imaginary-part, fixed point, rescaled */ - ekFRH_DATASTREAM__IFDATA_32RE_32IM_FIX_RESCALED = 0x04, /**< IF data : struct < typIFD_IFDATA_FRAME */ - - /* Type for IF data 32Bit Real- und 32Bit Imaginary-part, floating point, rescaled */ - ekFRH_DATASTREAM__IFDATA_32RE_32IM_FLOAT_RESCALED = 0x05, /**< IF data : struct < typIFD_IFDATA_FRAME */ - - /* Type for Spectrum-data 8Bit */ - ekFRH_DATASTREAM__SPECDATA_8BIT = 0x10, /**< CURRENTLY NOT USED */ - - /* Type for Spectrum-data 16Bit */ - ekFRH_DATASTREAM__SPECDATA_16BIT = 0x11, /**< CURRENTLY NOT USED */ - - /* Type for Spectrum-data 32Bit */ - ekFRH_DATASTREAM__SPECDATA_32BIT = 0x12, /**< CURRENTLY NOT USED */ - - /* Type for Spectrum-data 32Bit float */ - ekFRH_DATASTREAM__SPECDATA_FLOAT = 0x13, /**< Spectrum data : struct < typSPECDATA_FLOAT */ - - /* Type for Segmentation Spectrum-data 32Bit float */ - ekFRH_DATASTREAM__SEGMENTATION_SPECDATA_FLOAT = 0x14, /**< Segmentation spectrum data : struct < typSEGMENTATION_SPECDATA_FLOAT */ - - /* Data stream types for Tuner HF (EM010) data */ - ekFRH_DATASTREAM__HF_TUNING_INDICATOR_DATA = 0x20, /**< Tuning Indicator data : struct < typFRAME_EM010_HF_TUNING_INDICATOR_AND_STATUS_DATA*/ - - ekFRH_DATASTREAM__HF_SCF_DATA = 0x22, /**< Scan Channel Found : struct < typFRAME_EM010_HF_SCAN_CHANNEL_FOUND */ - - ekFRH_DATASTREAM__HF_SFF_DATA = 0x23, /**< Scan Frequency Found : struct < typFRAME_EM010_HF_SCAN_FREQUENCY_FOUND */ - - ekFRH_DATASTREAM__HF_SSR_DATA = 0x24, /**< Scan Sweep Restarted : struct < typFRAME_EM010_HF_SCAN_RESTARTED */ - - /* Type for HF Single Channel Demodulation*/ - ekFRH_DATASTREAM__DEMOD_HF_SYMBOLSTREAM_DATA = 0x30, /**< HF Demod data (symbol stream) : struct < struDEMOD_HF_SYMBOL_DATA*/ - - /* Type for(ACID) HF Single Channel Decoding*/ - ekFRH_DATASTREAM__DECODER_TEXT_DATA = 0x40, /**< Decoder (ACID) decoded text data : struct < typDECODER__DECODED_TXT_FRAME*/ - - /* Types for HF SingleChannel Realtime Visualisation*/ - ekFRH_DATASTREAM__VISUALISATION_HF_SPECTRUM_DATA = 0x50, /**< RTVIS spectrum data : struct < struVISUALISATION_HF_SPECTRUM_DATA*/ - - ekFRH_DATASTREAM__VISUALISATION_HF_TIMEDOMAIN_DATA = 0x51, /**< RTVIS timedomain data : struct < struVISUALISATION_HF_TIMESIGNAL_DATA*/ - - /* Type for audio data */ - ekFRH_DATASTREAM__AUDIODATA = 0x100, /**< Audio data : struct < typAUDIODATAFRAME */ - - /* Type for signal emissions list data */ - ekFRH_DATASTREAM__EMISSION_LIST_DATA = 0x110, /**< Signal emission list data : struct < typEMISSION_LIST_FRAME */ - - /* Type for level indicator data */ - ekFRH_DATASTREAM__LEVELDATA = 0x120, /**< Level indicator data : struct < typLEVELDATA_FRAME */ - - /* Type for demodulation result data (symbol stream) */ - ekFRH_DATASTREAM__SYMBOLDATA = 0x130, /**< Demodulation result data (symbol stream) : struct < typSYMBOLDATAFRAME */ - - /* Type for demodulation instantaneous data */ - ekFRH_DATASTREAM__INSTANTANEOUSDATA = 0x140, /**< Demodulation instantaneous data : struct < typINSTANTANEOUSDATAFRAME */ - - /* Type for burst emission list data */ - ekFRH_DATASTREAM__BURST_EMISSIONS_LIST = 0x150, /**< Burst emission list data : struct < typBURST_EMISSION_LIST_DATAFRAME */ - - /* Type for demodulation result data (image stream) */ - ekFRH_DATASTREAM__IMAGEDATA = 0x160, /**< Demodulation result data (image stream) : struct < typIMAGEDATAFRAME */ - - /* Type for transmission system result data */ - ekFRH_DATASTREAM__TRANSMISSION_SYSTEM_RESULT_DATA = 0x170, /**< Transmission system result data : struct < typTRANSMISSION_SYSTEM_RESULT_FRAME */ - - /* Type for pulse description words */ - ekFRH_DATASTREAM__PULSE_DESCRIPTION_WORD_DATA = 0x200, /**< Pulse description word data : struct < TODO */ - - /* Type for pulse repetition words */ - ekFRH_DATASTREAM__PULSE_REPETITION_WORD_DATA = 0x210, /**< Pulse repetition word data : struct < TODO */ - - /* Type for scan datastream - no content */ - ekFRH_DATASTREAM__SCAN = 0x4000, /**< Scan data for EM050 - reserved area from 0x4000 to 0x43ff */ - - /* Type for scan datastream -- tuning level indicator data */ - ekFRH_DATASTREAM__SCAN__LEVEL = 0x4001, /**< Scan datastream -- tuning level indicator data : struct < typSCANDATA_FRAME__LEVEL */ - - /* Type for scan for datastream -- frequency tuning offset */ - ekFRH_DATASTREAM__SCAN__TUNING = 0x4002, /**< Scan datastream -- frequency tuning offset : struct < typSCANDATA_FRAME__TUNING */ - - /* Type for scan datastream -- tuning level indicator and frequency tuning offset */ - ekFRH_DATASTREAM__SCAN__LEVEL_TUNING = 0x4003, /**< Scan datastream -- tuning level indicator data and frequency tuning offset : struct < typSCANDATA_FRAME__LEVEL_TUNING */ - - /* DDF datastreams - no content */ - ekFRH_DATASTREAM__DDF_RESERVED_RANGE_BEGIN = 0x5000, /**< DDF data streams - reserved area from 0x5000 to 0x50ff */ - - /* DDF datastreams - no content */ - ekFRH_DATASTREAM__DDF_RESERVED_RANGE_END = 0x50FF - -}; Index: g/README.md =================================================================== --- org/README.md (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,62 +1,0 @@ -# AmmosReader - -Project containing necessary File- or SocketStream-Reader to read the -various datastreams of Ammos - -# Installation - -## Standard pip install (not tested on all platforms) - -``` shell -git clone http://gitlab.kid.local/kidzg/ammosreader.git -cd ammosreader -python3 -m pip install . -``` - -## Using the guix functional package manager - -First subscribe to the GUIX oai channel (see Chapter Add channel to -system) in -) - -Then install using the guix - -``` shell -guix package -i python-ammosreader -``` - -# Usage - -There are two scripts, which demonstrate the usage of the ammosreader -package. - -## iqdw_reader.py - -``` shell -python3 iqdw_reader.py 'path_to_iqdw_file' -``` - -## pdw_reader.py - -``` shell -python3 pdw_reader.py 'path_to_pdw_file' -``` - -To minimize depencencies of the core package other scripts -(audio_reader, ammos_viewer) are now stored in separate packages. - -# Logging - -If you use ammosreader in your own scripts a logger can be imported and -used as follows: - -``` python -from ammosreader import logger -logger.warning("Put your warning here") -``` - -The name of the logfile is ammos.log The logger uses the directory -stored in the AMMOS_LOG_DIR environment variable. If this variable is -not set, /tmp is used as default. If the logfile is not writable, the -standard logger config is used and the logs go directly to console -(tty). (see ammos_logging.conf file in this repository) Index: g/README.org =================================================================== --- org/README.org (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,75 +1,0 @@ -#+OPTIONS: ^:nil - -* AmmosReader - -Project containing necessary File- or SocketStream-Reader to read the various datastreams of Ammos - -* Installation - -** Standard pip install (not tested on all platforms) - -#+BEGIN_SRC shell -git clone http://gitlab.kid.local/kidzg/ammosreader.git -cd ammosreader -python3 -m pip install . -#+END_SRC - -** Using the guix functional package manager - -First subscribe to the GUIX oai channel (see Chapter Add channel to system) in -https://gitlab.kid.local/kidzg/guix-deployment/-/blob/master/DevOps.org) - -Then install using the guix - -#+BEGIN_SRC shell -guix package -i python-ammosreader -#+END_SRC - -* Usage - -There are two scripts, which demonstrate the usage of the ammosreader package. - -** iqdw_reader.py - -#+BEGIN_SRC shell -python3 iqdw_reader.py 'path_to_iqdw_file' -#+END_SRC - -** pdw_reader.py - -#+BEGIN_SRC shell -python3 pdw_reader.py 'path_to_pdw_file' -#+END_SRC - -To minimize depencencies of the core package other scripts (audio_reader, ammos_viewer) -are now stored in separate packages. - -* Logging - -If you use ammosreader in your own scripts a logger can be imported and used as follows: - -#+BEGIN_SRC python -from ammosreader import logger -logger.warning("Put your warning here") -#+END_SRC - -** Log-File - -The name of the logfile is ammos.log -The logger uses the directory stored in the AMMOS_LOG_DIR environment variable. -If this variable is not set, /tmp is used as default. This might change to /var/log/ -in future versions running under GUIX system. -If the logfile is not writable, the standard logger config is used and -the logs go directly to console (tty). -(see ammos_logging.conf file in this repository) - -** Log-Level - -You can change the log-Level by setting the environment variable AMMOS_LOG_LEVEL. -To turn off logging completely set the level to 'CRITICAL' - -#+BEGIN_SRC shell -export AMMOS_LOG_LEVEL=CRITICAL -#+END_SRC - -If the environment variable does not exist, the logging level from ammos_logging.conf is used. Index: g/audio.html =================================================================== --- org/audio.html (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,183 +1,0 @@ - - - - -Audio - - - - - - - -
-
-
-

Audio

Enrico Schwass

Created: 2022-05-03 Tue 13:09

-
-
- -
- -
-
-

1. Description Audio sample file AudioAufzeichnungLOW.bin

-

-Little Endian!!! -

- -
-
-

1.1. Global Frame Header (6 x 32 Bit) Offset 0x00000000

-
-
-
-
-

1.1.1. 32 Bit Magic word 0x726574fb = 0xfb746572 next magic bytes found at offsets

-

-0x0000 (0000) -0x0544 (1348) -0x0a88 (2696) -0x0fcf (4047) -0x1510 (5392) -0x1a57 (6743) -0x1f9b (8091) -0x24df (9439) -

- -
-
-

1.1.2. 32 Bit Frame length in 32 bit words

-

-0x51010000 = 0x00000151 = Integer 337 * 4 bytes = 1348 Bytes = 0x0544 -

- -
-
-

1.1.3. 32 Bit Running Frame number

-

-0x23450000 = 0x00004523 = Integer 17699 -

- -
-
-

1.1.4. 32 Bit Frame Type

-

-0x00010000 = 0x00000100 = Integer 256 -

- -
-
-

1.1.5. 32 Bit Data header length in 32 bit words

-

-0x0b000000 = 0x0000000b = Integer 11 * 4 = 44 Bytes -

- -
-
-

1.1.6. 32 Bit Reserved

-

-0x00000000 = 0x00000000 = Integer 0 -

- -
-
-

1.2. Ammos Data header 44 bytes

-

-4 Bytes Sample rate -4 Bytes Status -8 Bytes Frequency -4 Bytes Demod Bandwidth -4 Bytes Demod Type -4 Bytes Sample Count -4 Bytes Channel Count -4 Bytes Sample Size -

- -

-8 Bytes timestamp if extended -

- -

-0xd4300000 = 0x000030d4 = 12500 Sample rate -0x00000000 = 0x00000000 = 0 Squelch Status -0xc0d3f205 0x00000000 = 0x00000000 0x05f2d3c0 = 99800000 = 99.8 MHz -0xc0d40100 = 0x0001d4c0 = 120000 = 120 KHz -0x00000000 = 0x00000000 = 0 = FM -0x80020000 = 0x00000280 = Integer 640 = Sample Count per Channel -0x01000000 = 0x00000001 = 1 = 1 Channel -0x02000000 = 0x00000002 = 2 = 2 Bytes Sample size -0x1d26fe89 0xcada0500 = 0x0005daca 0x89fe261d = 1647938316938525 microseconds Unixtime = 2022-03-22T09:38:36 -

- -
-
-

1.3. Ammos Audio Data Body

-

-Offset 0x0044 = Integer 68 - Next offset 1348 - Differenz = 1280 -

-
-
-
-
- - - - - - - - Index: g/audio.org =================================================================== --- org/audio.org (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,65 +1,0 @@ -#+title: Audio - -* Description Audio sample file AudioAufzeichnungLOW.bin - -Little Endian!!! - -** Global Frame Header (6 x 32 Bit) Offset 0x00000000 - -*** 32 Bit Magic word 0x726574fb = 0xfb746572 next magic bytes found at offsets - -0x0000 (0000) -0x0544 (1348) -0x0a88 (2696) -0x0fcf (4047) -0x1510 (5392) -0x1a57 (6743) -0x1f9b (8091) -0x24df (9439) - -*** 32 Bit Frame length in 32 bit words - -0x51010000 = 0x00000151 = Integer 337 * 4 bytes = 1348 Bytes = 0x0544 - -*** 32 Bit Running Frame number - -0x23450000 = 0x00004523 = Integer 17699 - -*** 32 Bit Frame Type - -0x00010000 = 0x00000100 = Integer 256 - -*** 32 Bit Data header length in 32 bit words - -0x0b000000 = 0x0000000b = Integer 11 * 4 = 44 Bytes - -*** 32 Bit Reserved - -0x00000000 = 0x00000000 = Integer 0 - -** Ammos Data header 44 bytes - -4 Bytes Sample rate -4 Bytes Status -8 Bytes Frequency -4 Bytes Demod Bandwidth -4 Bytes Demod Type -4 Bytes Sample Count -4 Bytes Channel Count -4 Bytes Sample Size - -8 Bytes timestamp if extended - -0xd4300000 = 0x000030d4 = 12500 Sample rate -0x00000000 = 0x00000000 = 0 Squelch Status -0xc0d3f205 0x00000000 = 0x00000000 0x05f2d3c0 = 99800000 = 99.8 MHz -0xc0d40100 = 0x0001d4c0 = 120000 = 120 KHz -0x00000000 = 0x00000000 = 0 = FM -0x80020000 = 0x00000280 = Integer 640 = Sample Count per Channel -0x01000000 = 0x00000001 = 1 = 1 Channel -0x02000000 = 0x00000002 = 2 = 2 Bytes Sample size -0x1d26fe89 0xcada0500 = 0x0005daca 0x89fe261d = 1647938316938525 microseconds Unixtime = 2022-03-22T09:38:36 - -** Ammos Audio Data Body - -Offset 0x0044 = Integer 68 - Next offset 1348 - Differenz = 1280 Index: g/iqdw.org =================================================================== --- org/iqdw.org (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,157 +1,0 @@ -* Description IQDW sample file 17151-1.iqdw - -Little Endian!!! - - -** Global Frame Header (6 x 32 Bit) Offset 0x00000000 - -*** 32 Bit Magic word 0x726574fb = 0xfb746572 next magic bytes found at offsets - - 0x0000 (0000) - 0x019c (412) - 0x037c (892) - 0x0574 (1396) - 0x0880 (2176) - 0x0be0 (3040) - 0x1010 (4112) - 0x1454 (5204) - - -*** 32 Bit Frame length in 32 bit words - - 0x67000000 = 0x00000067 = Integer 103 * 4 bytes = 412 Bytes = 0x019c - - -*** 32 Bit Running Frame number - - 0xd3122800 = 0x002812d3 = Integer 2626259 - - -*** 32 Bit Frame Type - - 0x02000000 = 0x00000002 = Integer 2 - - -*** 32 Bit Data Header length in 32 bit words - - 0x13000000 = 0x00000013 = Integer 19 * 4 = 76 Bytes - - -*** 32 Bit Reserved - - 0x01000000 = 0x00000001 = Integer 1 - - -** Ammos Data header 76 bytes - -4 Bytes Block count -4 Bytes Block Length -8 Bytes timestamp - -0x203546b4 0x00000000 = 0x00000000 0xb4463520 = possible center frequency 3.024.500.000 Hz -0x286bee00 = 0x00ee6b28 = possible bandwidth 15.625.000 Hz -0x00a02526 = 0x2625a000 = possible sample rate 640000000 Hz - -** Tx Block (Frame length - Data Header - Global Frame header) - -*** 32 Bytes PDW -*** n samples * 4 bytes - -** PDW Data Header (2 x 32 Bit = 8 bytes) Offset 0x00000018 - -*** 32 Bit Number of PDWs in Frame - - 0x01000000 = 0x00000001 = Integer 1 - -*** 32 Bit Size of tx block in 32 bit words - - 0x4d000000 = 0x0000004D = 77 * 4 = 308 Bytes - - -** PDW Data Body (8 x 32 Bit = 32 bytes) Offset 0x00000050 next is 1ec (diff = 412) - -*** 64 Bit Unix time - - next time stamp at 01bc = 144 bytes - - 0x52b0837b - 0x0492c414 - - vs - - 0x14c49204 - 0x7b83b052 - - Integer 1496481524143599698 nanoseconds = 2017-06-03 09:18:44 - -*** 32 Bit Center Frequency - - 0x00000000 = 0x00000000 = Integer 0 - no valid frequency - -*** 32 Bit Flags/Marks - - 0x00000000 = 0x00000000 = Integer 0 - invalid - -*** 32 Bit BW/Level (00000000) -32 Bit IO_DFE_Mod_Chip_Low (b4463520) -32 Bit POL_DFQ_EOA_AOA_Low (00000000) -32 Bit Reserved (00ee6b28) - -*** 32 Bit IO_DFE_Mod_Chip_Low - -1-011010-00100-0110001101010010-0000 - -IO Flag - 1 (inside of region of interest) -Azimuth confidence 011010 (0.0-6.2 degree in steps of 0.1 degree) -Modulation ID 00100 (PSK-2) -Reserved 0110001101010010 -Sector reference 0000 - -*** 32 Bit POL_DFQ_EOA_AOA_Low - -to be done - -*** 32 Bit Reserved and channel number - -000000-00111011100110101100101000 - -Number of channel (000000) - -to be done - - -** Offset 0x00000040 - -** 348 Bytes unidentified -*** 12*4 AMMOS Data header? 48 Bytes -*** 32 Bytes PDW (see ppdw.org) -*** n * 4 bytes IQ samples (maybe n=67) - -** Next global frame header Offset 0x0000019c = Integer 412 - -** AMMOS Frame header (24 bytes) - - -32 Bit Magic Word (fb746572) -32 Bit Frame length -32 Bit Running Frame number -32 Bit Frame Type (00000002) -32 Bit Data Header length -32 Bit Reserved (00000001) - - -** AMMOS Data header (48 bytes) - -64 Bit Frequency - -Low 0x00a02526 = 0x2625a000 -High 0x01000000 = 0x00000001 - -Integer = 4.934967296 GHz - -32 Bit Bandwidth (00000020) 32 Hz -32 Bit Sample rate (000004bd) 1213 Hz -32 Bit Interpolation / Decimation (7b83b052) -32 Bit Antenna Voltage Ref (14c49204) -32 Bit Start time low (00000000) -32 Bit Start time high (00000000) Index: g/ppdw.org =================================================================== --- org/ppdw.org (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,95 +1,0 @@ -* General structure of a PPDW sample file 17151-1.ppdw - - 1. PPDW files include only the data bodies of a signal, but neither headers nor iq data) - 2. size of ppdw file = number of pulses * 8 * 32 bit - 3. data like integers are stored little endian - -* PDW Body (8 words - 32 bit each word) - -** Word 1 and 2 - -*** 64 bit timestamp - - 0x60b6837b0492c414 = 0x14c492047b83b660 = Integer 1496481524143601248 nanoseconds = 2017-06-03 09:18:44 - - -** Word 3 - -*** 32 bit format and center frequency - - 0x0a212e00 = 0x002e210a = Integer 3023114 KHz = 3.023114 GHz - - 00000-000-00101110-00100001-00001010 - - -** Word 4 - -*** 32 bit flags and pulse width in nanoseconds (0-24) - - 0xbc020078 = 0x780002bc = 0-1-1-1-1-00-0-00000000-00000010-10111100 - - 0 - valid flag - invalid - 1 - pulse flag - pulse - 1 - level unit - dbµV - 1 - signal no start - signal started before time of arrival - 1 - signal no end - signal stops after time of arrival - 00 - reserved - - Pulse width = 0-00000000-00000010-10111100 = Integer 700 nano seconds - - -** Word 5 - -*** 32 bit frequency shift or bandwidth and level - - 0xf9023a00 = 0x003a02f9 = 00000000-00111010-0000-001011111001 - - Frequency shift (12-31) 00000000-00111010-0000 = Integer 928 KHz - Level = (0-11) 001011111001 = 761 dbµV - - -** Word 6 - -*** 32 bit IO, confidence, modulation, sector - - 0x0000b07e = 0x7eb00000 = 0-111111-01011-0000000000000000-0000 - - 0 - Signal invalid - - 111111 - confidence not valid - - 01011 - modulation (pulse too short) - - 0000000000000000 - reserved - - 0000 - sector invalid - - -** Word 7 - -*** 32 bit polarity, quality, elevation, azimuth - - 0xff0f4000 = 0x00400fff = 00-0000000-10000000000-11111111-1111 - - 00 - polarity - horizontal or unknown - 0000000 - quality zero - 10000000000 - elevation = invalid - 11111111-1111 - azimuth = no azimuth - - -** Word 8 - -*** 32 bit reserved and channel number - - 0x00000010 = 0x10000000 = 0001-0000-00000000-00000000-00000000 - - 0001 - channel one - - rest zero - reserved - - -** Word 1 and 2 of next data body - -*** 64 bit timestamp - - 0xcb5d927b0492c414 = 0x14c492047b925dcb = Integer 149648152414561611 seconds = 2017-06-03 09:18:44 Index: project.toml =================================================================== --- pyproject.toml (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,30 +1,0 @@ -[build-system] -requires = ["poetry-core>=1.0.7"] -build-backend = "poetry.core.masonry.api" - -[tool.poetry] -name = "ammosreader" -version = "0.0.5" -description = "A package to manage ammos datastreams" -authors = ["See Contributors"] -homepage = "https://gitlab.kokyou.de/enno/ammosreader" -repository = "https://gitlab.kokyou.de/enno/ammosreader" -license = "MIT" -readme = "README.md" -packages = [ - { include = "ammosreader" } -] - -include = [ - {path = 'tests/*.py'}, - {path = 'scripts/*.py'}, - {path = 'sampleData/*.csv'}, - {path = 'ammos_logger.conf'} -] -[tool.poetry.scripts] -iqdw_reader = 'scripts.iqdw_reader:main' -pdw_reader = 'scripts.pdw_reader:main' - -[tool.poetry.dependencies] -numpy = "~1.23.2" -bitstring = "~4.2.0" Index: ripts/iqdw_reader.py =================================================================== --- scripts/iqdw_reader.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,22 +1,0 @@ -import sys - -from ammosreader.AmmosIFReader import AmmosIFReader - -def main(): - if len(sys.argv) != 2: - sys.exit() - - file_name = sys.argv[1] - - print("File name:", file_name) - - dat_file = AmmosIFReader(file_name) - - dat_file.read_all_frames_left() - print("Frequencies", dat_file.container.frequencies()) - print("Frame types:", dat_file.container.frame_types()) - print("Total frame sizes:", sum(dat_file.container.frame_sizes())) - print("Homogenic:", dat_file.container.is_homogenic()) - -if __name__ == "__main__": - main() Index: ripts/pdw_reader.py =================================================================== --- scripts/pdw_reader.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,28 +1,0 @@ -import sys - -from ammosreader.PDW import PDW - -def main(): - - if len(sys.argv) != 2: - print("Specify name of .ppdw file") - sys.exit() - - file_name = sys.argv[1] - - with open(file_name, 'rb') as f: - - while(True): - current_bytes = f.read(32) - if current_bytes == '': - print("End of file detected") - break - if len(current_bytes) != 32: - print("Can not read all 32 bytes of next PDW") - break - - current_pdw = PDW.from_bytes(current_bytes) - print(current_pdw) - -if __name__ == '__main__': - main() Index: sts/test_mock.py =================================================================== --- tests/test_mock.py (revision 7c078db26ece7461302f88aef3eba1c85cce4c44) +++ (revision ) @@ -1,5 +1,0 @@ -class TestAmmosReader: - - def test_mock(self): - - assert True