Miscellaneous

Miscellaneous utilities (especially saving, loading, logging).

ecgan.utils.miscellaneous.load_model(model_reference, device)[source]

Load a trained module from disk (file path) or wand reference to device.

Return type

Dict

ecgan.utils.miscellaneous.save_pickle(data, filepath, filename)[source]

Save a generic object to a binary file.

Parameters
  • data (object) -- Object to be saved.

  • filepath (str) -- Saving destination.

  • filename (str) -- Name of file data is saved to.

Return type

None

ecgan.utils.miscellaneous.save_tensor(data, filepath, filename)[source]

Save a torch tensor to a binary file.

Parameters
  • data (Tensor) -- Object to be saved.

  • filepath (str) -- Saving destination.

  • filename (str) -- Name of file data is saved to.

Return type

None

ecgan.utils.miscellaneous.load_pickle(filepath)[source]

Load a binary file to a python object.

Parameters

filepath (str) -- File to be loaded

Return type

Any

ecgan.utils.miscellaneous.load_pickle_numpy(filepath)[source]

Load a binary file to a numpy array.

Parameters

filepath (str) -- File to be loaded

Return type

ndarray

ecgan.utils.miscellaneous.load_yml(filepath)[source]

Load a yml file to memory as dict.

Return type

Dict

Differentiates between local data and W&B data.

Return type

bool

ecgan.utils.miscellaneous.load_wandb_config(run_uri)[source]

Load a config from W&B and convert it to a local one.

Return type

Dict

ecgan.utils.miscellaneous.convert_wandb_config(wandb_config)[source]

Convert a wandb config into a config dict to reconstruct a run.

The wandb includes some additional metadata, especially a value field for each key which has to be removed to fit the original form.

Parameters

wandb_config (str) -- The downloaded wandb config.

Return type

Dict

Returns

The cleaned config parsed into a dict.

ecgan.utils.miscellaneous.update_dicts(original_dict, update_dict, additional_dict)[source]

Update dict and the update dict with an additional dict.

Return type

None

ecgan.utils.miscellaneous.merge_dicts(dict_0, dict_1, duplicates=True)[source]

Recursively merges two dicts into the first one.

Params:

dict_0: The dictionary that is merged into. dict_1: The dictionary that is merged from. duplicates: If False the merging will exit with RunTimeError when merging duplicate keys.

Return type

None

ecgan.utils.miscellaneous.calc_conv_output(in_size, kernel_size, stride)[source]

Calculate the output shape for a network with convolutions.

The calculation is done for every dimension individually. That means if your training input is a 5D Tensor with shape (B x C x H x W x D) you need to call this function 3 times separately for H, W, D in case the dimensions differ. Hence input size is an int, which represents one dimension.

Parameters
  • in_size (int) -- Input size in the CNN.

  • kernel_size (Union[int, List[int]]) -- All kernel sizes in the CNN layers.

  • stride (Union[int, List[int]]) -- All strides in the CNN layers.

Return type

int

Returns

Output size in the given dimension.

ecgan.utils.miscellaneous.calc_pool_output(in_size, kernel_size, stride)[source]

Calculate the output size of a pooling layer.

Return type

int

ecgan.utils.miscellaneous.select_device(gpu_flag)[source]

Select device the model shall be trained on.

Either GPU if GPU is set in config and GPU is available or CPU if GPU is selected but not available or CPU is selected.

Parameters

gpu_flag (bool) -- Flag indicating if GPU shall be used.

Return type

device

Returns

Device for torch (gpu or cpu).

ecgan.utils.miscellaneous.save_epoch(highest_vals, epoch, checkpoint_interval, metrics, final_epoch)[source]

Check if epoch should be saved based on the performance on the validation data.

It will be saved if: 1. auroc/f1 are at its maximum, 2. the model would not be saved due to the checkpoint interval anyways, and 3. the auroc/f1 are above the threshold of 0.7 to avoid excessive saving during first epochs

Parameters
  • highest_vals (Dict) -- Dictionary of the maximum epoch values.

  • epoch (int) -- Current epoch.

  • checkpoint_interval (int) -- Regular checkpoint interval.

  • metrics (List[str]) -- Additional metrics.

  • final_epoch (int) -- Last epoch, saved by default.

Return type

bool

Returns

Flag, indicating if epoch should be saved.

ecgan.utils.miscellaneous.update_highest_metrics(new_vali_metrics, artifacts, highest_metrics, epoch, minimum_metric_improvement=0.005)[source]

Compare validation metrics of current epoch with existing max values.

A value is only saved as a new highest value if the previous highest value is exceeded by at least minimum_metric_improvement. This means that the real highest metric might be higher than the highest metric saved here but it will only be a slight improvement. This avoids too many saved checkpoints which would happen if any relevant metric is improved marginally (see ecgan.utils.miscellaneous.save_epoch().

Parameters
  • new_vali_metrics (Dict) -- Dict containing metric keys and float values for current epoch.

  • artifacts (List[Artifact]) -- List of artifacts that is checked for valid metrics.

  • highest_metrics (Dict[str, Tuple]) -- Dict containing highest metrics. Metric keys and values are Tuples of (epoch, max value).

  • epoch (int) -- Current epoch.

  • minimum_metric_improvement (float) -- Minimum required relative improvement of the metric.

Return type

Dict

Returns

Updated dict with highest metric values.

ecgan.utils.miscellaneous.generate_seed()[source]

Generate a random seed which can later be used as manual seed.

Return type

int

ecgan.utils.miscellaneous.get_num_workers()[source]

Return the number of available CPU cores (minus one).

Return type

int

ecgan.utils.miscellaneous.list_from_tuple_list(metric_tuple_list, position=1)[source]

Retrieve all values at position of a tuple.

Return type

List

ecgan.utils.miscellaneous.nested_list_from_dict_list(metric_list)[source]

Create a nested List from a given Dict.

ecgan.utils.miscellaneous.scale_to_unit_circle(data)[source]

Rescales data to [-1,1] range.

ecgan.utils.miscellaneous.retrieve_model_specification(run_path)[source]

Retrieve model uri, fold and version from existing model path.

Parameters

run_path (str) -- Path of previous run.

Return type

Tuple[str, str, str]

Returns

URI, fold and version of run.

ecgan.utils.miscellaneous.scale_weights(real_label, loss, percentage_normal=None)[source]

Scale the loss of some input based on the training data imbalance.

Assumes a binary classification. The imbalance weighting is calculated per batch. Manual reduction is possible because "reduction='none'" was passed during loss init.

Parameters
  • real_label (Tensor) -- Tensor of binary labels.

  • loss (Tensor) -- Network loss.

  • percentage_normal (Optional[float]) -- Share of labels in the whole dataset which are normal.

Return type

Tensor

Returns

The scaled average loss.

ecgan.utils.miscellaneous.to_numpy(tensor)[source]

Convert a Tensor of type Union[np.ndarray, torch.Tensor] to np.ndarray.

Return type

ndarray

ecgan.utils.miscellaneous.to_torch(tensor)[source]

Convert a Tensor of type Union[np.ndarray, torch.Tensor] to torch.Tensor.

Return type

Tensor