text:
# Repository: https://gitlab.com/quantify-os/quantify-scheduler
# Licensed according to the LICENCE file on the master branch
"""Module containing Qblox InstrumentCoordinator Components."""
from __future__ import annotations

import dataclasses
from typing import Any, Dict, Optional, Tuple, Callable, Union, Type, List
from collections import namedtuple

import logging

import copy
from abc import abstractmethod
from dataclasses import dataclass

import numpy as np
from pulsar_qcm import pulsar_qcm
from pulsar_qrm import pulsar_qrm
from qcodes.instrument.base import Instrument
from quantify_scheduler.instrument_coordinator.components import base
from quantify_scheduler.backends.types.qblox import (
    PulsarSettings,
    PulsarRFSettings,
    SequencerSettings,
)
from quantify_scheduler.types import AcquisitionMetadata
from quantify_scheduler.backends.qblox import constants

logger = logging.getLogger(__name__)
logger.setLevel(logging.WARNING)

_SequencerStateType = Dict[str, Union[str, List[str]]]
"""
Type of the return value of get_sequencer_state. Returned value format is always a dict
with a str state under 'status' and a list of str flags under 'flags'.
"""

@dataclasses.dataclass(frozen=True)
class _SequencerStateInfo:
    message: str
    """The text to pass as the logger message."""
    logging_level: int
    """The logger level to use."""

_SEQUENCER_STATE_FLAG_INFO: Dict[str, _SequencerStateInfo] = {
    "DISARMED": _SequencerStateInfo(
        message="Sequencer was disarmed.", logging_level=logging.INFO
    ),
    "FORCED STOP": _SequencerStateInfo(
        message="Sequencer was stopped while still running.",
        logging_level=logging.INFO,
    ),
    "SEQUENCE PROCESSOR Q1 ILLEGAL INSTRUCTION": _SequencerStateInfo(
        message="Classical sequencer part executed an unknown instruction.",
        logging_level=logging.ERROR,
    ),
    "SEQUENCE PROCESSOR RT EXEC ILLEGAL INSTRUCTION": _SequencerStateInfo(
        message="Real-time sequencer part executed an unknown instruction.",
        logging_level=logging.ERROR,
    ),
    "AWG WAVE PLAYBACK INDEX INVALID PATH 0": _SequencerStateInfo(
        message="AWG path 0 tried to play an unknown waveform.",
        logging_level=logging.ERROR,
    ),
    "AWG WAVE PLAYBACK INDEX INVALID PATH 1": _SequencerStateInfo(
        message="AWG path 1 tried to play an unknown waveform.",
        logging_level=logging.ERROR,
    ),
    "ACQ WEIGHT PLAYBACK INDEX INVALID PATH 0": _SequencerStateInfo(
        message="Acquisition path 0 tried to play an unknown weight.",
        logging_level=logging.ERROR,
    ),
    "ACQ WEIGHT PLAYBACK INDEX INVALID PATH 1": _SequencerStateInfo(
        message="Acquisition path 1 tried to play an unknown weight.",
        logging_level=logging.ERROR,
    ),
    "ACQ SCOPE DONE PATH 0": _SequencerStateInfo(
        message="Scope acquisition for path 0 has finished.",
        logging_level=logging.DEBUG,
    ),
    "ACQ SCOPE OUT-OF-RANGE PATH 0": _SequencerStateInfo(
        message="Scope acquisition data for path 0 was out-of-range.",
        logging_level=logging.WARNING,
    ),
    "ACQ SCOPE OVERWRITTEN PATH 0": _SequencerStateInfo(
        message="Scope acquisition data for path 0 was overwritten.",
        logging_level=logging.INFO,
    ),
    "ACQ SCOPE DONE PATH 1": _SequencerStateInfo(
        message="Scope acquisition for path 1 has finished.",
        logging_level=logging.DEBUG,
    ),
    "ACQ SCOPE OUT-OF-RANGE PATH 1": _SequencerStateInfo(
        message="Scope acquisition data for path 1 was out-of-range.",
        logging_level=logging.WARNING,
    ),
    "ACQ SCOPE OVERWRITTEN PATH 1": _SequencerStateInfo(
        message="Scope acquisition data for path 1 was overwritten.",
        logging_level=logging.INFO,
    ),
    "ACQ BINNING DONE": _SequencerStateInfo(
        message="Acquisition binning completed.", logging_level=logging.DEBUG
    ),
    "ACQ BINNING FIFO ERROR": _SequencerStateInfo(
        message="Acquisition binning encountered internal FIFO error.",
        logging_level=logging.ERROR,
    ),
    "ACQ BINNING COMM ERROR": _SequencerStateInfo(
        message="Acquisition binning encountered internal communication error.",
        logging_level=logging.ERROR,
    ),
    "ACQ BINNING OUT-OF-RANGE": _SequencerStateInfo(
        message="Acquisition binning data out-of-range.", logging_level=logging.WARNING
    ),
    "ACQ INDEX INVALID": _SequencerStateInfo(
        message="Acquisition tried to process an invalid acquisition.",
        logging_level=logging.ERROR,
    ),
    "ACQ BIN INDEX INVALID": _SequencerStateInfo(
        message="Acquisition tried to process an invalid bin.",
        logging_level=logging.ERROR,
    ),
    "CLOCK INSTABILITY": _SequencerStateInfo(
        message="Clock source instability occurred.", logging_level=logging.ERROR
    ),
}
"""Used to link all the messages for the logger and levels to specific flags given by
the hardware."""

@dataclass(frozen=True)
class _StaticHardwareProperties:
    """Dataclass that holds all the static differences between the different Qblox
    devices that are relevant for configuring them correctly."""

    settings_type: Type[PulsarSettings]
    """The settings dataclass to use that the hardware needs to configure to."""
    has_internal_lo: bool
    """Specifies if an internal lo source is available."""
    number_of_sequencers: int
    """The number of sequencers the hardware has available."""
    number_of_output_paths: int
    """The number of output paths that can be used."""

_PULSAR_QCM_PROPERTIES = _StaticHardwareProperties(
    settings_type=PulsarSettings,
    has_internal_lo=False,
    number_of_sequencers=constants.NUMBER_OF_SEQUENCERS_QCM,
    number_of_output_paths=4,
)
_PULSAR_QRM_PROPERTIES = _StaticHardwareProperties(
    settings_type=PulsarSettings,
    has_internal_lo=False,
    number_of_sequencers=constants.NUMBER_OF_SEQUENCERS_QRM,
    number_of_output_paths=2,
)
_PULSAR_QCM_RF_PROPERTIES = _StaticHardwareProperties(
    settings_type=PulsarRFSettings,
    has_internal_lo=True,
    number_of_sequencers=constants.NUMBER_OF_SEQUENCERS_QCM,
    number_of_output_paths=4,
)
_PULSAR_QRM_RF_PROPERTIES = _StaticHardwareProperties(
    settings_type=PulsarRFSettings,
    has_internal_lo=True,
    number_of_sequencers=constants.NUMBER_OF_SEQUENCERS_QRM,
    number_of_output_paths=2,
)

class PulsarInstrumentCoordinatorComponent(base.InstrumentCoordinatorComponentBase):
    """Qblox Pulsar InstrumentCoordinator component base class."""

    def __init__(self, instrument: Instrument, **kwargs) -> None:
        """Create a new instance of PulsarInstrumentCoordinatorComponent base class."""
        super().__init__(instrument, **kwargs)
        if (
            instrument._get_lo_hw_present()
            is not self._hardware_properties.has_internal_lo
        ):
            raise RuntimeError(
                "PulsarInstrumentCoordinatorComponent not compatible with the "
                "provided instrument. Please confirm whether your device "
                "is a RF module (has an internal LO)."
            )
        self._seq_name_to_idx_map = {
            f"seq{idx}": idx
            for idx in range(self._hardware_properties.number_of_sequencers)
        }

    @property
    def is_running(self) -> bool:
        """
        Finds if any of the sequencers is currently running.

        Returns
        -------
        :
            True if any of the sequencers reports the "RUNNING" status.
        """
        for seq_idx in range(self._hardware_properties.number_of_sequencers):
            seq_state = self.instrument.get_sequencer_state(seq_idx)
            if seq_state["status"] == "RUNNING":
                return True
        return False

    def wait_done(self, timeout_sec: int = 10) -> None:
        """
        Blocks the instrument until all the sequencers are done running.

        Parameters
        ----------
        timeout_sec
            The timeout in seconds. N.B. the instrument takes the timeout in minutes
            (int), therefore it is rounded down to whole minutes with a minimum of 1.
        """
        timeout_min = timeout_sec // 60
        if timeout_min == 0:
            timeout_min = 1
        for idx in range(self._hardware_properties.number_of_sequencers):
            state: _SequencerStateType = self.instrument.get_sequencer_state(
                idx, timeout_min
            )
            flags = state.get("flags", None)
            if flags:
                for flag in flags:
                    if flag not in _SEQUENCER_STATE_FLAG_INFO:
                        logger.error(
                            f"[{self.name}|seq{idx}] Encountered flag {flag} in "
                            f"returned value by `get_sequencer_state` which is not "
                            f"defined in {self.__module__}."
                        )
                    else:
                        flag_info = _SEQUENCER_STATE_FLAG_INFO[flag]
                        msg = f"[{self.name}|seq{idx}] {flag} - {flag_info.message}"
                        logger.log(level=flag_info.logging_level, msg=msg)

    def start(self) -> None:
        """
        Starts execution of the schedule.
        """
        for idx in range(self._hardware_properties.number_of_sequencers):
            state = self.instrument.get_sequencer_state(idx)
            if state["status"] == "ARMED":
                self.instrument.start_sequencer(idx)

    def stop(self) -> None:
        """
        Stops all execution.
        """
        self.instrument.stop_sequencer()

    def _configure_global_settings(self, settings: PulsarSettings) -> None:
        """
        Configures all settings that are set globally for the whole instrument.

        Parameters
        ----------
        settings
            The settings to configure it to.
        """
        self.instrument.set("reference_source", settings.ref)

    def _configure_sequencer_settings(
        self, seq_idx: int, settings: SequencerSettings
    ) -> None:
        """
        Configures all sequencer-specific settings.

        Parameters
        ----------
        seq_idx
            Index of the sequencer to configure.
        settings
            The settings to configure it to.
        """
        self.instrument.set(f"sequencer{seq_idx}_sync_en", settings.sync_en)

        nco_en: bool = settings.nco_en
        self.instrument.set(f"sequencer{seq_idx}_mod_en_awg", nco_en)
        if nco_en:
            self.instrument.set(
                f"sequencer{seq_idx}_nco_freq", settings.modulation_freq
            )
        self.instrument.set(
            f"sequencer{seq_idx}_mixer_corr_phase_offset_degree",
            settings.mixer_corr_phase_offset_degree,
        )
        self.instrument.set(
            f"sequencer{seq_idx}_mixer_corr_gain_ratio",
            settings.mixer_corr_gain_ratio,
        )

        for output_idx in range(self._hardware_properties.number_of_output_paths):
            connected: bool = output_idx in settings.connected_outputs
            self.instrument.set(
                _get_channel_map_parameter_name(
                    sequencer_index=seq_idx, output_index=output_idx
                ),
                connected,
            )

    def _arm_all_sequencers_in_program(self, program: Dict[str, Any]):
        """Arms all the sequencers that are part of the program."""
        for seq_name in program:
            if seq_name in self._seq_name_to_idx_map:
                seq_idx = self._seq_name_to_idx_map[seq_name]
                self.instrument.arm_sequencer(sequencer=seq_idx)

    @property
    @abstractmethod
    def _hardware_properties(self) -> _StaticHardwareProperties:
        """
        Holds all the differences between the different modules.

        Returns
        -------
        :
            A dataclass with all the hardware properties for this specific module.
        """

