Add files from test project
This commit is contained in:
0
python/hccd/__init__.py
Normal file
0
python/hccd/__init__.py
Normal file
6
python/hccd/__main__.py
Normal file
6
python/hccd/__main__.py
Normal file
@@ -0,0 +1,6 @@
|
||||
def main():
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
117
python/hccd/homotopy_generator.py
Normal file
117
python/hccd/homotopy_generator.py
Normal file
@@ -0,0 +1,117 @@
|
||||
import numpy as np
|
||||
import sympy as sp
|
||||
from typing import List, Callable
|
||||
|
||||
|
||||
class HomotopyGenerator:
|
||||
"""Generates homotopy functions from a binary parity check matrix."""
|
||||
|
||||
def __init__(self, parity_check_matrix: np.ndarray):
|
||||
"""
|
||||
Initialize with a parity check matrix.
|
||||
|
||||
Args:
|
||||
parity_check_matrix: Binary matrix where rows represent parity checks
|
||||
and columns represent variables.
|
||||
"""
|
||||
self.H_matrix = parity_check_matrix
|
||||
self.num_checks, self.num_vars = parity_check_matrix.shape
|
||||
|
||||
# Create symbolic variables
|
||||
self.x_vars = [sp.symbols(f'x{i+1}') for i in range(self.num_vars)]
|
||||
self.t = sp.symbols('t')
|
||||
|
||||
# Generate G, F, and H
|
||||
self.G = self._create_G()
|
||||
self.F = self._create_F()
|
||||
self.H = self._create_H()
|
||||
|
||||
# Convert to callable functions
|
||||
self._H_lambda = self._create_H_lambda()
|
||||
self._DH_lambda = self._create_DH_lambda()
|
||||
|
||||
def _create_G(self) -> List[sp.Expr]:
|
||||
"""Create G polynomial system (the starting system)."""
|
||||
# For each variable xi, add the polynomial [xi]
|
||||
G = []
|
||||
for var in self.x_vars:
|
||||
G.append(var)
|
||||
|
||||
return G
|
||||
|
||||
def _create_F(self) -> List[sp.Expr]:
|
||||
"""Create F polynomial system (the target system)."""
|
||||
F = []
|
||||
|
||||
# Add 1 - xi^2 for each variable
|
||||
for var in self.x_vars:
|
||||
F.append(1 - var**2)
|
||||
|
||||
# Add parity check polynomials: 1 - x1*x2*...*xk for each parity check
|
||||
for row in self.H_matrix:
|
||||
# Create product of variables that participate in this check
|
||||
term = 1
|
||||
for i, bit in enumerate(row):
|
||||
if bit == 1:
|
||||
term *= self.x_vars[i]
|
||||
|
||||
if term != 1: # Only add if there are variables in this check
|
||||
F.append(1 - term)
|
||||
|
||||
return F
|
||||
|
||||
def _create_H(self) -> List[sp.Expr]:
|
||||
"""Create the homotopy H = (1-t)*G + t*F."""
|
||||
H = []
|
||||
|
||||
# Make sure G and F have the same length
|
||||
# Repeat variables from G if needed to match F's length
|
||||
G_extended = self.G.copy()
|
||||
while len(G_extended) < len(self.F):
|
||||
# Cycle through variables to repeat
|
||||
for i in range(min(self.num_vars, len(self.F) - len(G_extended))):
|
||||
G_extended.append(self.x_vars[i % self.num_vars])
|
||||
|
||||
# Create the homotopy
|
||||
for g, f in zip(G_extended, self.F):
|
||||
H.append((1 - self.t) * g + self.t * f)
|
||||
|
||||
return H
|
||||
|
||||
def _create_H_lambda(self) -> Callable:
|
||||
"""Create a lambda function to evaluate H."""
|
||||
all_vars = self.x_vars + [self.t]
|
||||
return sp.lambdify(all_vars, self.H, 'numpy')
|
||||
|
||||
def _create_DH_lambda(self) -> Callable:
|
||||
"""Create a lambda function to evaluate the Jacobian of H."""
|
||||
all_vars = self.x_vars + [self.t]
|
||||
jacobian = sp.Matrix([[sp.diff(expr, var)
|
||||
for var in all_vars] for expr in self.H])
|
||||
return sp.lambdify(all_vars, jacobian, 'numpy')
|
||||
|
||||
def evaluate_H(self, y: np.ndarray) -> np.ndarray:
|
||||
"""
|
||||
Evaluate H at point y.
|
||||
|
||||
Args:
|
||||
y: Array of form [x1, x2, ..., xn, t] where xi are the variables
|
||||
and t is the homotopy parameter.
|
||||
|
||||
Returns:
|
||||
Array containing H evaluated at y.
|
||||
"""
|
||||
return np.array(self._H_lambda(*y))
|
||||
|
||||
def evaluate_DH(self, y: np.ndarray) -> np.ndarray:
|
||||
"""
|
||||
Evaluate the Jacobian of H at point y.
|
||||
|
||||
Args:
|
||||
y: Array of form [x1, x2, ..., xn, t] where xi are the variables
|
||||
and t is the homotopy parameter.
|
||||
|
||||
Returns:
|
||||
Matrix containing the Jacobian of H evaluated at y.
|
||||
"""
|
||||
return np.array(self._DH_lambda(*y), dtype=float)
|
||||
84
python/hccd/path_tracker.py
Normal file
84
python/hccd/path_tracker.py
Normal file
@@ -0,0 +1,84 @@
|
||||
import numpy as np
|
||||
import typing
|
||||
import scipy
|
||||
|
||||
|
||||
def _sign(val):
|
||||
return -1 * (val < 0) + 1 * (val >= 0)
|
||||
|
||||
|
||||
class PathTracker:
|
||||
"""
|
||||
Path trakcer for the homotopy continuation method. Uses a
|
||||
predictor-corrector scheme to trace a path defined by a homotopy.
|
||||
|
||||
References:
|
||||
[1] T. Chen and T.-Y. Li, “Homotopy continuation method for solving
|
||||
systems of nonlinear and polynomial equations,” Communications in
|
||||
Information and Systems, vol. 15, no. 2, pp. 119–307, 2015
|
||||
"""
|
||||
|
||||
def __init__(self, Homotopy, euler_step_size=0.05, euler_max_tries=10, newton_max_iter=5,
|
||||
newton_convergence_threshold=0.001, sigma=1):
|
||||
self.Homotopy = Homotopy
|
||||
self._euler_step_size = euler_step_size
|
||||
self._euler_max_tries = euler_max_tries
|
||||
self._newton_max_iter = newton_max_iter
|
||||
self._newton_convergence_threshold = newton_convergence_threshold
|
||||
self._sigma = sigma
|
||||
|
||||
def step(self, y):
|
||||
"""Perform one predictor-corrector step."""
|
||||
return self.transparent_step(y)[0]
|
||||
|
||||
def transparent_step(self, y) -> typing.Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray,]:
|
||||
"""Perform one predictor-corrector step, returning intermediate results."""
|
||||
for i in range(self._euler_max_tries):
|
||||
step_size = self._euler_step_size / (1 << i)
|
||||
|
||||
y_hat, y_prime = self._perform_euler_predictor_step(y, step_size)
|
||||
y_hat_n = self._perform_newtown_corrector_step(y_hat)
|
||||
|
||||
return y, y_prime, y_hat, y_hat_n
|
||||
|
||||
raise RuntimeError("Newton corrector did not converge")
|
||||
|
||||
def _perform_euler_predictor_step(self, y, step_size) -> typing.Tuple[np.ndarray, np.ndarray]:
|
||||
# Obtain y_prime
|
||||
|
||||
DH = self.Homotopy.evaluate_DH(y)
|
||||
ns = scipy.linalg.null_space(DH)
|
||||
|
||||
y_prime = ns[:, 0] * self._sigma
|
||||
|
||||
# Q, R = np.linalg.qr(np.transpose(DH), mode="complete")
|
||||
# y_prime = Q[:, 2]
|
||||
#
|
||||
# if _sign(np.linalg.det(Q)*np.linalg.det(R[:2, :])) != _sign(self._sigma):
|
||||
# y_prime = -y_prime
|
||||
|
||||
# Perform prediction
|
||||
|
||||
y_hat = y + step_size*y_prime
|
||||
|
||||
return y_hat, y_prime
|
||||
|
||||
def _perform_newtown_corrector_step(self, y) -> np.ndarray:
|
||||
prev_y = y
|
||||
|
||||
for _ in range(self._newton_max_iter):
|
||||
# Perform correction
|
||||
|
||||
DH = self.Homotopy.evaluate_DH(y)
|
||||
DH_pinv = np.linalg.pinv(DH)
|
||||
|
||||
y = y - DH_pinv @ self.Homotopy.evaluate_H(y)
|
||||
|
||||
# Check stopping criterion
|
||||
|
||||
if np.linalg.norm(y - prev_y) < self._newton_convergence_threshold:
|
||||
return y
|
||||
|
||||
prev_y = y
|
||||
|
||||
raise RuntimeError("Newton corrector did not converge")
|
||||
Reference in New Issue
Block a user