Instructions:
The project focuses on the use of FMCW radar.
It is used for data collection, point detection, tracking and other functions.
The project is broken down into several modules.
The aim of this division is to decouple as much as possible the functionalities present in each module, and make the development of new functionalities more flexible and rapid.

Some information will be presented to you in json format:
- module: the name of the module
- module_structure: the structure of the module (the files making up the module and their hierarchy)
- module_files_already_generated_doc: the documentation of the module's files if already generated
- other_modules_doc: the documentation of the other modules on which the module depends (possibly none at all)
- gen_doc_of_file: the file you will be given to generate the documentation of. If no file is given, the documentation of the entire module must be generated.
Your goal is to create a markdown documentation of the file you will be given, or the entire module if no file is given.
This documentation is intended to guide a developer who is new to the project.
You can therefore add whatever you feel is relevant (data flow, functionalities, etc.).


Informations:
{"module": "radar_toolbox", "module_structure": "radar_toolbox/utils.py;radar_toolbox\\capture_session/main.py;radar_toolbox\\conf/capture_conf.json;radar_toolbox\\conf/profile_vs1642_exp.cfg;radar_toolbox\\conf/raw_bin_to_numpy.json;radar_toolbox\\raw_data/adc.py;radar_toolbox\\raw_data/raw_bin_to_numpy.py;radar_toolbox\\raw_data/record_raw_data.py;radar_toolbox\\TI_data/exceptions.py;radar_toolbox\\TI_data/record_ti_data.py;radar_toolbox\\TI_data/sensor_com.py;", "module_files_already_generated_doc": [{"file": "radar_toolbox\\fmcw_utils.md", "doc": "# radar_toolbox/fmcw_utils.py Documentation\n\n## Overview\nfmcw_utils.py is a part of the radar_toolbox module and provides utility functions, classes, and constants specifically for frequency modulated continuous wave (FMCW) radar systems. This file is crucial for managing radar models, loading and processing radar data, and preparing captured raw data for analysis.\n\n## Classes\n\n### RadarModel (Enum)\nAn enumeration for the supported radar models.\n\n- IWR1642\n- IWR6843AOP\n- IWRL6432\n- get_model_from_str(string): Static method that takes a string identifier and returns the corresponding RadarModel enum. Raises NotImplementedError if the radar model is not implemented.\n\n### PointsAttributes (Enum)\nEnumeration of possible point cloud attributes\n\n- RANGE\n- AZIMUTH\n- ELEVATION\n- VELOCITY\n- points_attributes_abs(points, attribute_idx): Modifies the given points array by taking the absolute value of the specified attribute.\n\n### AntennaDataSaver\nA helper class for storing calculated FFT data.\n\n- __init__(save_abs_1D_FFTs, save_abs_2D_FFTs): Constructor specifying whether to save FFT data.\n- add_abs_1D_FFTs(FFTs_1D): Stores one-dimensional FFT results.\n- add_abs_2D_FFTs(FFTs_2D): Stores two-dimensional FFT results.\n\n### RadarDataSaver\nDesigned to store common detection matrices for further analysis.\n\n- __init__(save_common_detection_matrices): Constructor specifying whether to save common detection matrices.\n- add_common_detection_matrices(common_detection_matrices): Adds a new common detection matrix.\n\n## Functions\n\n### get_phase_compensation_for_radar_model(model)\nReturns the phase compensation array for a specific radar model. Raises NotImplementedError if the radar model is not handled.\n\n### load_IQ_file(path, is_complex, nb_samples, nb_chirps)\nLoads interleaved I/Q data from a file and returns a numpy array reshaped for chirps and samples. Accommodates complex data if is_complex is True.\n\n### load_IQ_data(path_prefix, is_complex, nb_antennas, nb_samples, nb_chirps)\nLoads interleaved I/Q data across multiple antennas and returns a combined numpy array for further processing.\n\n### raw_capture_bin_preparation(capture_bin_file_path, cfg_file_path, radar, output_bin_file_path, verbose)\nProcesses a binary capture file and prepares it in a structured format suitable for analysis, including handling of missing packets.\n\n### prepared_bin_to_numpy(bin_file_path, cfg_file_path, radar, nb_lanes)\nConverts a prepared binary radar data file to numpy format, taking into account missing packets and ensures only complete frames are returned.\n\n### coordinates_from_range_and_angles(azimuths, elevations, ranges)\nCalculates Cartesian coordinates from radar data specified by ranges, azimuths, and elevation angles.\n\n### read_cfg_file(path, radar_model, verbose)\nReads a .cfg file and returns a configuration dictionary with key parameters for the radar operation. Handles different radar models.\n\n### describe_config(config)\nPrints the configuration dictionary in a readable format for better understanding of the radar setup.\n\n### safe_memmap(filename, mode, dtype, shape)\nCreates a memory-mapped file for large arrays ensuring compatibility with the numpy .npy format.\n\n### Frame\nA class managing the construction of radar frames from raw data, handling cross-frame packet scenarios.\n\n- __init__( ...): Initializes the Frame object, which holds data for a single radar frame.\n- is_complete(): Returns whether the frame is complete.\n- add_values(data_to_append): Appends new data to the current frame.\n- get_frame_values(): Retrieves the final frame values.\n- get_next_frame(): Builds and returns a new frame with the remaining data if the current frame is complete.\n- get_frame_index(): Returns the index of the current frame.\n- get_nb_values_in_frame(): Returns the number of values in a frame.\n- order_frame(flat_unordered_frame): Orders a flat unordered frame into a structured numpy array.\n\n### AntennaConfig\nRepresents the configuration of an FMCW radar antenna, with attributes like start frequency, sampling rate, etc.\n\n- __init__( ...): Initializes the AntennaConfig instance with radar parameters.\n- get_antenna_parameters_from_config_file(config): Static method that creates an instance of AntennaConfig from radar configuration parameters.\n\n## Data Flow\nThe individual functions and classes deal with various stages of data acquisition, preprocessing, and transformation into formats that can be utilized for subsequent analysis. It goes from reading radar configurations, capturing raw binary data, preprocessing the data, extracting I/Q samples, organizing data into frames, and finally converting into coordinates relevant for point cloud generation.\n\n## Usage\nA developer would typically use these utilities to interface with FMCW radar hardware, manage radar configurations, process captured radar signals, and transform them into more analytical forms for further tasks such as object detection or tracking.\n\nTo work with fmcw_utils.py, familiarity with numpy and an understanding of FMCW radar processing are recommended.\n\n\u00e2\u20ac\u201d\nThis documentation is intended to serve as a guide for developers interested in the functionalities of the fmcw_utils.py file within the radar_toolbox module."}], "other_modules_doc": null, "gen_doc_of_file": {"path": "radar_toolbox\\utils.py", "content": "import argparse\nimport json\nimport os\nimport re\nfrom copy import deepcopy\nfrom dataclasses import dataclass\n\n\n@dataclass\nclass GridSize:\n    x: int\n    y: int\n    z: int\n\n    def get_shape(self):\n        return (self.x, self.y, self.z)\n\n\n@dataclass\nclass CaptureBoundaries:\n    x_min: float\n    x_max: float\n    y_min: float\n    y_max: float\n    z_min: float\n    z_max: float\n\n\ndef chunks(lst, n):\n    \"\"\"Yield successive n-sized chunks from lst.\"\"\"\n    for i in range(0, len(lst), n):\n        yield lst[i:i + n]\n\n\ndef configure_argument_parser_for_conf_file() -> argparse.Namespace:\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--conf\", help=\"Path to the conf file\", type=str, required=True)\n    args = parser.parse_args()\n    return args\n\n\ndef dict_deepmerge(x, y):\n    \"\"\"\n    Return a new dict that is a deep merge of x and y.\n    If two keys are present in both dict :\n        - if value is a dict, then we try to merge them\n        - else, we copy value from x in priority.\n\n    example:\n    >>> x = {\"a\": 1, \"b\": {\"c\": 2}}\n    >>> y = {\"a\": 11, \"b\": {\"d\": 3}}\n    >>> print(dict_deepmerge(x, y))\n\n    {'b': {'c': 2, 'd': 3}, 'a': 1}\n    \"\"\"\n    z = {}\n    if not isinstance(x, dict) or not isinstance(y, dict):\n        return x\n    overlapping_keys = x.keys() & y.keys()\n    for key in overlapping_keys:\n        z[key] = dict_deepmerge(x[key], y[key])\n    for key in x.keys() - overlapping_keys:\n        z[key] = deepcopy(x[key])\n    for key in y.keys() - overlapping_keys:\n        z[key] = deepcopy(y[key])\n    return z\n\n\ndef load_conf(config_file: str) -> dict:\n    from jsonschema import Draft7Validator\n    \"\"\"\n    Loads a config file.\n    :param config_file: the path of the config file.\n    :return: The configuration dictionary.\n    \"\"\"\n    with open(config_file, 'r') as file_content:\n        config = json.load(file_content)\n\n    # If configuration file contains a \"load_conf\" key, it will load this additional configuration file\n    # (using the value) and update the current conf file with it.\n    if \"load_conf\" in config:\n        path = config[\"load_conf\"]\n        if path.startswith(\"$:\"):\n            path = os.path.join(os.path.dirname(config_file), path[2:])\n        additional_conf = load_conf(path)\n        additional_conf.update(config)  # Parameters of parent conf has priority\n        config = dict_deepmerge(config, additional_conf)\n\n    if \"schema\" in config:\n        path = config[\"schema\"]\n        if path.startswith(\"$:\"):\n            path = os.path.join(os.path.dirname(config_file), path[2:])\n        schema = load_conf(path)\n        validator = Draft7Validator(schema)\n        validator.validate(config)\n    return config\n\n\ndef get_class(cls: str):\n    parts = cls.split('.')\n    module = \".\".join(parts[:-1])\n    m = __import__(module)\n    for comp in parts[1:]:\n        m = getattr(m, comp)\n    return m\n\n\ndef load_data_from_conf(config_dataloader):\n    dataloader_cls = get_class(config_dataloader['class'])\n    dataloader = dataloader_cls(**config_dataloader['parameters'])\n    data = dataloader.load_data()\n    return data\n\n\ndef create_pytorch_estimator(entry_point: str, config_nn: dict,\n                             instance_type: str, bucket: str, arn_role: str,\n                             max_run=3 * 60 * 60 * 24,\n                             max_wait=4 * 60 * 60 * 24,\n                             use_spot=True,\n                             volume_size=30,\n                             instance_count=1):\n    from sagemaker.pytorch import PyTorch\n    from sagemaker.debugger import TensorBoardOutputConfig\n    tensorboard_output_config = TensorBoardOutputConfig(\"s3://\" + bucket + \"/tensorboard-v2\")\n    estimator = PyTorch(\n        entry_point=entry_point,\n        source_dir='src',\n        role=arn_role,\n        framework_version='2.0.0',\n        py_version='py310',\n        instance_count=instance_count,\n        instance_type=instance_type,\n        tensorboard_output_config=tensorboard_output_config,\n        use_spot_instances=use_spot,\n        max_run=max_run,\n        # max_wait=max_wait,\n        hyperparameters=config_nn,\n        volume_size=volume_size,\n        checkpoint_s3_uri=f\"s3://{bucket}/checkpoints\",\n        checkpoint_local_path=\"/opt/ml/checkpoints\"\n    )\n\n    return estimator\n\n\ndef create_custom_estimator(image_uri: str, entry_point: str, config_nn: dict,\n                            instance_type: str, bucket: str, arn_role: str,\n                            max_run=3 * 60 * 60 * 24,\n                            max_wait=4 * 60 * 60 * 24,\n                            use_spot=True,\n                            volume_size=30):\n    from sagemaker.pytorch import PyTorch\n    from sagemaker.debugger import TensorBoardOutputConfig\n    tensorboard_output_config = TensorBoardOutputConfig(\"s3://\" + bucket + \"/tensorboard-v2\")\n    estimator = PyTorch(\n        entry_point=entry_point,\n        source_dir='src',\n        role=arn_role,\n        image_uri=image_uri,\n        instance_count=1,\n        instance_type=instance_type,\n        tensorboard_output_config=tensorboard_output_config,\n        use_spot_instances=use_spot,\n        max_run=max_run,\n        # max_wait=max_wait,\n        hyperparameters=config_nn,\n        volume_size=volume_size,\n        checkpoint_s3_uri=f\"s3://{bucket}/checkpoints\",\n        checkpoint_local_path=\"/opt/ml/checkpoints\"\n    )\n\n    return estimator\n\n\ndef get_image_uri(account_id, region, ecr_repository_name, version):\n    return \"{}.dkr.ecr.{}.amazonaws.com/{}:{}\".format(account_id, region, ecr_repository_name, version)\n\n\ndef parse_extra_args(args: dict, extra_args):\n    if not extra_args:\n        return\n\n    it = iter(extra_args)\n    pattern = re.compile(r'(\\w*(?<!\\d))(\\d+)$')\n    parsed_extra = list()\n\n    for arg_name in it:\n        value = next(it)\n        match = pattern.search(arg_name)\n\n        key, pos = match.group(1, 2)\n        parsed_extra.append((key, pos, value))\n\n    parsed_extra = sorted(parsed_extra, key=lambda x: (x[0], x[1]))\n\n    for key, _, value in parsed_extra:\n        args.setdefault(key, []).append(value)\n\n\ndef csv_capture_player(config):\n    from pandas import read_csv\n    \"\"\"\n    Generator that reads a csv file of a capture, extracts step's points and yield them as a numpy array.\n    :returns Array of points of the current step of shape: (nb_points_in_step, 3).\n        The last dimension is of size 3: (x, y, z) in this order.\n    \"\"\"\n    frame_grouping_size = config.get(\"frame_grouping_size\", 1)  # How many frames do we want per step\n    data = read_csv(config[\"capture_file\"], ';')\n    if \"t\" not in data.columns.values:\n        data['t'] = data['frame'] // frame_grouping_size  # Creating time column\n    else:\n        data['t'] = data['t'] // frame_grouping_size  # Creating time column\n    data.reset_index(inplace=True, drop=True)\n    data_cluster = data[['x', 'y', 'z', 't']]  # Selecting and reordering columns\n    nb_time_steps = max(data_cluster[\"t\"])\n    for t in range(nb_time_steps):\n        # We select the sample according to current time step\n        data_t = data_cluster.loc[data_cluster['t'] == t].drop(columns=['t'])\n        yield data_t.to_numpy()\n\n\ndef get_callbacks_from_params(callbacks_params):\n    from pytorch_lightning.callbacks import ModelCheckpoint, EarlyStopping, ModelPruning\n    callbacks = []\n    if \"early_stopping\" in callbacks_params:\n        early_stopping_params = callbacks_params[\"early_stopping\"]\n        early_stopping = EarlyStopping(\n            monitor=early_stopping_params[\"monitor\"],\n            mode=early_stopping_params[\"mode\"],\n            patience=early_stopping_params[\"patience\"],\n            check_finite=False,\n            verbose=1)\n        callbacks.append(early_stopping)\n    if \"checkpoint\" in callbacks_params:\n        checkpoint_params = callbacks_params[\"checkpoint\"]\n        checkpoint = ModelCheckpoint(\n            dirpath=checkpoint_params[\"dirpath\"],\n            filename=checkpoint_params[\"filename\"],\n            save_top_k=1,\n            verbose=True,\n            monitor=checkpoint_params[\"monitor\"],\n            mode=checkpoint_params[\"mode\"])\n        callbacks.append(checkpoint)\n    if \"pruning\" in callbacks_params:\n        pruning_params = callbacks_params[\"pruning\"]\n        start_epoch, end_epoch, pruning_ratio = pruning_params[\"start_epoch\"], pruning_params[\"end_epoch\"], \\\n                                                pruning_params[\n                                                    \"pruning_ratio\"]\n\n        def compute_amount(epoch):\n            if start_epoch <= epoch < end_epoch:\n                return pruning_ratio\n            return 0\n\n        pruning_callback = ModelPruning(\n            pruning_fn=pruning_params[\"pruning_fn\"],\n            amount=compute_amount,\n            verbose=pruning_params[\"verbose\"],\n            parameter_names=pruning_params[\"parameter_names\"])\n        callbacks.append(pruning_callback)\n    return callbacks\n\n\ndef get_reader_from_config(config):\n    from radar_toolbox.fmcw_utils import RadarModel, read_cfg_file\n    from radar_toolbox.TI_data.sensor_com import Reader68XX, Reader64XX, Reader16XX\n    radar_model = RadarModel.get_model_from_str(config[\"model\"])\n\n    if radar_model == RadarModel.IWR6843AOP:\n        reader = Reader68XX()\n    elif radar_model == RadarModel.IWR1642:\n        from radar_toolbox.fmcw_utils import AntennaConfig\n        radar_config = read_cfg_file(path=config[\"cfg_file_path\"], radar_model=radar_model)\n        antenna_config = AntennaConfig.get_antenna_parameters_from_config_file(radar_config)\n        reader = Reader16XX(antenna_config=antenna_config)\n    elif radar_model == RadarModel.IWRL6432:\n        reader = Reader64XX()\n    else:\n        raise Exception(f\"{radar_model} model not handled.\")\n    return reader\n\n\ndef coord_rotation(first_axis_values, second_axis_values, theta):\n    \"\"\"\n    Rotates given coordinates in the first-axis/second-axis plane anti-clockwise through an angle theta about the origin\n    \"\"\"\n    import numpy as np\n    new_values_first_axis = np.cos(theta) * first_axis_values + np.sin(theta) * second_axis_values\n    new_values_second_axis = - np.sin(theta) * first_axis_values + np.cos(theta) * second_axis_values\n    return new_values_first_axis, new_values_second_axis\n\n\ndef points_inclination_correction(points, horizontal_inclination, vertical_inclination):\n    \"\"\"\n    :param points: xyz coordinates\n    :param horizontal_inclination: inclination in the widht-depth plane, in radians\n    :param vertical_inclination: inclination in the height-depth plane, in radians\n    :return:\n    \"\"\"\n    if horizontal_inclination != None:\n        y_values = points[:, 1]\n        z_values = points[:, 2]\n        points[:, 1], points[:, 2] = coord_rotation(y_values, z_values, horizontal_inclination)\n    if vertical_inclination != None:\n        x_values = points[:, 0]\n        y_values = points[:, 1]\n        points[:, 0], points[:, 1] = coord_rotation(x_values, y_values, vertical_inclination)\n    return points\n"}}