# pylint: disable=too-many-ancestors
class PulsarQCMComponent(PulsarInstrumentCoordinatorComponent):
    """
    Pulsar QCM specific InstrumentCoordinator component.
    """

    _hardware_properties = _PULSAR_QCM_PROPERTIES

    def __init__(self, instrument: pulsar_qcm.pulsar_qcm_qcodes, **kwargs) -> None:
        """Create a new instance of PulsarQCMComponent."""
        assert isinstance(instrument, pulsar_qcm.pulsar_qcm_qcodes)
        super().__init__(instrument, **kwargs)

    @property
    def instrument(self) -> pulsar_qcm.pulsar_qcm_qcodes:
        return super().instrument

    def retrieve_acquisition(self) -> None:
        """
        Retrieves the previous acquisition.

        Returns
        -------
        :
            QCM returns None since the QCM has no acquisition.
        """
        return None

    def prepare(self, options: Dict[str, dict]) -> None:
        """
        Uploads the waveforms and programs to the sequencers and
        configures all the settings required. Keep in mind that values set directly
        through the driver may be overridden (e.g. the offsets will be set according to
        the specified mixer calibration parameters).

        Parameters
        ----------
        options
            Program to upload to the sequencers. The key is a sequencer, e.g.,
            :code:`"seq0"`, or :code:`"settings"`,
            the value is the global settings dict or a sequencer-specific configuration.
        """
        program = copy.deepcopy(options)

        if "settings" in program:
            settings_entry = program.pop("settings")
            pulsar_settings = self._hardware_properties.settings_type.from_dict(
                settings_entry
            )
            self._configure_global_settings(pulsar_settings)
        for seq_idx in range(self._hardware_properties.number_of_sequencers):
            self.instrument.set(f"sequencer{seq_idx}_sync_en", False)

        for seq_name, seq_cfg in program.items():
            if seq_name in self._seq_name_to_idx_map:
                seq_idx = self._seq_name_to_idx_map[seq_name]
            else:
                raise KeyError(
                    f"Invalid program. Attempting to access non-existing sequencer with"
                    f' name "{seq_name}".'
                )
            if "settings" in seq_cfg:
                seq_settings = SequencerSettings.from_dict(seq_cfg["settings"])
                self._configure_sequencer_settings(
                    seq_idx=seq_idx, settings=seq_settings
                )

            self.instrument.set(
                f"sequencer{seq_idx}_waveforms_and_program", seq_cfg["seq_fn"]
            )

        self._arm_all_sequencers_in_program(program)

    def _configure_global_settings(self, settings: PulsarSettings):
        """
        Configures all settings that are set globally for the whole instrument.

        Parameters
        ----------
        settings
            The settings to configure it to.
        """
        super()._configure_global_settings(settings)
        # configure mixer correction offsets
        if settings.offset_ch0_path0 is not None:
            self.instrument.set("out0_offset", settings.offset_ch0_path0)
        if settings.offset_ch0_path1 is not None:
            self.instrument.set("out1_offset", settings.offset_ch0_path1)
        if settings.offset_ch1_path0 is not None:
            self.instrument.set("out2_offset", settings.offset_ch1_path0)
        if settings.offset_ch1_path1 is not None:
            self.instrument.set("out3_offset", settings.offset_ch1_path1)

# pylint: disable=too-many-ancestors
class PulsarQRMComponent(PulsarInstrumentCoordinatorComponent):
    """
    Pulsar QRM specific InstrumentCoordinator component.
    """

    _hardware_properties = _PULSAR_QRM_PROPERTIES

    def __init__(self, instrument: pulsar_qrm.pulsar_qrm_qcodes, **kwargs) -> None:
        """Create a new instance of PulsarQRMComponent."""
        assert isinstance(instrument, pulsar_qrm.pulsar_qrm_qcodes)
        self._acquisition_manager: Optional[_QRMAcquisitionManager] = None
        """Holds all the acquisition related logic."""
        super().__init__(instrument, **kwargs)

    @property
    def instrument(self) -> pulsar_qrm.pulsar_qrm_qcodes:
        return super().instrument

    def retrieve_acquisition(self) -> Union[Dict[Tuple[int, int], Any], None]:
        """
        Retrieves the latest acquisition results.

        Returns
        -------
        :
            The acquired data.
        """
        if self._acquisition_manager is None:  # No acquisition has been prepared.
            return None
        return self._acquisition_manager.retrieve_acquisition()

    def prepare(self, options: Dict[str, dict]) -> None:
        """
        Uploads the waveforms and programs to the sequencers and
        configures all the settings required. Keep in mind that values set directly
        through the driver may be overridden (e.g. the offsets will be set according to
        the specified mixer calibration parameters).

        Parameters
        ----------
        options
            Program to upload to the sequencers. The key is a sequencer, e.g.,
            :code:`"seq0"`, or :code:`"settings"`,
            the value is the global settings dict or a sequencer-specific configuration.
        """
        program = copy.deepcopy(options)
        if "acq_metadata" in program:
            acq_metadata = program.pop("acq_metadata")
        if "acq_mapping" in program:  # Resets everything to do with acquisition.

            acq_mapping = program.pop("acq_mapping")
            self._acquisition_manager = _QRMAcquisitionManager(
                self,
                number_of_sequencers=self._hardware_properties.number_of_sequencers,
                acquisition_mapping=acq_mapping,
                acquisition_metadata=acq_metadata,
            )
        else:
            self._acquisition_manager = None

        if "settings" in program:
            settings_entry = program.pop("settings")
            pulsar_settings = self._hardware_properties.settings_type.from_dict(
                settings_entry
            )
            if self._acquisition_manager is not None:
                self._acquisition_manager.scope_mode_sequencer = (
                    pulsar_settings.scope_mode_sequencer
                )
            self._configure_global_settings(pulsar_settings)

        for path in [0, 1]:
            self.instrument.set(f"scope_acq_trigger_mode_path{path}", "sequencer")
            self.instrument.set(f"scope_acq_avg_mode_en_path{path}", True)

        for seq_name, seq_cfg in program.items():
            if seq_name in self._seq_name_to_idx_map:
                seq_idx = self._seq_name_to_idx_map[seq_name]
            else:
                raise KeyError(
                    f"Invalid program. Attempting to access non-existing sequencer with"
                    f' name "{seq_name}".'
                )
            if "settings" in seq_cfg:
                seq_settings = SequencerSettings.from_dict(seq_cfg["settings"])
                self._configure_sequencer_settings(
                    seq_idx=seq_idx, settings=seq_settings
                )

            self.instrument.set(
                f"sequencer{seq_idx}_waveforms_and_program", seq_cfg["seq_fn"]
            )

        self._arm_all_sequencers_in_program(program)

    def _configure_global_settings(self, settings: PulsarSettings):
        """
        Configures all settings that are set globally for the whole instrument.

        Parameters
        ----------
        settings
            The settings to configure it to.
        """
        super()._configure_global_settings(settings)
        # configure mixer correction offsets
        if settings.offset_ch0_path0 is not None:
            self.instrument.set("out0_offset", settings.offset_ch0_path0)
        if settings.offset_ch0_path1 is not None:
            self.instrument.set("out1_offset", settings.offset_ch0_path1)

    def _configure_sequencer_settings(
        self, seq_idx: int, settings: SequencerSettings
    ) -> None:
        super()._configure_sequencer_settings(seq_idx, settings)
        if settings.integration_length_acq is not None:
            self.instrument.set(
                f"sequencer{seq_idx}_integration_length_acq",
                settings.integration_length_acq,
            )
            self._acquisition_manager.integration_length_acq = (
                settings.integration_length_acq
            )
        self.instrument.set(f"sequencer{seq_idx}_demod_en_acq", settings.nco_en)

class PulsarQCMRFComponent(PulsarQCMComponent):
    """
    Pulsar QCM-RF specific InstrumentCoordinator component.
    """

    _hardware_properties = _PULSAR_QCM_RF_PROPERTIES

    def _configure_global_settings(self, settings: PulsarSettings):
        """
        Configures all settings that are set globally for the whole instrument.

        Parameters
        ----------
        settings
            The settings to configure it to.
        """
        self.instrument.set("reference_source", settings.ref)

        if settings.lo0_freq is not None:
            self.instrument.set("out0_lo_freq", settings.lo0_freq)
        if settings.lo1_freq is not None:
            self.instrument.set("out1_lo_freq", settings.lo1_freq)

        # configure mixer correction offsets
        if settings.offset_ch0_path0 is not None:
            self.instrument.set("out0_offset_path0", settings.offset_ch0_path0)
        if settings.offset_ch0_path1 is not None:
            self.instrument.set("out0_offset_path1", settings.offset_ch0_path1)
        if settings.offset_ch1_path0 is not None:
            self.instrument.set("out1_offset_path0", settings.offset_ch1_path0)
        if settings.offset_ch1_path1 is not None:
            self.instrument.set("out1_offset_path1", settings.offset_ch1_path1)

class PulsarQRMRFComponent(PulsarQRMComponent):
    """
    Pulsar QRM-RF specific InstrumentCoordinator component.
    """

    _hardware_properties = _PULSAR_QRM_RF_PROPERTIES

    def _configure_global_settings(self, settings: PulsarSettings):
        """
        Configures all settings that are set globally for the whole instrument.

        Parameters
        ----------
        settings
            The settings to configure it to.
        """
        self.instrument.set("reference_source", settings.ref)

        if settings.lo0_freq is not None:
            self.instrument.set("out0_in0_lo_freq", settings.lo0_freq)

        # configure mixer ccorrection offsets
        if settings.offset_ch0_path0 is not None:
            self.instrument.set("out0_offset_path0", settings.offset_ch0_path0)
        if settings.offset_ch0_path1 is not None:
            self.instrument.set("out0_offset_path1", settings.offset_ch0_path1)

