WIP: Add post processing with error reconciliation and privacy amplification

.gitignore

@@ -193,4 +193,5 @@ cython_debug/
 
 config.toml
 config*.toml
-config/*
+config/*
+deps/

pyproject.toml

@@ -29,11 +29,12 @@ classifiers = [
 [tool.poetry.dependencies]
 python = ">=3.9,<3.14"
 #qosst-core = "^0.10.0"
-qosst-core = { git = "https://github.com/qosst/qosst-core", branch = "dev" }
+qosst-core = { git = "https://github.com/qosst/qosst-core", branch = "post-processing" }
 #qosst-hal = "^0.10.0"
 qosst-hal = { git = "https://github.com/qosst/qosst-hal", branch = "dev" }
 #qosst-skr = "^0.10.0"
-qosst-skr = { git = "https://github.com/qosst/qosst-skr", branch = "dev" }
+qosst-skr = { git = "https://github.com/qosst/qosst-skr", branch = "post-processing" }
+qosst-pp = { git = "https://github.com/qosst/qosst-pp", branch = "main" }
 matplotlib = [
     { version = "^3.5.1", python = ">=3.9, <3.11" },
     { version = "^3.7.1", python = ">=3.11, <3.14" },
@@ -78,3 +79,4 @@ qosst-bob-optimize = "qosst_bob.optimization.commands:main"
 qosst-bob-transmittance = "qosst_bob.transmittance:main"
 qosst-bob-tools = "qosst_bob.tools.commands:main"
 qosst-bob-offline-dsp = "qosst_bob.dsp.offline_dsp:main"
+qosst-bob-continuous = "qosst_bob.continuous:main"

qosst_bob/bob.py

@@ -20,7 +20,7 @@
 """
 import logging
 import time
-from typing import Any, Optional
+from typing import Any, Optional, List
 import uuid
 import traceback
 from copy import deepcopy
@@ -35,6 +35,8 @@
 from qosst_core.control_protocol.sockets import QOSSTClient
 from qosst_core.notifications import QOSSTNotifier
 from qosst_core.participant import Participant
+from qosst_core.utils import complex_to_real
+from qosst_core.schema.detection import SINGLE_POLARISATION_RF_HETERODYNE
 from qosst_hal.switch import GenericSwitch
 from qosst_hal.laser import GenericLaser
 from qosst_hal.adc import GenericADC
@@ -44,6 +46,9 @@
 )
 from qosst_hal.powermeter import GenericPowerMeter
 
+from qosst_pp.reconciliation import reconcile_bob
+from qosst_pp.privacy_amplification import privacy_amplification_bob
+
 from qosst_bob.dsp import dsp_bob, special_dsp
 from qosst_bob.dsp.dsp import find_global_angle
 from qosst_bob.data import ElectronicNoise, ElectronicShotNoise
@@ -93,7 +98,16 @@ class Bob:
     excess_noise_bob: float  #: Excess noise estimated at Bob.
     vel: float  #: Normalised electronic noise.
     skr: float  #: Secret key rate.
+    secret_key_ratio: float  #: Secret key ration in bits/symbol.
     photon_number: float  #: Mean photon number at Alice's side.
+    signal_to_noise_ratio: float  #: Signal to noise ratio of the quantum symbols.
+
+    raw_key_material: Optional[
+        np.ndarray
+    ]  #: Raw key material before error reconciliation, after parameter estimation.
+
+    reconciled_key: Optional[List[int]]  #: Key after the reconciliation step.
+    final_key: Optional[List[int]]  #: Final key after privacy amplification step.
 
     adc: Optional[GenericADC]  #: ADC device for Bob.
     switch: Optional[GenericSwitch]  #: Switch device for Bob.
@@ -123,9 +137,27 @@ def __init__(self, config_path: str, enable_laser: bool = True):
         self.is_connected = False
         self.config = None
 
+        self.enable_laser = enable_laser
+
+        self.laser = None
+        self.switch = None
+        self.adc = None
+        self.polarisation_controller = None
+        self.powermeter = None
+
+        logger.info("Initializing Bob...")
+
         self.electronic_noise = None
         self.electronic_shot_noise = None
 
+        self._reset()
+
+        self.load_configuration()
+        self._init_socket()
+        self._init_notifier()
+
+    def _reset(self):
+
         self.electronic_symbols = None
         self.electronic_shot_symbols = None
 
@@ -138,6 +170,10 @@ def __init__(self, config_path: str, enable_laser: bool = True):
         self.excess_noise_bob = 0
         self.vel = 0
         self.skr = 0
+        self.secret_key_ratio = 0
+
+        self.reconciled_key = None
+        self.final_key = None
 
         self.adc_data = None
         self.signal_data = None
@@ -148,20 +184,8 @@ def __init__(self, config_path: str, enable_laser: bool = True):
         self.quantum_data_phase_noisy = None
         self.quantum_symbols = None
 
-        self.enable_laser = enable_laser
-
-        self.laser = None
-        self.switch = None
-        self.adc = None
-
-        logger.info("Initializing Bob...")
-
         self.frame_uuid = None
 
-        self.load_configuration()
-        self._init_socket()
-        self._init_notifier()
-
     def load_configuration(self) -> None:
         """
         Load or reload the configuration.
@@ -555,17 +579,17 @@ def parameters_estimation(self) -> bool:
 
         logger.info("Computing secret key rate")
         try:
-            self.skr = (
-                self.config.frame.quantum.symbol_rate
-                * self.config.bob.parameters_estimation.skr_calculator.skr(
+            self.secret_key_ratio = (
+                self.config.bob.parameters_estimation.skr_calculator.skr(
                     Va=2 * self.photon_number,
                     T=self.transmittance / self.config.bob.eta,
                     xi=self.excess_noise_bob / self.transmittance,
                     eta=self.config.bob.eta,
                     Vel=self.vel,
                     beta=0.95,
                 )
-            )
+            )  # Not final as computed with beta=0.95
+            self.skr = self.config.frame.quantum.symbol_rate * self.secret_key_ratio
         except ValueError:
             self.skr = -1
 
@@ -584,33 +608,92 @@ def parameters_estimation(self) -> bool:
             },
         )
 
+        if self.config.bob.schema == SINGLE_POLARISATION_RF_HETERODYNE:
+            self.signal_to_noise_ratio = (
+                self.transmittance
+                * 2
+                * self.photon_number
+                / (2 + 2 * self.vel + self.excess_noise_bob)
+            )
+
         if code == QOSSTCodes.PE_APPROVED:
+            logger.info(
+                "Removing symbols used for parameters estimation from raw key material"
+            )
+            mask = np.ones(
+                len(self.quantum_symbols), dtype=bool
+            )  # Create an array of True, same length of quantum symbols
+            mask[self.indices] = False  # Set False where to remove
+            self.raw_key_material = np.copy(self.quantum_symbols[mask])
+            logger.info("%i raw key symbols", len(self.raw_key_material))
             return True
         return False
 
     def error_correction(self) -> bool:
         """
         Apply error correction on the data.
 
-        Raises:
-            NotImplementedError: This function is not yet implemented.
-
         Returns:
             bool: True if the operation was successful, False otherwise.
         """
-        raise NotImplementedError("Error correction has not yet been implemented.")
+        # Perform reconciliation
+        # Data should be normalised
+        shot_variance = np.var(self.electronic_shot_symbols) - np.var(
+            self.electronic_symbols
+        )
+
+        # Also the SNR needs to be given in dB
+        snr_db = 10 * np.log10(self.signal_to_noise_ratio)
+        self.reconciled_key = reconcile_bob(
+            self.socket,
+            complex_to_real(self.raw_key_material / np.sqrt(shot_variance)),
+            self.config.post_processing.reconciliation.beta,
+            snr_db,
+            self.config.post_processing.reconciliation.dimension,
+        )
+        return self.reconciled_key is not None
 
     def privacy_amplification(self) -> bool:
         """
         Apply privacy amplification on the data.
 
-        Raises:
-            NotImplementedError: This function is not yet implemented.
-
         Returns:
             bool: True if the operation was successful, False otherwise.
         """
-        raise NotImplementedError("Privacy amplification has not yet been implemented.")
+        if len(self.reconciled_key):
+            self.final_key = privacy_amplification_bob(
+                self.socket,
+                self.reconciled_key,
+                self.secret_key_ratio,
+                self.config.post_processing.privacy_amplification.extractor,
+            )
+            return self.final_key is not None
+        logger.error(
+            "Reconciled key has length 0. Cannot perform privacy amplification."
+        )
+        return False
+
+    def end_frame(self) -> bool:
+        """
+        Push key to KMS, send end of frame to Alice, and reset attributes.
+        """
+        ret_code = True
+        if self.final_key:
+            if self.config.pushkey:
+                logger.info("Pushing key to KMS")
+                self.config.pushkey.interface(
+                    self.frame_uuid, self.final_key, **self.config.pushkey.kwargs
+                )
+            else:
+                logger.warning("No interface defined for KMS. Discarding key.")
+            code, _ = self.socket.request(QOSSTCodes.FRAME_ENDED)
+            if code != QOSSTCodes.FRAME_ENDED_ACK:
+                logger.error("Alice responded FRAME_ENDED with %s", str(code))
+                ret_code = False
+
+        logger.info("Resetting values")
+        self._reset()
+        return ret_code
 
     def _start_acquisition(self):
         """
@@ -705,8 +788,8 @@ def _do_dsp(self):
 
         logger.info("Applying DSP on elec and elec+shot noise data")
 
-        params.elec_noise_ratio = config.bob.dsp.elec_noise_ratio
-        params.elec_shot_noise_ratio = config.bob.dsp.elec_shot_noise_ratio
+        params.elec_noise_ratio = self.config.bob.dsp.elec_noise_ratio
+        params.elec_shot_noise_ratio = self.config.bob.dsp.elec_shot_noise_ratio
         self.electronic_symbols, self.electronic_shot_symbols = special_dsp(
             self.electronic_noise.data, self.electronic_shot_noise.data, params
         )