Module adannealing.plotting
Expand source code
from typing import Union
import pandas as pd
import numpy as np
import ast
import logging
logger = logging.getLogger(__name__)
class Sampler:
"""A class used by Annealer to keep track of its progress."""
@staticmethod
def _manage_data(data: pd.DataFrame) -> pd.DataFrame:
def _handle_weights(weights: str) -> list:
while " " in weights:
weights = weights.replace(" ", " ")
return ast.literal_eval(weights.replace(" ", ",").replace("[,", "["))
expected_columns = ["weights", "iteration", "acc_ratio", "accepted", "loss", "parameter"]
for col in expected_columns:
if col not in data:
raise IndexError(f"Missing columns '{col}'")
if data.empty:
return data
if type(data["weights"].iloc[0]) == str:
data.loc[:, "weights"] = data.loc[:, "weights"].apply(_handle_weights)
return data
def __init__(self, data: pd.DataFrame = None):
if data is not None:
self._data = Sampler._manage_data(data)
self.points = None
else:
self._data = pd.DataFrame()
self.points = []
def append(self, value):
if self.points is None:
raise ValueError("Sampler was initialised with an outside history : can not add more points.")
self.points.append(value)
def clean(self):
self._data = pd.DataFrame()
self.points = []
def __len__(self):
if self.points is None:
return len(self._data.index)
else:
return len(self.points)
def _process(self):
if self.points is None:
raise ValueError("Sampler was initialised with an outside history : nothing to process.")
self._data = pd.DataFrame(
[[p.weights, p.iteration, p.acc_ratio, p.accepted, p.loss, p.parameter] for p in self.points],
columns=["weights", "iteration", "acc_ratio", "accepted", "loss", "parameter"],
)
@property
def weights(self):
if self._data.empty or len(self._data.index) != len(self):
self._process()
return pd.DataFrame(index=self._data.index, data=np.array([w for w in self._data.loc[:, "weights"].values]))
@property
def acc_ratios(self):
if self._data.empty or len(self._data.index) != len(self):
self._process()
return self._data.loc[:, "acc_ratio"]
@property
def accepted(self):
if self._data.empty or len(self._data.index) != len(self):
self._process()
return self._data.loc[:, "accepted"]
@property
def losses(self):
if self._data.empty or len(self._data.index) != len(self):
self._process()
return self._data.loc[:, "loss"]
@property
def parameters(self):
if self._data.empty or len(self._data.index) != len(self):
self._process()
return self._data.loc[:, "parameter"]
@property
def iterations(self):
if self._data.empty or len(self._data.index) != len(self):
self._process()
return self._data.index
@property
def data(self):
if self._data.empty or len(self._data.index) != len(self):
self._process()
return self._data
class SamplePoint:
"""A class used by Annealer to keep track of its progress."""
def __init__(
self,
weights,
iteration,
acc_ratio,
accepted,
loss,
temp=None,
demon_loss=None,
sampler: Union[None, Sampler] = None,
):
if sampler is not None and not isinstance(sampler, Sampler):
raise TypeError(f"Sampler must be of type 'Sampler', got {type(sampler)}")
self.weights = weights
self.iteration = iteration
self.acc_ratio = acc_ratio
self.accepted = accepted
self.loss = loss
self.temp = temp
self.demon_loss = demon_loss
if self.demon_loss is None and self.temp is None:
raise ValueError("One and only one of temp and demon_loss must be specified")
if self.demon_loss is not None and self.temp is not None:
raise ValueError("One and only one of temp and demon_loss must be specified")
if sampler is not None:
sampler.append(self)
@property
def parameter(self):
if self.demon_loss is None and self.temp is None:
raise ValueError("One and only one of temp and demon_loss must be specified")
if self.demon_loss is not None and self.temp is not None:
raise ValueError("One and only one of temp and demon_loss must be specified")
if self.temp is not None:
return self.temp
else:
return self.demon_loss
Classes
class SamplePoint (weights, iteration, acc_ratio, accepted, loss, temp=None, demon_loss=None, sampler: Optional[None] = None)
-
A class used by Annealer to keep track of its progress.
Expand source code
class SamplePoint: """A class used by Annealer to keep track of its progress.""" def __init__( self, weights, iteration, acc_ratio, accepted, loss, temp=None, demon_loss=None, sampler: Union[None, Sampler] = None, ): if sampler is not None and not isinstance(sampler, Sampler): raise TypeError(f"Sampler must be of type 'Sampler', got {type(sampler)}") self.weights = weights self.iteration = iteration self.acc_ratio = acc_ratio self.accepted = accepted self.loss = loss self.temp = temp self.demon_loss = demon_loss if self.demon_loss is None and self.temp is None: raise ValueError("One and only one of temp and demon_loss must be specified") if self.demon_loss is not None and self.temp is not None: raise ValueError("One and only one of temp and demon_loss must be specified") if sampler is not None: sampler.append(self) @property def parameter(self): if self.demon_loss is None and self.temp is None: raise ValueError("One and only one of temp and demon_loss must be specified") if self.demon_loss is not None and self.temp is not None: raise ValueError("One and only one of temp and demon_loss must be specified") if self.temp is not None: return self.temp else: return self.demon_loss
Instance variables
var parameter
-
Expand source code
@property def parameter(self): if self.demon_loss is None and self.temp is None: raise ValueError("One and only one of temp and demon_loss must be specified") if self.demon_loss is not None and self.temp is not None: raise ValueError("One and only one of temp and demon_loss must be specified") if self.temp is not None: return self.temp else: return self.demon_loss
class Sampler (data: pandas.core.frame.DataFrame = None)
-
A class used by Annealer to keep track of its progress.
Expand source code
class Sampler: """A class used by Annealer to keep track of its progress.""" @staticmethod def _manage_data(data: pd.DataFrame) -> pd.DataFrame: def _handle_weights(weights: str) -> list: while " " in weights: weights = weights.replace(" ", " ") return ast.literal_eval(weights.replace(" ", ",").replace("[,", "[")) expected_columns = ["weights", "iteration", "acc_ratio", "accepted", "loss", "parameter"] for col in expected_columns: if col not in data: raise IndexError(f"Missing columns '{col}'") if data.empty: return data if type(data["weights"].iloc[0]) == str: data.loc[:, "weights"] = data.loc[:, "weights"].apply(_handle_weights) return data def __init__(self, data: pd.DataFrame = None): if data is not None: self._data = Sampler._manage_data(data) self.points = None else: self._data = pd.DataFrame() self.points = [] def append(self, value): if self.points is None: raise ValueError("Sampler was initialised with an outside history : can not add more points.") self.points.append(value) def clean(self): self._data = pd.DataFrame() self.points = [] def __len__(self): if self.points is None: return len(self._data.index) else: return len(self.points) def _process(self): if self.points is None: raise ValueError("Sampler was initialised with an outside history : nothing to process.") self._data = pd.DataFrame( [[p.weights, p.iteration, p.acc_ratio, p.accepted, p.loss, p.parameter] for p in self.points], columns=["weights", "iteration", "acc_ratio", "accepted", "loss", "parameter"], ) @property def weights(self): if self._data.empty or len(self._data.index) != len(self): self._process() return pd.DataFrame(index=self._data.index, data=np.array([w for w in self._data.loc[:, "weights"].values])) @property def acc_ratios(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.loc[:, "acc_ratio"] @property def accepted(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.loc[:, "accepted"] @property def losses(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.loc[:, "loss"] @property def parameters(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.loc[:, "parameter"] @property def iterations(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.index @property def data(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data
Instance variables
var acc_ratios
-
Expand source code
@property def acc_ratios(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.loc[:, "acc_ratio"]
var accepted
-
Expand source code
@property def accepted(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.loc[:, "accepted"]
var data
-
Expand source code
@property def data(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data
var iterations
-
Expand source code
@property def iterations(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.index
var losses
-
Expand source code
@property def losses(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.loc[:, "loss"]
var parameters
-
Expand source code
@property def parameters(self): if self._data.empty or len(self._data.index) != len(self): self._process() return self._data.loc[:, "parameter"]
var weights
-
Expand source code
@property def weights(self): if self._data.empty or len(self._data.index) != len(self): self._process() return pd.DataFrame(index=self._data.index, data=np.array([w for w in self._data.loc[:, "weights"].values]))
Methods
def append(self, value)
-
Expand source code
def append(self, value): if self.points is None: raise ValueError("Sampler was initialised with an outside history : can not add more points.") self.points.append(value)
def clean(self)
-
Expand source code
def clean(self): self._data = pd.DataFrame() self.points = []