diff --git a/src/py/calix/scripts/calix_generate_default_calibration.py b/src/py/calix/scripts/calix_generate_default_calibration.py new file mode 100644 index 0000000000000000000000000000000000000000..0d6325df0cbb2f78515352fed52feeb55a23e6b3 --- /dev/null +++ b/src/py/calix/scripts/calix_generate_default_calibration.py @@ -0,0 +1,232 @@ +#!/usr/bin/env python3 +""" +Execute all available default calibration and saves them in all available data +formats. If result files already exist, they are overwritten. +""" + +import argparse +import pickle +import gzip +from abc import ABCMeta, abstractmethod +from pathlib import Path +from typing import Union, List + +from dlens_vx_v1.hxcomm import ConnectionHandle, ManagedConnection +from dlens_vx_v1.sta import PlaybackProgramBuilderDumper, ExperimentInit, \ + run, to_json, to_portablebinary + +import calix.hagen +from calix.hagen import HagenCalibrationResult +import calix.spiking +from calix.spiking import SpikingCalibrationResult + +# TODO @JW: There should be a common base for these (Hagen, Spiking, ...) +CalibrationResult = Union[HagenCalibrationResult, SpikingCalibrationResult] + + +class CalibrationRecorder(metaclass=ABCMeta): + """ + Recorder for various calibration results, to be implemented for HAGEN, + Spiking etc. + """ + + @property + @abstractmethod + def calibration_type(self) -> str: + """ + Identifier for the acquired calibration data. + """ + raise NotImplementedError + + @abstractmethod + def generate_calib(self, + connection: ConnectionHandle) -> CalibrationResult: + """ + Execute calibration routines and create a result object. + :param connection: Connection used for acquiring calibration data. + :return: Calibration result data + """ + raise NotImplementedError + + +class CalibrationDumper(metaclass=ABCMeta): + """ + Dumper for calibration data into various formats, e.g. calix internal, + write instructions etc. + """ + + @property + @abstractmethod + def format_name(self) -> str: + """ + Identifier for the dumped data format. + """ + raise NotImplementedError + + @property + @abstractmethod + def format_extension(self) -> str: + """ + File extension for the dumped data format, including a leading '.'. + """ + raise NotImplementedError + + @abstractmethod + def dump_calibration(self, calibration_result: CalibrationResult, + target_file: Path): + """ + Read a calibration result and serialize it to a given file. + :param calibration_result: Calibration result to be serialized. + :param target_file: Path to the file the result is written to. + """ + raise NotImplementedError + + +class RecorderAndDumper(metaclass=ABCMeta): + """ + Record and dump calibration data. + """ + + @property + @abstractmethod + def recorder(self) -> CalibrationRecorder: + """ + Recorder used for acquiring the calibration data + """ + raise NotImplementedError + + @property + @abstractmethod + def dumpers(self) -> List[CalibrationDumper]: + """ + List of Dumpers used for serializing the calibration data. All will + serialize the data that has been acquired in a single calibration run. + """ + raise NotImplementedError + + def record_and_dump(self, connection: ConnectionHandle, + deployment_folder: Path): + """ + Record calibration data and dump it to a file. + :param connection: Connection used to acquire the calibration data + :param deployment_folder: Folder the file with serialized results is + created in. + """ + result = self.recorder.generate_calib(connection) + + for dumper in self.dumpers: + filename = f"{self.recorder.calibration_type}_" \ + f"{dumper.format_name}{dumper.format_extension}" + target_file = deployment_folder.joinpath(filename) + dumper.dump_calibration(result, target_file) + + +class HagenCalibRecorder(CalibrationRecorder): + """ + Recorder for a canonical Hagen-Mode calibration. + """ + calibration_type = "hagen" + + def generate_calib(self, + connection: ConnectionHandle) -> CalibrationResult: + builder, _ = ExperimentInit().generate() + run(connection, builder.done()) + return calix.hagen.calibrate(connection) + + +class SpikingCalibRecorder(CalibrationRecorder): + """ + Recorder for a default Spiking-Mode calibration. + """ + calibration_type = "spiking" + + def generate_calib(self, + connection: ConnectionHandle) -> CalibrationResult: + builder, _ = ExperimentInit().generate() + run(connection, builder.done()) + return calix.spiking.calibrate(connection) + + +class CalixFormatDumper(CalibrationDumper): + """ + Dumper for the calix-internal data format. + """ + format_name = "calix-native" + format_extension = ".pkl" + + def dump_calibration(self, calibration_result: CalibrationResult, + target_file: Path): + with target_file.open(mode="wb") as target: + pickle.dump(calibration_result, target) + + +class CocoListPortableBinaryFormatDumper(CalibrationDumper): + """ + Dumper for the Coordinate-Container-List data in portable binary format. + """ + format_name = "cocolist" + format_extension = ".pbin" + + def dump_calibration(self, calibration_result: CalibrationResult, + target_file: Path): + builder = PlaybackProgramBuilderDumper() + calibration_result.apply(builder) + + with target_file.open(mode="wb") as target: + target.write(to_portablebinary(builder.done())) + + +class CocoListJsonFormatDumper(CalibrationDumper): + """ + Dumper for the Coordinate-Container-List data in json format + with gzip compression. + """ + format_name = "cocolist" + format_extension = ".json.gz" + + def dump_calibration(self, calibration_result: CalibrationResult, + target_file: Path): + builder = PlaybackProgramBuilderDumper() + calibration_result.apply(builder) + + with gzip.open(target_file, mode="wt") as target: + target.write(to_json(builder.done())) + + +class HagenCalibration(RecorderAndDumper): + recorder = HagenCalibRecorder() + dumpers = [CalixFormatDumper(), + CocoListPortableBinaryFormatDumper(), + CocoListJsonFormatDumper()] + + +class SpikingCalibration(RecorderAndDumper): + recorder = SpikingCalibRecorder() + dumpers = [CalixFormatDumper(), + CocoListPortableBinaryFormatDumper(), + CocoListJsonFormatDumper()] + + +def run_and_save_all(deployment_folder: Path): + """ + Executes all available default calibrations and saves them to all + available file formats. + + :param deployment_folder: Path calibration results are deployed to. + """ + with ManagedConnection() as connection: + for calib in [HagenCalibration(), SpikingCalibration()]: + calib.record_and_dump(connection, deployment_folder) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "deployment_folder", + help="Path to save all calibration result to. Directories are " + + "created if not already present.") + args = parser.parse_args() + + depl_folder = Path(args.deployment_folder) + depl_folder.mkdir(parents=True, exist_ok=True) + run_and_save_all(depl_folder) diff --git a/tests/hw/py/test_generate_default_calib.py b/tests/hw/py/test_generate_default_calib.py new file mode 100644 index 0000000000000000000000000000000000000000..9446675d93572caed5f1b59f1b2a5a139051a39e --- /dev/null +++ b/tests/hw/py/test_generate_default_calib.py @@ -0,0 +1,37 @@ +""" +Test the script that generates and serializes default calibrations. +""" + +import unittest +import os +from pathlib import Path + +from calix.scripts import calix_generate_default_calibration + + +class TestGenerateDefaultCalibration(unittest.TestCase): + """ + Run the default calibrations, assert they run and the generated files + are not empty. + """ + + def test_run_and_save_all(self): + calix_generate_default_calibration.run_and_save_all(Path(".")) + + expected_prefixes = ["hagen", "spiking"] + expected_suffixes = ["calix-native.pkl", "cocolist.pbin", + "cocolist.json.gz"] + + for expected_prefix in expected_prefixes: + for expected_suffix in expected_suffixes: + expected_filename = "_".join( + [expected_prefix, expected_suffix]) + + self.assertGreater( + os.path.getsize(expected_filename), 0, + "File size of expected calibration result " + + f"{expected_filename} is zero.") + + +if __name__ == '__main__': + unittest.main() diff --git a/wscript b/wscript index 428f23952ccbf2d61ed6081e7063c8976450ae57..5742a813566075ee0f15b310509a793006919184 100644 --- a/wscript +++ b/wscript @@ -1,5 +1,6 @@ import os from os.path import join +from waflib import Utils from waflib.extras.test_base import summary from waflib.extras.symwaf2ic import get_toplevel_path @@ -38,6 +39,18 @@ def build(bld): test_timeout=120, ) + bld(name='calix_scripts', + features='py pylint pycodestyle', + source=bld.path.ant_glob('src/py/calix/scripts/**/*.py'), + use='calix_pylib', + install_path='${PREFIX}/bin', + install_from='src/py/calix/scripts', + chmod=Utils.O755, + pylint_config=join(get_toplevel_path(), "code-format", "pylintrc"), + pycodestyle_config=join(get_toplevel_path(), "code-format", "pycodestyle"), + test_timeout=120, + ) + bld(name='calix_pyswtests', tests=bld.path.ant_glob('tests/sw/py/**/*.py'), features='pytest pylint pycodestyle',