def _get_channel_map_parameter_name(sequencer_index: int, output_index: int):
    path_idx = output_index % 2  # even or odd output
    return f"sequencer{sequencer_index}_channel_map_path{path_idx}_out{output_index}_en"

AcquisitionIndexing = namedtuple("AcquisitionIndexing", "acq_channel acq_index")
"""
Named tuple to clarify how the indexing of acquisitions works inside the
`_QRMAcquisitionManager`.
"""

class _QRMAcquisitionManager:
    """
    Utility class that handles the acquisitions performed with the QRM.

    An instance of this class is meant to exist only for a single prepare-start-
    retrieve_acquisition cycle to prevent stateful behavior.
    """

    def __init__(
        self,
        parent: PulsarQRMComponent,
        number_of_sequencers: int,
        acquisition_mapping: Dict[Tuple[int, int], Tuple[str, str]],
        acquisition_metadata: AcquisitionMetadata,
    ):
        """
        Constructor for `_QRMAcquisitionManager`.

        Parameters
        ----------
        parent
            Reference to the parent QRM IC component.
        number_of_sequencers
            The number of sequencers capable of acquisitions.
        acquisition_mapping
            The acquisition mapping extracted from the schedule, this mapping links the
            `acq_channel` and `acq_index` to the sequencer name and acquisition
            protocol. The key is a tuple (`acq_channel`, `acq_index`), the values
            (seq_name, protocol).
        acquisition_metadata
            Provides a summary of the used channels bins and acquisition protocols.
        """
        self.parent: PulsarQRMComponent = parent
        self.number_of_sequencers: int = number_of_sequencers
        self.acquisition_mapping: Dict[
            Tuple[int, int], Tuple[str, str]
        ] = acquisition_mapping
        self.acquisition_metadata: AcquisitionMetadata = acquisition_metadata

        self.scope_mode_sequencer: Optional[str] = None
        self.integration_length_acq: Optional[int] = None
        self.seq_name_to_idx_map = {
            f"seq{idx}": idx for idx in range(number_of_sequencers)
        }

    @property
    def instrument(self):
        """Returns the QRM driver from the parent IC component."""
        return self.parent.instrument

    def retrieve_acquisition(self) -> Dict[Tuple[int, int], Any]:
        """
        Retrieves all the acquisition data in the correct format.

        Returns
        -------
        :
            The acquisitions with the protocols specified in the `acq_mapping` as a
            `dict` with the `(acq_channel, acq_index)` as keys.
        """

        protocol_to_function_mapping = {
            "weighted_integrated_complex": self._get_integration_data,
            "ssb_integration_complex": self._get_integration_amplitude_data,
            "trace": self._get_scope_data,
            # NB thresholded protocol is still missing since there is nothing in
            # the acquisition library for it yet.
        }
        self._store_scope_acquisition()

        formatted_acquisitions: Dict[AcquisitionIndexing, Any] = {}

        for seq_idx in range(self.number_of_sequencers):
            if f"seq{seq_idx}" not in self.acquisition_metadata:
                continue
            acq_metadata = self.acquisition_metadata[f"seq{seq_idx}"]
            acquisition_function: Callable = protocol_to_function_mapping[
                acq_metadata.acq_protocol
            ]

            # retrieve the raw data from the qrm sequencer module
            acquisitions = self.instrument.get_acquisitions(seq_idx)
            for acq_channel, acq_indices in acq_metadata.acq_indices.items():
                # the acquisition_function retrieves the right part of the acquisitions
                # data structure returned by the qrm
                i_vals, q_vals = acquisition_function(
                    acquisitions=acquisitions, acq_channel=acq_channel
                )

                # the Qblox compilation backend verifies that the
                # acquisition indices start at 0 and increment in steps of 1.
                # this enables us to simply stride over the bin_idx as if they
                # correspond to acq_indices.
                for acq_idx in acq_indices:
                    acq_stride = len(acq_indices)
                    # N.B. the stride idx ensures that in append mode all data
                    # corresponding to the same acq_index appears in the
                    # same acq_ch, acq_idx part of the returned formatted acquisitions.
                    formatted_acquisitions[
                        AcquisitionIndexing(acq_channel=acq_channel, acq_index=acq_idx)
                    ] = (
                        i_vals[acq_idx::acq_stride],
                        q_vals[acq_idx::acq_stride],
                    )

        return formatted_acquisitions

    def _store_scope_acquisition(self):
        sequencer_index = self.seq_name_to_idx_map.get(self.scope_mode_sequencer)
        if sequencer_index is None:
            return

        if sequencer_index > self.number_of_sequencers:
            raise ValueError(
                f"Attempting to retrieve scope mode data from sequencer "
                f"{sequencer_index}. A QRM has only "
                f"{self.number_of_sequencers} sequencers."
            )
        scope_ch_and_idx = self._get_scope_channel_and_index()
        if scope_ch_and_idx is not None:
            acq_channel, _ = scope_ch_and_idx
            acq_name = self._channel_index_to_channel_name(acq_channel)
            self.instrument.store_scope_acquisition(sequencer_index, acq_name)

    def _get_protocol(self, acq_channel, acq_index) -> str:
        """
        Returns the acquisition protocol corresponding to acq_channel with
        acq_index.
        """
        return self.acquisition_mapping[(acq_channel, acq_index)][1]

    def _get_sequencer_index(self, acq_channel, acq_index) -> str:
        """
        Returns the seq idx corresponding to acq_channel with
        acq_index.
        """
        seq_name = self.acquisition_mapping[(acq_channel, acq_index)][0]
        return self.seq_name_to_idx_map[seq_name]

    def _get_scope_channel_and_index(self) -> Optional[AcquisitionIndexing]:
        """
        Returns the first `(acq_channel, acq_index)` pair that uses `"trace"`
        acquisition. Returns `None` if none of them do.
        """
        ch_and_idx: Optional[AcquisitionIndexing] = None
        for key, value in self.acquisition_mapping.items():
            if value[1] == "trace":
                if ch_and_idx is not None:
                    # Pylint seems to not care we explicitly check for None
                    # pylint: disable=unpacking-non-sequence
                    acq_channel, acq_index = ch_and_idx
                    raise RuntimeError(
                        f"A scope mode acquisition is defined for both acq_channel "
                        f"{acq_channel} with acq_index {acq_index} as well as "
                        f"acq_channel {key[0]} with acq_index {key[1]}. Only a single "
                        f"trace acquisition is allowed per QRM."
                    )
                ch_and_idx: AcquisitionIndexing = key
        return ch_and_idx

    def _get_scope_data(
        self, acquisitions: dict, acq_channel: int = 0
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Retrieves the scope mode acquisition associated with an `acq_channel`.

        Parameters
        ----------
        acq_channel
            The acq_channel to get the scope mode acquisition for.

        Returns
        -------
        scope_data_i
            The scope mode data for `path0`.
        scope_data_q
            The scope mode data for `path1`.
        """
        acq_name = self._channel_index_to_channel_name(acq_channel)
        scope_data = acquisitions[acq_name]["acquisition"]["scope"]
        for path_label in ("path0", "path1"):
            if scope_data[path_label]["out-of-range"]:
                logger.warning(
                    f"The scope mode data of {path_label} of {self.parent.name} with "
                    f"acq_channel={acq_channel}  was out-of-range."
                )
        # NB hardware already divides by avg_count for scope mode
        scope_data_i = scope_data["path0"]["data"]
        scope_data_q = scope_data["path1"]["data"]
        return scope_data_i, scope_data_q

    def _get_integration_data(
        self, acquisitions: dict, acq_channel: int = 0
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Retrieves the integrated acquisition data associated with an `acq_channel`.

        Parameters
        ----------
        acquisitions
            The acquisitions dict as returned by the sequencer.
        acq_channel
            The `acq_channel` from which to get the data.

        Returns
        -------
        i_data
            The integrated data for path0.
        q_data
            The integrated data for path1.
        """

        bin_data = self._get_bin_data(acquisitions, acq_channel)

        i_data, q_data = (
            np.array(bin_data["integration"]["path0"]),
            np.array(bin_data["integration"]["path1"]),
        )

        return i_data, q_data

    def _get_integration_amplitude_data(
        self, acquisitions: dict, acq_channel: int = 0
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Gets the integration data but normalized to the integration time (number of
        samples summed). The return value is thus the amplitude of the demodulated
        signal directly and has volt units (i.e. same units as a single sample of the
        integrated signal).

        Parameters
        ----------
        acquisitions
            The acquisitions dict as returned by the sequencer.
        acq_channel
            The `acq_channel` from which to get the data.

        Returns
        -------
        data_i
            Array containing I-quadrature data.
        data_q
            Array containing Q-quadrature data.
        """
        if self.integration_length_acq is None:
            raise RuntimeError(
                "Retrieving data failed. Expected the integration length to be defined,"
                " but it is `None`."
            )
        compensated_data_i, compensated_data_q = self._get_integration_data(
            acquisitions=acquisitions, acq_channel=acq_channel
        )
        compensated_data_i, compensated_data_q = (
            compensated_data_i / self.integration_length_acq,
            compensated_data_q / self.integration_length_acq,
        )
        return compensated_data_i, compensated_data_q

    def _get_threshold_data(
        self, acquisitions: dict, acq_channel: int = 0, acq_index: int = 0
    ) -> float:
        """
        Retrieves the thresholded acquisition data associated with `acq_channel` and
        `acq_index`.

        Parameters
        ----------
        acquisitions
            The acquisitions dict as returned by the sequencer.
        acq_channel
            The acq_channel to get the thresholded acquisition data for.
        acq_index
            The acq_index to get the thresholded acquisition data for.

        Returns
        -------
        :
            The value of the thresholded acquisition for `acq_channel` and `acq_index`.
            Should always be 0.0 <= val <= 1.0.
        """
        bin_data = self._get_bin_data(acquisitions, acq_channel)
        data = bin_data["threshold"]

        if acq_index > len(data):
            raise ValueError(
                f"Attempting to access acq_index {acq_index} on "
                f"{self.parent.name} but only {len(data)} values found "
                f"in acquisition data."
            )
        return data[acq_index]

    @staticmethod
    def _channel_index_to_channel_name(acq_channel: int) -> str:
        """Returns the name of the acquisition from the acq_channel."""
        return str(acq_channel)

    @classmethod
    def _get_bin_data(cls, acquisitions: dict, acq_channel: int = 0) -> dict:
        """Returns the bin entry of the acquisition data dict."""
        acq_name = cls._channel_index_to_channel_name(acq_channel)
        channel_data = acquisitions[acq_name]
        if channel_data["index"] != acq_channel:
            raise RuntimeError(
                f"Name does not correspond to a valid acquisition for name {acq_name}, "
                f'which has index {channel_data["index"]}.'
            )
        return channel_data["acquisition"]["bins"]

ClusterModule = Union[
    PulsarQCMComponent, PulsarQRMComponent, PulsarQRMRFComponent, PulsarQCMRFComponent
]
"""Type that combines all the possible modules for a cluster."""

class ClusterComponent(base.InstrumentCoordinatorComponentBase):
    """
    Class that represents an instrument coordinator component for a Qblox cluster.
    """

    def __init__(self, instrument: Instrument, **kwargs) -> None:
        """
        Create a new instance of the ClusterComponent.

        Parameters
        ----------
        instrument
            Reference to the cluster driver object.
        **kwargs
            Keyword arguments passed to the parent class.
        """
        super().__init__(instrument, **kwargs)
        self._cluster_modules: Dict[str, ClusterModule] = {}

    def add_modules(self, *modules: Instrument) -> None:
        """
        Add modules to the cluster.

        Parameters
        ----------
        *modules
            The QCoDeS drivers of the modules to add.
        """
        for mod in modules:
            self._cluster_modules[
                mod.name
            ] = _construct_component_from_instrument_driver(mod)

    @property
    def is_running(self) -> bool:
        """Returns true if any of the modules are currently running."""
        return any(comp.is_running for comp in self._cluster_modules.values())

    def start(self) -> None:
        """Starts all the modules in the cluster."""
        for comp in self._cluster_modules.values():
            comp.start()

    def stop(self) -> None:
        """Stops all the modules in the cluster."""
        for comp in self._cluster_modules.values():
            comp.stop()

    def _configure_cmm_settings(self, settings: Dict[str, Any]):
        """
        Sets all the settings of the CMM (Cluster Management Module) that have been
        provided by the backend.

        Parameters
        ----------
        settings
            A dictionary containing all the settings to set.
        """
        if "reference_source" in settings:
            self.instrument.set("reference_source", settings["reference_source"])

    def prepare(self, options: Dict[str, dict]) -> None:
        """
        Prepares the cluster component for execution of a schedule.

        Parameters
        ----------
        options
            The compiled instructions to configure the cluster to.
        """
        settings = options.pop("settings")
        self._configure_cmm_settings(settings=settings)
        for name, comp_options in options.items():
            if name not in self._cluster_modules:
                raise KeyError(
                    f"Attempting to prepare module {name} of cluster {self.name}, while"
                    f" module has not been added to the cluster component."
                )
            self._cluster_modules[name].prepare(comp_options)

    def retrieve_acquisition(self) -> Optional[Dict[Tuple[int, int], Any]]:
        """
        Retrieves all the data from the instruments.

        Returns
        -------
        :
            The acquired data or ``None`` if no acquisitions have been performed.
        """
        acquisitions: Dict[Tuple[int, int], Any] = {}
        for comp in self._cluster_modules.values():
            comp_acq = comp.retrieve_acquisition()
            if comp_acq is not None:
                acquisitions.update(comp_acq)
        return acquisitions if len(acquisitions) > 0 else None

    def wait_done(self, timeout_sec: int = 10) -> None:
        """
        Blocks until all the components are done executing their programs.

        Parameters
        ----------
        timeout_sec
            The time in seconds until the instrument is considered to have timed out.
        """
        for comp in self._cluster_modules.values():
            comp.wait_done(timeout_sec=timeout_sec)

def _construct_component_from_instrument_driver(
    driver: Instrument,
) -> ClusterModule:
    """
    Determines the corresponding ClusterModule type and constructs an IC component from
    the :doc:`qblox_instruments <qblox-instruments:index>` driver.

    Parameters
    ----------
    driver
        The ``qblox_instruments`` instrument driver.

    Returns
    -------
    :
        The corresponding IC component.
    """
    is_qcm: bool = isinstance(driver, pulsar_qcm.pulsar_qcm_qcodes)
    if not is_qcm and not isinstance(driver, pulsar_qrm.pulsar_qrm_qcodes):
        raise TypeError(
            f"Invalid driver type for '{driver.name}'. Cannot construct an instrument "
            f"coordinator component for driver of type '{type(driver)}'. "
            f"Expected types: {type(pulsar_qcm.pulsar_qcm_qcodes)} or "
            f"{type(pulsar_qrm.pulsar_qrm_qcodes)}."
        )
    is_rf: bool = driver._get_lo_hw_present()
    icc_class: type = {
        (True, False): PulsarQCMComponent,
        (True, True): PulsarQCMRFComponent,
        (False, False): PulsarQRMComponent,
        (False, True): PulsarQRMRFComponent,
    }[(is_qcm, is_rf)]
    return icc_class(driver)

ext: py

max_stars_repo_name: quantify-os/quantify-scheduler

max_issues_repo_name: quantify-os/quantify-scheduler

max_forks_repo_name: quantify-os/quantify-scheduler

reason: import numpy


text:
import scipy as sp

print(sp.zeros((3, 4, 3), sp.uint8))

max_stars_repo_name: MisterZhouZhou/pythonLearn

max_issues_repo_name: MisterZhouZhou/pythonLearn

max_forks_repo_name: MisterZhouZhou/pythonLearn

reason: import scipy


text:
#!/usr/bin/env python
import sys
sys.path.append("../")
import numpy as np
from utils import  WordDict
import cPickle, os
from utils import Utils

class AlphabetOptimiseSpacingRandomiser(object):
    def __init__(self):
        self.utils=  Utils()
        self.__randomiser = AlphabetRandomiser()
        self.__loadDictionary(i_load_from_file=True)

    def setParams(self, i_nchannels,  i_min_dist=3, i_max_repeat=2):
        self.__randomiser.setParams(i_nchannels, i_min_dist, i_max_repeat)

    def channel3Hack(self, i_possible_sequences):
        print "================================================"
        print "Using channel 3 hack"
        o_sequences = []
        for s in i_possible_sequences:
            group_space = np.array(self.__randomiser.getGrouping( s,i_nchannels=3, i_erase_spacers=False, i_display=False))
            group_no_space = self.__randomiser.getGrouping( s,i_nchannels=3, i_erase_spacers=True, i_display=False)
            for n in range(0, len(group_no_space)):
                g = np.array(group_no_space[n])
                for letter in g:
                    idx = np.nonzero(g == letter)[0]
                    if len(idx) < 2:
                        group_space[n][-1] = letter
            new_seq = group_space.transpose().flatten()
            o_sequences.append(list(new_seq))
        return (o_sequences, True)

    def compute(self, i_alphabet):
        possible_alphabets = self.__getPossibleAlphabets(i_alphabet)
        print "POSSIBLE"
        print possible_alphabets

        final_sequences = []
        nchannels = self.__randomiser.getChannels()
        for alphabet in possible_alphabets:
            print "*************************************"
            print "".join(alphabet)
            group = self.__randomiser.getGrouping( alphabet, i_nchannels=nchannels, i_erase_spacers=False, i_display=True )
            print "group = ", group
            (possible_sequences, niterations, max_iterations, is_valid) = self.__randomiser.compute(alphabet, i_display=False)
            #FIXME:
            if nchannels == 3:
                (possible_sequences, is_valid) = self.channel3Hack( possible_sequences)
            if is_valid:
                for s in possible_sequences:
                    final_sequences.append(s)
        print "NUMBER OF FINAL SEQ SIZE= ", len(final_sequences)
        if len(final_sequences) < 1:
            return final_sequences
        self.displayDistances(i_alphabet,final_sequences)

    def displayDistances(self, i_alphabet, final_sequences):
        nchannels = self.__randomiser.getChannels()
        final_scores  = []
        for seq  in final_sequences:
            alphabet_length = len(self.__randomiser.getSequenceAlphabet(i_alphabet))
            seq_alphabet = self.__randomiser.getSequenceAlphabet(seq )
            start_index = len(seq_alphabet) - alphabet_length
            #score  = self.getSeqMutualInformation( start_index, list(seq_alphabet) )
            score  = self.getSeqDistances( start_index, list(seq_alphabet) )
            final_scores.append(score)
        final_scores = np.array(final_scores)
        sort_arg = np.argsort(final_scores)
        for idx in sort_arg:
            print "============================================================================"
            print   " [ score = ", final_scores[idx] , "] sequence = ", ''.join(final_sequences[idx])
            for letter in final_sequences[idx]:
                print letter,
            print " "
            print "-------------------------------------------------------------------------------------------------------------------"
            groups = np.array(self.__randomiser.getGrouping(final_sequences[idx] , nchannels, i_erase_spacers=False, i_display=False))
            for group in groups:
                for letter in group:
                    print letter,
                print " "

    ##################################  Optimisation based on integer letter distances
    def getSeqDistances(self, i_start_index, i_letter_seq):
        distance = 0.0
        window_length = self.__randomiser.getMinDist()
        search_seq = np.array(i_letter_seq)[0:i_start_index]

        for n in range(i_start_index, len(i_letter_seq)-window_length):
            cur_letter= i_letter_seq[n]
            end_index = n+window_length+1
            if end_index > i_letter_seq:
                end_index = len(i_letter_seq)
            if end_index <= (n+1):
                continue
            next_letters = list(i_letter_seq[n-window_length:n])
            next_letters.extend(i_letter_seq[n+1:end_index])
            print "Cur letter = ",  cur_letter, " neighbours = ", ''.join(next_letters), " letter seq = ", "".join(i_letter_seq)

        return  distance

    ################################## Validation tests
    def __validateGroupsForSimilarSounds(self, i_alphabet):
        #validate the sequence to see if any successive sounds are similar
        for n in range(1, len(i_alphabet)):
            letter = i_alphabet[n]
            neighbour = i_alphabet[n-1]
            is_valid = self.__randomiser.validateSimilarSoundContraints(letter, neighbour)
            if not is_valid:
                return False
        return True
    ################## Determining the possible alphabets

    def __getPossibleAlphabets(self, i_alphabet):
        #Get all possible combinations of the alphabet, based on the constraint
        #That no group can be preceded by more than 1 spacer.
        nchannels = self.__randomiser.getChannels()
        modulus = len(i_alphabet) % nchannels
        if modulus == 0:
            return [self.__seqAlphabetToChannels(i_alphabet)]
        possible_alphabets = []
        nspacers =  nchannels - modulus
        spacer_combinations = self.__getSpaceCombinations(nchannels, nspacers)
        #Insert a spacer when there is a 1 in the spacer combination
        group_length = self.__getGroupLength(  i_alphabet, i_nspacers= nspacers)
        for combination in spacer_combinations:
            alphabet = list(i_alphabet)
            for n in range(0, len(combination)):
                if combination[n] < 1:
                    alphabet.insert(n*group_length, '*')
            list_alphabet = self.__seqAlphabetToChannels( alphabet, nspacers)
            if self.__validateGroupsForSimilarSounds( list_alphabet ):
                possible_alphabets.append(list_alphabet)
                #print grouped_alphabet.transpose()
                #print ''.join(list_alphabet)
        return possible_alphabets

    def __seqAlphabetToChannels(self, i_alphabet, i_nspacers=0):
        """Map the alphabet sequence abc to channel display so that abc is in sequence in a specific channel"""
        group_length = self.__getGroupLength(  i_alphabet, i_nspacers= i_nspacers)
        alphabet = np.array(i_alphabet)
        grouped_alphabet = []
        for n in range(0, group_length):
            grouped_alphabet.append(alphabet[range(n,len(alphabet),group_length)])
        grouped_alphabet = np.array(grouped_alphabet)
        list_alphabet = list(grouped_alphabet.flatten())
        return list_alphabet

    def __getGroupLength(self, i_alphabet, i_nspacers=0):
        nchannels = self.__randomiser.getChannels()
        group_length = (len(i_alphabet) + i_nspacers) / nchannels
        return group_length

    def __getSpaceCombinations(self, i_nchannels, i_nspacers):
        """Get the possible ways the spacer could be placed - only one spacer per group
           is allowed. First determine the number of spacers, e.g., if there are
           2 spacers for five channels a vector of the form [0,0,1,1,1] will be computed,
           and all its permutations will be returned. A zero is returned when the group
           should start with a spacer"""
        #Only allow a channel to start with one spacer
        combinations = np.int32(np.ones(i_nchannels))
        for n in range(0, i_nspacers):
            combinations[n] = 0
        combinations = list(combinations)
        #Find all permutations of the values in combinations
        permutations = np.array([ p  for p in self.utils.xpermutations(combinations) ])
        #Only use the unique ones
        o_permutations = [permutations[0,:]]
        for n in range(1, permutations.shape[0]):
            combination = permutations[n,:]
            stored = np.array(o_permutations)
            unique_thresh = permutations.shape[1]
            idx = np.nonzero( np.sum( np.int32( stored == combination ), 1) == unique_thresh )[0]
            if len(idx) == 0:
                o_permutations.append(combination)
        return np.array(o_permutations)

    ################################# Optimisation based on entropy

    def getSeqMutualInformation(self, i_start_index, i_letter_seq):
        mutual_info = 0.0
        window_length = 1 #self.__randomiser.getMinDist()
        for n in range(i_start_index, len(i_letter_seq)-window_length):
            cur_letter= i_letter_seq[n]
            end_index = n+window_length+1
            if end_index > i_letter_seq:
                end_index = len(i_letter_seq)
            if end_index <= (n+1):
                continue
            #next_letters = list(i_letter_seq[n-window_length:n])
            next_letters = i_letter_seq[n+1:end_index]
            letter_mutual_info = [self.getLetterMutualInformation(cur_letter,letter)for letter in next_letters]
            mutual_info += np.sum(np.array(letter_mutual_info))
        return mutual_info

    def getLetterMutualInformation(self, i_cur_letter, i_next_letter):
        log_px = np.log(self.__letter_probs[i_cur_letter]) - np.log(2)
        log_py = np.log(self.__letter_probs[i_next_letter]) - np.log(2)
        key = ''.join([i_cur_letter, i_next_letter])
        if not self.__letter_pair_probs.has_key(key):
            key = ''.join([i_next_letter, i_cur_letter])
        pxy =  self.__letter_pair_probs[key]
        log_pxy = np.log(pxy)
        if np.abs(pxy) < 1E-6:
            mutual_information = 0.0
        else:
            log_pxy -= np.log(2)
            mutual_information = np.sum(pxy*(log_pxy - log_px - log_py)  )
        return mutual_information

    def __marginalLetterLogProbs(self):
        alphabet = np.unique(self.__word_list)
        max_word_length = self.__word_list.shape[1]
        #Compute the probabilities that ln=to a specific letter for all letters in the alphabet
        o_probs = {}
        for letter in alphabet:
            o_probs[letter] = 0.0
        for letter in alphabet:
            for n in range(0 , max_word_length):
                #Compute p(l1,l2,... | ln=letter)
                idx = np.nonzero( self.__word_list[:,n] == letter )[0]
                if len(idx) < 1:
                    continue
                log_probs = np.array(self.__word_log_probs[idx].flatten())
                log_prob_sum = self.utils.expTrick( log_probs.reshape([1,len(log_probs)]))
                o_probs[letter] += np.exp(log_prob_sum)
        for letter in alphabet:
            o_probs[letter] /= 26.0
        return o_probs

    def __marginalJointLogProbs(self):
        alphabet = np.unique(self.__word_list)
        max_word_length = self.__word_list.shape[1]
        o_prob_pairs = {}
        for j in range(0, len(alphabet)):
            for k in range(j+1, len(alphabet)):
                letter1 = alphabet[j]
                letter2  = alphabet[k]
                print "Computing pair ", letter1, letter2
                for n in range(0 , max_word_length):
                    idx = np.nonzero( np.logical_or( self.__word_list[:,n] == letter2, self.__word_list[:,n] == letter1) )[0]
                    if len(idx) < 1:
                        continue
                    key = ''.join([letter1, letter2])
                    if not o_prob_pairs.has_key(key):
                        o_prob_pairs[key] = 0.0
                    else:
                        log_probs = np.array(self.__word_log_probs[idx].flatten())
                        log_prob_sum = self.utils.expTrick( log_probs.reshape([1,len(log_probs)]))
                        o_prob_pairs[key] += np.exp(log_prob_sum)
        prob_sum = np.sum(o_prob_pairs.values())
        for key in o_prob_pairs.keys():
            o_prob_pairs[key] /= prob_sum
        return o_prob_pairs

    def __wordListsToDict(self, i_list_words, i_vals):
        wi = [ [i_list_words[n], i_vals[n]] for n in range(0, len(i_vals))]
        return dict(wi)

    def __loadDictionary(self, i_load_from_file=True):
        filename = "entropy_dict.cPickle"
        if i_load_from_file and os.path.exists(filename):
            d = self.utils.loadPickle(filename)
            self.__word_list = np.array(d['word_list'])
            self.__word_log_probs = np.array(d['word_log_probs'])
            self.__letter_probs = d['letter_probs']
            self.__letter_pair_probs = d['letter_pair_probs']
            prob_sum = np.sum(self.__letter_pair_probs.values())
            for key in self.__letter_pair_probs.keys():
                self.__letter_pair_probs[key] /= prob_sum
            prob_sum = np.sum(self.__letter_probs.values())
            for key in self.__letter_probs:
                self.__letter_probs[key] /= prob_sum
            print "Loaded word list from file: ", self.__word_list.shape
            print "Loaded word_log_probs from file", self.__word_log_probs.shape
            print "letter_pair_probs len = ", len(self.__letter_pair_probs.values()), " sum = ", np.sum(self.__letter_pair_probs.values())
            print "letter probs len = ", len(self.__letter_probs.values()), " sum = ", np.sum(self.__letter_probs.values())
        else:
            d = WordDict("../dictionaries/nomon_dict.txt")
            self.__max_word_length = np.max(d.word_lengths)
            self.__word_log_probs = d.log_probs
            print "MAX WORD LENGTH = ", self.__max_word_length
            self.__word_list = []
            cnt_word = 1
            for word in d.words:
                if (cnt_word % 1000) == 0:
                    print " Word = ", cnt_word, " out of ", len(d.words),
                    print " Word list shape = ", np.array(self.__word_list).shape
                if word == '._':
                    cur_word = []
                    for n in range(0, self.__max_word_length):
                        cur_word.append('.')
                else:
                    max_repeat = np.int(np.ceil( float(self.__max_word_length) / float(len(word))))
                    cur_word = []
                    for n in range(0, max_repeat):
                        cur_word.extend(word)
                    cur_word = np.array(cur_word)[0:self.__max_word_length]
                if not(len(cur_word) == self.__max_word_length):
                    print "Word = ", cur_word, " len = ", len(cur_word), " should be ", self.__max_word_length
                    raise ValueError("Word length ERROR")
                self.__word_list.append(cur_word)
                cnt_word += 1
            self.__word_list = np.array(self.__word_list)
            self.__word_log_probs = np.array(self.__word_log_probs)
            self.__letter_probs = self.__marginalLetterLogProbs()
            self.__letter_pair_probs = self.__marginalJointLogProbs()

            print "SAVING TO FILE", filename, "Word list shape = ", self.__word_list.shape
            d = {'word_list': self.__word_list, 'word_log_probs': self.__word_log_probs,
                 'letter_probs': self.__letter_probs, 'letter_pair_probs':self.__letter_pair_probs}
            self.utils.savePickle(d, filename)

        #Make sure the word probs sum to one
        sum_probs = np.sum(np.exp(self.__word_log_probs))
        diff =  sum_probs - 1.0
        if diff > 1E-3:
            self.__word_log_probs -= np.log(sum_probs)
class AlphabetRandomiser(object):
    ########################################### Init functions
    def __init__(self):
        self.loadSimilarSoundConstraints()
        self.setParams( i_nchannels = 0)

    def loadSimilarSoundConstraints(self):
        constraints = [['a','h'],['q','k'],['m','n'],['b','d'],['a','i']]
        new_constraints = [[constraints[n][1], constraints[n][0]] for n in range(0,len(constraints))]
        constraints.extend(new_constraints)
        self.__constraints = np.array(constraints)

    def setParams(self, i_nchannels,  i_min_dist=3, i_max_repeat=2):
        self.__nchannels = i_nchannels
        self.__min_dist = i_min_dist
        self.__max_letter_repeat = i_max_repeat

    ########################################### Main functions

    def compute(self, i_alphabet, i_display=False):
        possible_sequences = [list(i_alphabet)]
        cur_group_idx = 0
        n_iterations = len(i_alphabet) * (self.__max_letter_repeat - 1)
        for n in range(0,  n_iterations):
            #With each iteration we choose a set of letters
            if i_display:
                print "****************************************************"
                print "Iteration", n, " out of ", n_iterations
            if np.array(possible_sequences).shape[1] >= (self.__max_letter_repeat*n_iterations):
                return (possible_sequences, n, n_iterations, True)
            new_sequences = []
            for cur_alphabet in possible_sequences:
                isGroupDone = self.__isGroupDone(i_alphabet, cur_alphabet, cur_group_idx)
                if isGroupDone:
                    new_sequences = self.__addLetter(cur_alphabet,'*', new_sequences, i_display=i_display)
                    continue
                #Get the alphabet without any spaces, as neighbours no not include spacings
                cur_alphabet_seq = self.getSequenceAlphabet(cur_alphabet)
                next_possible_letters = cur_alphabet_seq[0:(-self.__min_dist)]
                cur_group = self.getGrouping( cur_alphabet, self.__nchannels, i_display=False)[cur_group_idx]
                for letter in next_possible_letters:
                    if not self.validateLetter( letter, cur_group, cur_alphabet_seq ):
                        continue
                    new_sequences = self.__addLetter(cur_alphabet, letter, new_sequences, i_display=i_display)
            if len(new_sequences) < 1:
               return (self.__appendSpaces(possible_sequences, len(i_alphabet)), n, n_iterations, False)
            possible_sequences = list(new_sequences)
            cur_group_idx += 1
            if cur_group_idx >= self.__nchannels:
                cur_group_idx = 0
        return (self.__appendSpaces(possible_sequences, len(i_alphabet)), n, n_iterations, True)

    def getChannels(self):
        return self.__nchannels

    def getMinDist(self):
        return self.__min_dist

    ############################################# Display the alphabet in different ways
    def getGrouping( self, i_alphabet, i_nchannels, i_erase_spacers=True, i_display=False):
        alphabet = list(i_alphabet)
        modulus =  len(i_alphabet) % i_nchannels
        if modulus  > 0:
            n_spaces =  i_nchannels - modulus
            for n in range(0, n_spaces):
                alphabet.append('*')
        alphabet = np.array(alphabet)
        alphabet_groups = []
        for n in range(0, i_nchannels):
            group = alphabet[range(n, len(alphabet), i_nchannels)]
            if i_erase_spacers:
                group = [letter for letter in group if not letter=='*']
            alphabet_groups.append(group)
            if i_display:
                print group
        return alphabet_groups

    def getSequenceAlphabet(self, i_alphabet):
        #Return the alphabet in sequence without the spaces
        return np.array([letter for letter in i_alphabet if  not letter == '*'])

    def getAlphabetLength(self, i_alphabet):
        seq_alphabet = self.getSequenceAlphabet(i_alphabet)
        return len(seq_alphabet)

 ################################## Validation tests

    def validateLetter(self, i_letter, i_cur_group, i_sequence):
        #This is the main validation test to determine if a letter should be selected as part
        #of the random sequence
        if not self.validateLetterInCurrentGroup(i_cur_group, i_letter):
            return False
        if not self.validateSuccssiveGroupNeigbours(i_cur_group, i_letter):
            return False
        if not self.validateMaxLetterRepeat( i_sequence, i_letter ):
            return False
        if not self.validateSimilarSoundContraints( i_letter, i_sequence[-1]):
            return False
        if not self.validateNeigbourDistances(i_letter, i_sequence):
            return False
        return True

    def validateLetterInCurrentGroup(self, i_current_group, i_letter):
        #check if the letter appears in the right group - we always choose
        #letters from group 1 then group 2 etc
        #We therefore have to select a letter from the current group
        idx = np.nonzero( np.array(i_current_group) == i_letter )[0]
        if len(idx) == 0:
            return False
        return True

    def validateSimilarSoundContraints(self, i_letter, i_immediate_neigbour):
        #Make sure the other neighbour constraints based on how similar the sounds are, are valid
        #That is, sounds that are similar are not allowed to be neighbours
        #i_immediate_neigbour is a letter that has already been chosen.
        #i_letter can possibly be chosen - it can't be chosen if it sounds too similar to i_immediate_neigbour
        idx = np.nonzero(self.__constraints[:,0] == i_letter)[0]
        if len(idx) > 0:
            for neighbour in self.__constraints[idx,1]:
                valid_similar_sound = ( not(neighbour == i_immediate_neigbour ))
                if not valid_similar_sound:
                    return False
        return True

    def validateNeigbourDistances(self, i_letter, i_seq_alphabet):
        #Check if none of the current neighbours within max_dist have been
        #a neighbour within max_dist before
        idx = np.nonzero( i_seq_alphabet == i_letter)[0]
        start_index = idx - self.__min_dist
        end_index = idx + self.__min_dist + 1
        if start_index < 0:
            start_index = 0
        if end_index > len(i_seq_alphabet):
            end_index = len(i_seq_alphabet)
        prev_neighbours = i_seq_alphabet[range(start_index, end_index)]
        idx = np.nonzero( np.logical_not( prev_neighbours == i_letter) )[0]
        prev_neighbours = prev_neighbours[idx]
        cur_neighbours = i_seq_alphabet[-self.__min_dist:]
        for neighbour in cur_neighbours:
            valid_neighbours = len(np.nonzero(  prev_neighbours == neighbour )[0]) < 1
            if not valid_neighbours:
                return False
        return True

    def validateMaxLetterRepeat(self, i_sequence, i_letter):
        #Check how many times this letter already occurs, shoud be less than max num
        if len(np.nonzero( np.array(i_sequence) == i_letter)[0]) >= self.__max_letter_repeat:
            return False
        return True

    def validateSuccssiveGroupNeigbours(self, i_group, i_letter):
        #In any one group we do not allow the same letter to occur successively
        if i_letter == i_group[-1]:
            return False
        return True

    def validateFinalSequence(self, i_alphabet, i_display=True):
        print "========================================================"
        print "Validating the neighbours of group (rest can be verified manually)"
        grouping =  self.getGrouping(i_alphabet, self.__nchannels, i_erase_spacers=False,i_display=True)
        seq_alphabet = np.array(self.getSequenceAlphabet(i_alphabet))
        for letter in seq_alphabet:
            idx = np.nonzero(seq_alphabet == letter)[0]
            neighbours = []
            for n in idx:
                start_index = n - self.__min_dist
                if start_index < 0:
                    start_index = 0
                end_index = n + self.__min_dist + 1
                if end_index > len(seq_alphabet):
                    end_index = len(seq_alphabet)
                neighbours.extend(seq_alphabet[start_index:n])
                neighbours.extend(seq_alphabet[n+1:end_index])
            unique_neighbours = np.unique(neighbours)
            if i_display:
                print "Neigbours of all letters in the current alphabet:"
                print "Letter = ", letter, " Neigbours = ", neigbours
            if len(  unique_neighbours ) < len(   unique_neighbours ):
                print "INVALID SEQUENCE, BASED ON NEIGHBOURS!"
                return False
        print "VALID SEQUENCE"
        return True
    ###################################### Pivate Functions
    def __appendSpaces(self, i_sequences, i_original_alphabet_length):
        """Append spaces to all possible sequences until the correct length is reached"""
        o_sequences = list(i_sequences)
        for sequence in o_sequences:
            for n in range(len(sequence), self.__max_letter_repeat * i_original_alphabet_length):
                sequence.append('*')
        return o_sequences

    def __addLetter(self, i_sequence, i_letter, i_new_sequences, i_display=False):
        #All the validation tests were successful - add a possible sequence
        o_sequences = list(i_new_sequences)
        sequence = list(i_sequence)
        sequence.append(i_letter)
        o_sequences.append(sequence)
        if i_display:
            print "Adding new sequence ", ''.join(sequence)
        return o_sequences

    def __isGroupDone(self, i_original_alphabet, i_cur_alphabet, i_cur_group_idx):
        #A group can be done before the end of the sequence is reached (if there are spacers)
        original_group = self.getGrouping( i_original_alphabet, self.__nchannels, i_display=False)
        cur_group = self.getGrouping( i_cur_alphabet, self.__nchannels, i_display=False)[i_cur_group_idx]
        max_group_len = self.__max_letter_repeat*len(original_group[i_cur_group_idx])
        if len(cur_group) >= max_group_len:
            return True
        return False
    ####################################### Examples

    def fiveChannelExample(self, i_min_dist):
        self.setParams(i_nchannels=5, i_min_dist=i_min_dist)
        test_grouping = np.array([['a' , 'b' , 'c' ,'d', 'e','f'],
                                  ['*' , 'g' , 'h' , 'i' ,'j', 'k'],
                                  ['*' , 'l' , 'm' , 'n' ,'o', 'p'],
                                  ['q' , 'r' , 's' , 't' ,'u' ,'v'],
                                  ['w' , 'x' , 'y' , 'z' ,'_' ,'.']])
        cur_alphabet = np.array(test_grouping.transpose().flatten())
        print "ORIGINAL ALPHABET = ",  ''.join(cur_alphabet)
        print "GROUPING = "
        print test_grouping
        (o_sequences,  n, n_iterations, is_valid) = self.compute( cur_alphabet, i_display = True)
        print "OUTPUT:"
        print "------------"
        if is_valid:
            print "Valid sequences:"
            for s in o_sequences:
                self.validateFinalSequence(s, i_display=False)
        else:
            print "NO VALID SEQ"

class AlphabetSequenceComputer():
    def compute(self):
        disp_single_config =  False
        min_letter_separation = 4
        channels = 5
        if disp_single_config:
            randomiser = AlphabetRandomiser()
            randomiser.fiveChannelExample( min_letter_separation )
        else:
            test_alphabet = ['a','b','c','d','e','f','g','h','i','j','k','l','m','n','o','p','q','r','s','t','u','v','w','x','y','z','_', '.']
            randomiser = AlphabetOptimiseSpacingRandomiser()
            randomiser.setParams(i_nchannels=channels, i_min_dist=min_letter_separation)
            randomiser.compute(test_alphabet)

if __name__=="__main__":
    r = AlphabetSequenceComputer()
    r.compute()

max_stars_repo_name: nbonaker/ticker

max_issues_repo_name: nbonaker/ticker

max_forks_repo_name: nbonaker/ticker


text:
from math import ceil

import numpy as np
import os
import scipy
import cv2
import time
from keras import backend as K
from keras.callbacks import (CSVLogger, EarlyStopping, ModelCheckpoint,
                             ReduceLROnPlateau, TerminateOnNaN)
from keras.models import load_model
from keras.optimizers import Adam
from keras.preprocessing import image as kImage
from matplotlib import pyplot as plt
from skimage import io, transform
from keras.preprocessing import image as kImage

from draw_mask_on_image import draw_mask_on_image_array
from convert_frames_to_video import frames_to_video

from data_generator.data_augmentation_chain_constant_input_size import \
    DataAugmentationConstantInputSize
from data_generator.data_augmentation_chain_original_ssd import \
    SSDDataAugmentation
from data_generator.data_augmentation_chain_variable_input_size import \
    DataAugmentationVariableInputSize
from data_generator.object_detection_2d_data_generator import DataGenerator
from data_generator.object_detection_2d_misc_utils import \
    apply_inverse_transforms
from keras_layers.keras_layer_AnchorBoxes import AnchorBoxes
from keras_layers.keras_layer_DecodeDetections import DecodeDetections
from keras_layers.keras_layer_DecodeDetectionsFast import DecodeDetectionsFast
from keras_loss_function.keras_ssd_loss import SSDLoss
from models.keras_ssd7_new import build_model
from ssd_encoder_decoder.ssd_input_encoder import SSDInputEncoder
from ssd_encoder_decoder.ssd_output_decoder import (decode_detections,
                                                    decode_detections_fast)

#################################################################################################################
img_height = 300 # Height of the input images
img_width = 480 # Width of the input images
img_channels = 3 # Number of color channels of the input images
intensity_mean = 127.5 # Set this to your preference (maybe `None`). The current settings transform the input pixel values to the interval `[-1,1]`.
intensity_range = 127.5 # Set this to your preference (maybe `None`). The current settings transform the input pixel values to the interval `[-1,1]`.
n_classes = 1 # Number of positive classes
#scales = [0.08, 0.16, 0.32, 0.64, 0.96] # An explicit list of anchor box scaling factors. If this is passed, it will override `min_scale` and `max_scale`.
scales = [0.08, 0.16, 0.32, 0.64, 0.96, 1.05]
aspect_ratios = [1.0, 2.0, 0.5, 1.0/3.0] # The list of aspect ratios for the anchor boxes
two_boxes_for_ar1 = True # Whether or not you want to generate two anchor boxes for aspect ratio 1
steps = None # In case you'd like to set the step sizes for the anchor box grids manually; not recommended
offsets = None # In case you'd like to set the offsets for the anchor box grids manually; not recommended
clip_boxes = False # Whether or not to clip the anchor boxes to lie entirely within the image boundaries
variances = [1.0, 1.0, 1.0, 1.0] # The list of variances by which the encoded target coordinates are scaled
normalize_coords = True # Whether or not the model is supposed to use coordinates relative to the image size
##################################################################################################################

##################################################################################################################
# 1: Build the Keras model

def read_image(dir='', as_grey = False):
    imgs = os.listdir(dir)
    img_num = len(imgs)
    imgs.sort(key= lambda x:int(x[:-4]))

    if as_grey:
        pass
    else:
        train_images_all = []
        for i in range(img_num):
            #im = kImage.load_img(dir+ '/' + imgs[i])
            #im = kImage.img_to_array(im)
            im = io.imread(dir+ '/' + imgs[i])
            train_images_all.append(im)
            #print('data_dir:', dir + '/' + imgs[i])
    return np.asarray(train_images_all)

def evaluation(file_name):

    K.clear_session() # Clear previous models from memory.

    model = build_model(image_size=(img_height, img_width, img_channels),
                        n_classes=n_classes,
                        mode='training',
                        l2_regularization=0.0005,
                        scales=scales,
                        aspect_ratios_global=aspect_ratios,
                        aspect_ratios_per_layer=None,
                        two_boxes_for_ar1=two_boxes_for_ar1,
                        steps=steps,
                        offsets=offsets,
                        clip_boxes=clip_boxes,
                        variances=variances,
                        normalize_coords=normalize_coords,
                        subtract_mean=intensity_mean,
                        divide_by_stddev=intensity_range)

    # 2: Optional: Load some weights
    path= os.path.abspath('..')

    weight_saved_path = os.path.join(path, 'dataset', file_name, 'weight', 'ssd_segmentation_detection.h5')
    test_images_dir = os.path.join(path, 'dataset', file_name, 'test')
    mask_saved_dir = os.path.join(path, 'dataset', file_name, 'result')
    render_image_dir = os.path.join(path, 'dataset', file_name, 'rendering')
    demo_video_dir = os.path.join(path, 'dataset', file_name, 'demo', 'demo.avi')
    #original = os.path.join(path, 'dataset', file_name, 'original')

    model.load_weights(weight_saved_path, by_name=True)

    #######################################################################################################
    all_images = read_image(test_images_dir)

    #print('shape of batch_images:', batch_images.shape)

    #plt.figure(figsize=(20,12))
    plt.figure()
    plt.imshow(all_images[0])
    plt.show()

    img_num = all_images.shape[0]
    batch_size = 3
    batch_num = int(img_num/batch_size)

    id = 1

    start_time = time.time()

    for i in range(batch_num):

        mask_list = []

        batch_images = all_images[i*batch_size:(i+1)*batch_size]

        box_pred, mask_pred = model.predict(batch_images)
        # 4: Decode the raw prediction `y_pred`

        masks = mask_pred.reshape([mask_pred.shape[0], mask_pred.shape[1], mask_pred.shape[2]])
        #print(masks.shape)

        y_pred_decoded = decode_detections(box_pred,
                                        confidence_thresh=0.5,
                                        iou_threshold=0.2,
                                        top_k=200,
                                        normalize_coords=normalize_coords,
                                        img_height=img_height,
                                        img_width=img_width)

        #print('shape of y_pred_decoded:', y_pred_decoded)
        np.set_printoptions(precision=2, suppress=True, linewidth=90)
        #print("Predicted boxes:\n")
        #print('   class   conf xmin   ymin   xmax   ymax')

        #print(y_pred_decoded[i])

        #########################################################################################################
        images_roi = os.path.join(path, 'dataset', file_name, 'roi', 'roi.jpg')
        #images_roi= 'C:/Users/klickmal/Desktop/8.jpg'
        roi = io.imread(images_roi, as_grey = True)
        # 5: Draw the predicted boxes onto the image
        #print('shape of batch_images:', batch_images[i].shape)

        ones = np.ones((masks[0].shape[0], masks[0].shape[1]))
        zeros = np.zeros((masks[0].shape[0], masks[0].shape[1]))

        for j in range(len(y_pred_decoded)):

            masks[j] = np.where(roi, masks[j], roi)
            masks[j] = np.where(masks[j] > 0.1*ones, ones, zeros)

            #plt.figure(figsize=(20,12))
            #plt.rcParams['image.cmap'] = 'gray'
            #plt.imshow(masks[i])

            #current_axis = plt.gca()

            #colors = plt.cm.hsv(np.linspace(0, 1, n_classes+1)).tolist() # Set the colors for the bounding boxes
            classes = ['background', 'car'] # Just so we can print class names onto the image instead of IDs
            #masks[j] = cv2.cvtColor(masks[j], cv2.COLOR_GRAY2BGR)
            # Draw the predicted boxes in blue
            for box in y_pred_decoded[j]:
                xmin = box[-4]
                ymin = box[-3]
                xmax = box[-2]
                ymax = box[-1]
                #print(xmin)
                #color = colors[int(box[0])]
                label = '{}: {:.2f}'.format(classes[int(box[0])], box[1])

                #cv2.rectangle(masks[j],(int(xmin),int(ymin)),(int(xmax),int(ymax)),color,1)
                #cv2.putText(masks[j], label, (int(xmin),int(ymin-6)), cv2.FONT_HERSHEY_COMPLEX_SMALL,0.5, color)
                cv2.rectangle(batch_images[j],(int(xmin),int(ymin)),(int(xmax),int(ymax)), (55,255,155), 1)
                cv2.putText(batch_images[j], label, (int(xmin),int(ymin-6)), cv2.FONT_HERSHEY_COMPLEX_SMALL,0.5, (55,255,155), 1)

            mask_list.append(masks[j])

        mask_list = np.asarray(mask_list)
        #print(mask_list.shape)
        #print(all_images.shape)
        render_images = []
        for k in range(mask_list.shape[0]):
            render_image = draw_mask_on_image_array(batch_images[k], mask_list[k])
            render_images.append(render_image)
            #render_images = np.asarray(render_images)
        '''
        mask_array = np.array([0, 191, 255])
        mask_array = mask_array[np.newaxis, np.newaxis, np.newaxis, :]
        mask_array_tile = np.tile(mask_array, (batch_images.shape[0], batch_images.shape[1], batch_images.shape[2], 1))

        mask_list_new_axis = mask_list[:,:,:, np.newaxis]
        mask_list_tile = np.tile(mask_list_new_axis, (1, 1, 1, 3))
        print(mask_list_tile.shape)
        print(mask_list_tile[0][0])
        render_image = np.where(mask_list_tile, mask_array_tile + batch_images, batch_images)
        '''
        for i in range(mask_list.shape[0]):
            scipy.misc.imsave(mask_saved_dir + '/' + str(id) + '.jpg', mask_list[i])
            scipy.misc.imsave(render_image_dir + '/' + str(id) + '.jpg', render_images[i])

        id += 1

        del render_images
        del mask_list
        del render_image

    stop_time = time.time()
    print('time:', stop_time - start_time)

    frames_to_video(render_image_dir, demo_video_dir)

if __name__ == '__main__':

    #dic_dataset = {0: 'winterstreet', 1: 'highway'}
    dic_dataset = {0: 'cnitech_day_time'}
    for i in range(len(dic_dataset)):
        evaluation(dic_dataset[i])

reason: import numpy,import scipy


text:
import csv
import os

import numpy as np
from . import load_restore as lr

# A dataset is a collection of observations. Each observation is assigned
# an index. From that dataset, a subset such as a training, testing or
# validation set is defined by using a subset of indexes.

class DataSet:
    def __init__(self, dataset_name, hidden_nodes, input_columns=None, output_column=None):
        '''
		Initializes a dataset
		'''
        file_name = os.path.join("data", dataset_name + '.csv')
        # file_name = dataset_name + '.csv'
        with open(file_name, 'r') as f:
            data_iter = csv.reader(f, delimiter=',')
            data = [[float(attr) for attr in data] for data in data_iter]
        # If the input and output columns are not defined, then all columns
        # except the last are the input and the last column is the class value
        if not input_columns:
            input_columns = range(len(data[0]) - 1)
        if not output_column:
            # Index of class value in the data
            output_column = len(data[0]) - 1
        # If no classes are defined for the output column, these are 0 or 1
        # (it is a binary classification problem)
        self.classes = sorted(set(int(d[-1]) for d in data))
        self.number_examples = len(data)
        self.output_neurons = len(self.classes)
        self.network_length = len(hidden_nodes) + 2
        self.output_layer = len(hidden_nodes) + 1
        self.input_lenght = len(input_columns)
        self.examples = [None] * self.number_examples

        def to_one_hot(value, bits):
            one_hot = [0] * bits
            one_hot[value] = 1
            return one_hot

        for e in range(self.number_examples):
            self.examples[e] = Observation(e, self.network_length,
                                           int(data[e][output_column]),
                                           np.asarray(list(data[e][i] for i in input_columns)),
                                           to_one_hot(int(data[e][output_column]), self.output_neurons))

    def set_split(self, train_indexes, vali_indexes, test_indexes):
        '''
		Determines which instances make out the train, validation and test sets
		'''
        self.train_indexes = train_indexes
        self.vali_indexes = vali_indexes
        self.test_indexes = test_indexes
        self.num_train = len(train_indexes)
        self.num_vali = len(vali_indexes)
        self.num_test = len(test_indexes)

    def initialize_dictionary(self, output_condition):
        '''
		Initializes a dictionary that stores for each split point the indexes
		from the train and validation sets that exceed the threshold of that
		split-point.
		'''
        self.example_cond_dict = {}
        # If there is no separate validation set, then the post-pruning is performed on the train set
        # if self.num_vali > 0:
        #	self.dict_indexes = self.vali_indexes
        # else:
        self.dict_indexes = self.train_indexes
        self.update_dictionary([(output_condition[0], output_condition[1], output_condition[2])])
        # The target class value is stored under the same key as the output split point, but with layer index -1
        self.example_cond_dict[(-1, output_condition[1], output_condition[2])] = [e for e in self.dict_indexes if
                                                                                self.examples[e].class_value ==
                                                                                output_condition[1]]

    def update_dictionary(self, split_points):
        dictionary = self.example_cond_dict
        indexes = self.dict_indexes
        sps = set(split_points).difference(dictionary.keys())
        for sp in sps:
            dictionary[sp] = [e for e in indexes if self.examples[e].values[sp[0]][sp[1]] > sp[2]]

    def get_train_x_y(self):
        x = [self.examples[i].values[0] for i in self.train_indexes]
        y = [self.examples[i].y for i in self.train_indexes]
        return (x, y)

    def get_vali_x_y(self):
        x = [self.examples[i].values[0] for i in self.vali_indexes]
        y = [self.examples[i].y for i in self.vali_indexes]
        return (x, y)

    def get_test_x_y(self):
        x = [self.examples[i].values[0] for i in self.test_indexes]
        y = [self.examples[i].y for i in self.test_indexes]
        return (x, y)

    def set_train_act_values(self, layer, values=[]):
        for i in range(self.num_train):
            e = self.train_indexes[i]
            self.examples[e].set_act_values(layer, values[i])

    def set_vali_act_values(self, layer, values=[]):
        for i in range(self.num_vali):
            e = self.vali_indexes[i]
            self.examples[e].set_act_values(layer, values[i])

    def set_test_act_values(self, layer, values=[]):
        for i in range(self.num_test):
            e = self.test_indexes[i]
            self.examples[e].set_act_values(layer, values[i])

    def set_act_values(self, act_train, act_vali, act_test):
        for i in range(self.network_length - 1):
            if self.num_train > 0:
                self.set_train_act_values(i + 1, act_train[i])
            if self.num_vali > 0:
                self.set_vali_act_values(i + 1, act_vali[i])
            if self.num_test > 0:
                self.set_test_act_values(i + 1, act_test[i])

    def get_train_vali_obs(self):
        return [self.examples[e] for e in self.train_indexes] + [self.examples[e] for e in self.vali_indexes]

    def get_train_obs(self):
        return [self.examples[e] for e in self.train_indexes]

    def get_vali_obs(self):
        return [self.examples[e] for e in self.vali_indexes]

    def get_test_obs(self):
        return [self.examples[e] for e in self.test_indexes]

    # returns a list of all activation values for a node
    def get_act_all_examples(self, layer, node):
        vals = [e.values[layer][node] for e in self.examples if e.idx in self.train_indexes or e in self.vali_indexes]
        return vals

class Observation:
    def __init__(self, idx, network_length, class_value, x=[], y=[]):
        self.idx = idx
        self.values = [None] * network_length
        self.values[0] = x  # Layer with index 0 is the input
        self.y = y
        self.class_value = class_value
        self.orig_prediction = 0

    def set_nn_prediction(self, value):
        self.orig_prediction = value

    #  A 'split_point' has the form (layer, neuron, threshold)
    # The method returns True if >, False if <=
    def side_of_threshold(self, split_point):
        return self.values[split_point[0]][split_point[1]] > split_point[2]

    # A 'condition' has the form (layer, neuron, threshold, operator),
    # where operator is 0 for <=', 1 for '>' and 2 for '='
    def fulfills_cond(self, condition):
        if isinstance(condition, tuple):
            bigger = self.values[condition[0]][condition[1]] > condition[2]
            if condition[3]:
                if bigger:
                    return True
                else:
                    return False
            else:
                if bigger:
                    return False
                else:
                    return True
        else:
            return condition

    def fulfills_rule(self, rule):
        if isinstance(rule, list):
            if all((self.fulfills_cond(c) for c in rule)):
                return True
            else:
                return False
        else:
            return rule

    def fulfills_dnf(self, dnf):
        if isinstance(dnf, list):
            if any((self.fulfills_rule(r) for r in dnf)):
                return True
            else:
                return False
        else:
            return dnf

    def set_act_values(self, layer, values=[]):
        self.values[layer] = values

max_forks_repo_name: pietrobarbiero/deep-logic


text:
# Copyright 2020 - 2021 MONAI Consortium
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#     http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from typing import Any, Hashable, Mapping, Optional, Sequence, Union

import numpy as np

from monai.config import KeysCollection
from monai.transforms.smooth_field.array import RandSmoothFieldAdjustContrast, RandSmoothFieldAdjustIntensity
from monai.transforms.transform import MapTransform, RandomizableTransform, Transform
from monai.utils import InterpolateMode
from monai.utils.enums import TransformBackends

__all__ = ["RandSmoothFieldAdjustContrastd", "RandSmoothFieldAdjustIntensityd"]

class RandSmoothFieldAdjustContrastd(RandomizableTransform, MapTransform):
    """
    Dictionary version of RandSmoothFieldAdjustContrast. The field is randomized once per invocation by default so the
    same field is applied to every selected key.

    Args:
        keys: key names to apply the augment to
        spatial_size: size of input arrays, all arrays stated in `keys` must have same dimensions
        rand_size: size of the randomized field to start from
        padder: optional transform to add padding to the randomized field
        mode: interpolation mode to use when upsampling
        align_corners: if True align the corners when upsampling field
        prob: probability transform is applied
        gamma: (min, max) range for exponential field
        apply_same_field: if True, apply the same field to each key, otherwise randomize individually
    """

    backend = [TransformBackends.TORCH, TransformBackends.NUMPY]

    def __init__(
        self,
        keys: KeysCollection,
        spatial_size: Union[Sequence[int], int],
        rand_size: Union[Sequence[int], int],
        padder: Optional[Transform] = None,
        mode: Union[InterpolateMode, str] = InterpolateMode.AREA,
        align_corners: Optional[bool] = None,
        prob: float = 0.1,
        gamma: Union[Sequence[float], float] = (0.5, 4.5),
        apply_same_field: bool = True,
    ):
        RandomizableTransform.__init__(self, prob)
        MapTransform.__init__(self, keys)

        self.trans = RandSmoothFieldAdjustContrast(
            spatial_size=spatial_size,
            rand_size=rand_size,
            padder=padder,
            mode=mode,
            align_corners=align_corners,
            prob=1.0,
            gamma=gamma,
        )
        self.apply_same_field = apply_same_field

    def set_random_state(
        self, seed: Optional[int] = None, state: Optional[np.random.RandomState] = None
    ) -> "RandSmoothFieldAdjustContrastd":
        super().set_random_state(seed, state)
        self.trans.set_random_state(seed, state)
        return self

    def randomize(self, data: Optional[Any] = None) -> None:
        super().randomize(None)

        if self._do_transform:
            self.trans.randomize()

    def __call__(self, data: Mapping[Hashable, np.ndarray]) -> Mapping[Hashable, np.ndarray]:
        self.randomize()

        if not self._do_transform:
            return data

        d = dict(data)

        for key in self.key_iterator(d):
            if not self.apply_same_field:
                self.randomize()  # new field for every key

            d[key] = self.trans(d[key], False)

        return d

class RandSmoothFieldAdjustIntensityd(RandomizableTransform, MapTransform):
    """
    Dictionary version of RandSmoothFieldAdjustIntensity. The field is randomized once per invocation by default so
    the same field is applied to every selected key.

    Args:
        keys: key names to apply the augment to
        spatial_size: size of input arrays, all arrays stated in `keys` must have same dimensions
        rand_size: size of the randomized field to start from
        padder: optional transform to add padding to the randomized field
        mode: interpolation mode to use when upsampling
        align_corners: if True align the corners when upsampling field
        prob: probability transform is applied
        gamma: (min, max) range of intensity multipliers
        apply_same_field: if True, apply the same field to each key, otherwise randomize individually
    """

    backend = [TransformBackends.TORCH, TransformBackends.NUMPY]

    def __init__(
        self,
        keys: KeysCollection,
        spatial_size: Union[Sequence[int], int],
        rand_size: Union[Sequence[int], int],
        padder: Optional[Transform] = None,
        mode: Union[InterpolateMode, str] = InterpolateMode.AREA,
        align_corners: Optional[bool] = None,
        prob: float = 0.1,
        gamma: Union[Sequence[float], float] = (0.1, 1.0),
        apply_same_field: bool = True,
    ):
        RandomizableTransform.__init__(self, prob)
        MapTransform.__init__(self, keys)

        self.trans = RandSmoothFieldAdjustIntensity(
            spatial_size=spatial_size,
            rand_size=rand_size,
            padder=padder,
            mode=mode,
            align_corners=align_corners,
            prob=1.0,
            gamma=gamma,
        )
        self.apply_same_field = apply_same_field

    def set_random_state(
        self, seed: Optional[int] = None, state: Optional[np.random.RandomState] = None
    ) -> "RandSmoothFieldAdjustIntensityd":
        super().set_random_state(seed, state)
        self.trans.set_random_state(seed, state)
        return self

    def randomize(self, data: Optional[Any] = None) -> None:
        super().randomize(None)
        self.trans.randomize()

    def __call__(self, data: Mapping[Hashable, np.ndarray]) -> Mapping[Hashable, np.ndarray]:
        self.randomize()

        if not self._do_transform:
            return data

        d = dict(data)

        for key in self.key_iterator(d):
            if not self.apply_same_field:
                self.randomize()  # new field for every key

            d[key] = self.trans(d[key], False)

        return d