text:
# -*- coding: utf-8 -*-
import copy
import logging
from multiprocessing import Queue
import os
from statistics import stdev
import tempfile
import time

from freezegun import freeze_time
import h5py
from mantarray_desktop_app import FILE_WRITER_PERFOMANCE_LOGGING_NUM_CYCLES
from mantarray_desktop_app import FileWriterProcess
from mantarray_desktop_app import get_data_slice_within_timepoints
from mantarray_desktop_app import get_time_index_dataset_from_file
from mantarray_desktop_app import get_time_offset_dataset_from_file
from mantarray_desktop_app import get_tissue_dataset_from_file
from mantarray_desktop_app import MantarrayH5FileCreator
from mantarray_desktop_app import REF_INDEX_TO_24_WELL_INDEX
from mantarray_desktop_app import SERIAL_COMM_SENSOR_AXIS_LOOKUP_TABLE
from mantarray_desktop_app import UnrecognizedCommandFromMainToFileWriterError
from mantarray_file_manager import PLATE_BARCODE_UUID
from mantarray_file_manager import START_RECORDING_TIME_INDEX_UUID
import numpy as np
import pytest
from stdlib_utils import drain_queue
from stdlib_utils import InfiniteProcess
from stdlib_utils import invoke_process_run_and_check_errors

from ..fixtures import fixture_patch_print
from ..fixtures import QUEUE_CHECK_TIMEOUT_SECONDS
from ..fixtures_file_writer import fixture_four_board_file_writer_process
from ..fixtures_file_writer import fixture_runnable_four_board_file_writer_process
from ..fixtures_file_writer import fixture_running_four_board_file_writer_process
from ..fixtures_file_writer import GENERIC_BETA_1_START_RECORDING_COMMAND
from ..fixtures_file_writer import GENERIC_BETA_2_START_RECORDING_COMMAND
from ..fixtures_file_writer import GENERIC_NUM_CHANNELS_ENABLED
from ..fixtures_file_writer import GENERIC_NUM_SENSORS_ENABLED
from ..fixtures_file_writer import GENERIC_STOP_RECORDING_COMMAND
from ..fixtures_file_writer import WELL_DEF_24
from ..helpers import confirm_queue_is_eventually_empty
from ..helpers import confirm_queue_is_eventually_of_size
from ..helpers import is_queue_eventually_empty
from ..helpers import put_object_into_queue_and_raise_error_if_eventually_still_empty
from ..parsed_channel_data_packets import SIMPLE_BETA_1_CONSTRUCT_DATA_FROM_WELL_0
from ..parsed_channel_data_packets import SIMPLE_BETA_2_CONSTRUCT_DATA_FROM_ALL_WELLS

__fixtures__ = [
    fixture_patch_print,
    fixture_four_board_file_writer_process,
    fixture_runnable_four_board_file_writer_process,
    fixture_running_four_board_file_writer_process,
]

def test_get_data_slice_within_timepoints__raises_not_implemented_error_if_no_first_valid_index_found(
    patch_print,
):
    test_data = np.array([[1, 2, 3], [0, 0, 0]])
    min_timepoint = 4
    with pytest.raises(
        NotImplementedError,
        match=f"No timepoint >= the min timepoint of {min_timepoint} was found. All data passed to this function should contain at least one valid timepoint",
    ):
        get_data_slice_within_timepoints(test_data, min_timepoint)

def test_get_data_slice_within_timepoints__raises_not_implemented_error_if_no_last_valid_index_found(
    patch_print,
):
    test_data = np.array([[11, 12, 13], [0, 0, 0]])
    min_timepoint = 0
    max_timepoint = 10
    with pytest.raises(
        NotImplementedError,
        match=f"No timepoint <= the max timepoint of {max_timepoint} was found. All data passed to this function should contain at least one valid timepoint",
    ):
        get_data_slice_within_timepoints(test_data, min_timepoint, max_timepoint=max_timepoint)

def test_FileWriterProcess_super_is_called_during_init(mocker):
    error_queue = Queue()
    mocked_init = mocker.patch.object(InfiniteProcess, "__init__")
    FileWriterProcess((), Queue(), Queue(), error_queue)
    mocked_init.assert_called_once_with(error_queue, logging_level=logging.INFO)

def test_FileWriterProcess_setup_before_loop__calls_super(four_board_file_writer_process, mocker):
    spied_setup = mocker.spy(InfiniteProcess, "_setup_before_loop")

    fw_process = four_board_file_writer_process["fw_process"]
    invoke_process_run_and_check_errors(fw_process, perform_setup_before_loop=True)
    spied_setup.assert_called_once()

def test_FileWriterProcess_soft_stop_not_allowed_if_incoming_data_still_in_queue_for_board_0(
    four_board_file_writer_process,
):
    file_writer_process = four_board_file_writer_process["fw_process"]
    board_queues = four_board_file_writer_process["board_queues"]

    # The first communication will be processed, but if there is a second one in the queue then the soft stop should be disabled
    board_queues[0][0].put_nowait(SIMPLE_BETA_1_CONSTRUCT_DATA_FROM_WELL_0)
    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        SIMPLE_BETA_1_CONSTRUCT_DATA_FROM_WELL_0,
        board_queues[0][0],
    )

    confirm_queue_is_eventually_of_size(board_queues[0][0], 2)

    file_writer_process.soft_stop()
    invoke_process_run_and_check_errors(file_writer_process)
    assert file_writer_process.is_stopped() is False

    # Tanner (3/8/21): Prevent BrokenPipeErrors
    drain_queue(board_queues[0][0])

@pytest.mark.timeout(4)
def test_FileWriterProcess__raises_error_if_unrecognized_command_from_main(
    four_board_file_writer_process, mocker, patch_print
):

    file_writer_process = four_board_file_writer_process["fw_process"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]
    error_queue = four_board_file_writer_process["error_queue"]
    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        {"command": "do the hokey pokey"},
        from_main_queue,
    )
    file_writer_process.run(num_iterations=1)
    confirm_queue_is_eventually_of_size(error_queue, 1)

    raised_error, _ = error_queue.get(timeout=QUEUE_CHECK_TIMEOUT_SECONDS)
    assert isinstance(raised_error, UnrecognizedCommandFromMainToFileWriterError) is True
    err_str = str(raised_error)
    assert "do the hokey pokey" in err_str

def test_FileWriterProcess_soft_stop_not_allowed_if_command_from_main_still_in_queue(
    four_board_file_writer_process,
):
    file_writer_process = four_board_file_writer_process["fw_process"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]

    # The first communication will be processed, but if there is a second one in the queue then the soft stop should be disabled
    this_command = copy.deepcopy(GENERIC_BETA_1_START_RECORDING_COMMAND)
    this_command["active_well_indices"] = [1]
    from_main_queue.put_nowait(this_command)
    from_main_queue.put_nowait(copy.deepcopy(this_command))
    confirm_queue_is_eventually_of_size(from_main_queue, 2)
    file_writer_process.soft_stop()
    invoke_process_run_and_check_errors(file_writer_process)
    confirm_queue_is_eventually_of_size(from_main_queue, 1)
    assert file_writer_process.is_stopped() is False

    # Tanner (3/8/21): Prevent BrokenPipeErrors
    drain_queue(from_main_queue)

@pytest.mark.parametrize(
    "test_start_recording_command,test_description",
    [
        (GENERIC_BETA_1_START_RECORDING_COMMAND, "closes correctly with beta 1 files"),
        (GENERIC_BETA_2_START_RECORDING_COMMAND, "closes correctly with beta 2 files"),
    ],
)
def test_FileWriterProcess__close_all_files(
    test_start_recording_command, test_description, four_board_file_writer_process, mocker
):
    file_writer_process = four_board_file_writer_process["fw_process"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]

    if test_start_recording_command == GENERIC_BETA_2_START_RECORDING_COMMAND:
        file_writer_process.set_beta_2_mode()

    this_command = copy.deepcopy(test_start_recording_command)
    this_command["active_well_indices"] = [3, 18]

    put_object_into_queue_and_raise_error_if_eventually_still_empty(this_command, from_main_queue)
    invoke_process_run_and_check_errors(file_writer_process)
    open_files = file_writer_process._open_files  # pylint: disable=protected-access
    spied_file_3 = mocker.spy(open_files[0][3], "close")
    spied_file_18 = mocker.spy(open_files[0][18], "close")
    file_writer_process.close_all_files()
    assert spied_file_3.call_count == 1
    assert spied_file_18.call_count == 1

def test_FileWriterProcess__drain_all_queues__drains_all_queues_except_error_queue_and_returns__all_items(
    four_board_file_writer_process,
):
    expected = [[0, 1], [2, 3], [4, 5], [6, 7]]
    expected_error = "error"
    expected_from_main = "from_main"
    expected_to_main = "to_main"

    file_writer_process = four_board_file_writer_process["fw_process"]
    board_queues = four_board_file_writer_process["board_queues"]
    to_main_queue = four_board_file_writer_process["to_main_queue"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]
    error_queue = four_board_file_writer_process["error_queue"]
    for i, board in enumerate(board_queues):
        for j, iter_queue in enumerate(board):
            item = expected[i][j]
            put_object_into_queue_and_raise_error_if_eventually_still_empty(item, iter_queue)
    put_object_into_queue_and_raise_error_if_eventually_still_empty(expected_from_main, from_main_queue)
    put_object_into_queue_and_raise_error_if_eventually_still_empty(expected_to_main, to_main_queue)
    put_object_into_queue_and_raise_error_if_eventually_still_empty(expected_error, error_queue)

    actual = file_writer_process._drain_all_queues()  # pylint:disable=protected-access

    confirm_queue_is_eventually_of_size(error_queue, 1)
    actual_error = error_queue.get(timeout=QUEUE_CHECK_TIMEOUT_SECONDS)
    assert actual_error == expected_error

    for iter_queue_idx, iter_queue in enumerate(
        (
            board_queues[0][0],
            board_queues[0][1],
            board_queues[1][0],
            board_queues[2][0],
            board_queues[3][0],
            from_main_queue,
            to_main_queue,
        )
    ):
        assert (
            is_queue_eventually_empty(iter_queue, timeout_seconds=QUEUE_CHECK_TIMEOUT_SECONDS) is True
        ), f"Queue at index {iter_queue_idx} was not empty"

    assert actual["board_0"]["instrument_comm_to_file_writer"] == [expected[0][0]]
    assert actual["board_0"]["file_writer_to_data_analyzer"] == [expected[0][1]]
    assert actual["board_1"]["instrument_comm_to_file_writer"] == [expected[1][0]]
    assert actual["board_2"]["instrument_comm_to_file_writer"] == [expected[2][0]]
    assert actual["board_3"]["instrument_comm_to_file_writer"] == [expected[3][0]]
    assert actual["from_main_to_file_writer"] == [expected_from_main]
    assert actual["from_file_writer_to_main"] == [expected_to_main]

@pytest.mark.slow
def test_FileWriterProcess__logs_performance_metrics_after_appropriate_number_of_run_cycles(
    four_board_file_writer_process, mocker
):
    file_writer_process = four_board_file_writer_process["fw_process"]
    to_main_queue = four_board_file_writer_process["to_main_queue"]

    expected_iteration_dur = 0.001 * 10 ** 9
    expected_idle_time = expected_iteration_dur * FILE_WRITER_PERFOMANCE_LOGGING_NUM_CYCLES
    expected_start_timepoint = 0
    expected_stop_timepoint = 2 * expected_iteration_dur * FILE_WRITER_PERFOMANCE_LOGGING_NUM_CYCLES
    expected_latest_percent_use = 100 * (
        1 - expected_idle_time / (expected_stop_timepoint - expected_start_timepoint)
    )
    expected_percent_use_values = [27.4, 42.8, expected_latest_percent_use]
    expected_longest_iterations = [
        expected_iteration_dur for _ in range(file_writer_process.num_longest_iterations)
    ]

    perf_counter_vals = []
    for _ in range(FILE_WRITER_PERFOMANCE_LOGGING_NUM_CYCLES - 1):
        perf_counter_vals.append(0)
        perf_counter_vals.append(expected_iteration_dur)
    perf_counter_vals.append(0)
    perf_counter_vals.append(expected_stop_timepoint)
    perf_counter_vals.append(0)
    mocker.patch.object(time, "perf_counter_ns", autospec=True, side_effect=perf_counter_vals)

    file_writer_process._idle_iteration_time_ns = expected_iteration_dur  # pylint: disable=protected-access
    file_writer_process._minimum_iteration_duration_seconds = (  # pylint: disable=protected-access
        2 * expected_iteration_dur / (10 ** 9)
    )
    file_writer_process._start_timepoint_of_last_performance_measurement = (  # pylint: disable=protected-access
        expected_start_timepoint
    )
    file_writer_process._percent_use_values = expected_percent_use_values[  # pylint: disable=protected-access
        :-1
    ]

    invoke_process_run_and_check_errors(
        file_writer_process, num_iterations=FILE_WRITER_PERFOMANCE_LOGGING_NUM_CYCLES
    )
    confirm_queue_is_eventually_of_size(to_main_queue, 1)
    actual = to_main_queue.get(timeout=QUEUE_CHECK_TIMEOUT_SECONDS)
    actual = actual["message"]

    assert actual["communication_type"] == "performance_metrics"
    assert actual["percent_use"] == expected_latest_percent_use
    assert actual["percent_use_metrics"] == {
        "max": max(expected_percent_use_values),
        "min": min(expected_percent_use_values),
        "stdev": round(stdev(expected_percent_use_values), 6),
        "mean": round(sum(expected_percent_use_values) / len(expected_percent_use_values), 6),
    }
    num_longest_iterations = file_writer_process.num_longest_iterations
    assert actual["longest_iterations"] == expected_longest_iterations[-num_longest_iterations:]
    assert "idle_iteration_time_ns" not in actual
    assert "start_timepoint_of_measurements" not in actual

@pytest.mark.parametrize(
    "test_data_packet,test_description",
    [
        (SIMPLE_BETA_1_CONSTRUCT_DATA_FROM_WELL_0, "does not log recording metrics with beta 1 data"),
        (SIMPLE_BETA_2_CONSTRUCT_DATA_FROM_ALL_WELLS, "does not log recording metrics with beta 2 data"),
    ],
)
def test_FileWriterProcess__does_not_include_recording_metrics_in_performance_metrics_when_not_recording(
    test_data_packet, test_description, four_board_file_writer_process, mocker
):
    file_writer_process = four_board_file_writer_process["fw_process"]
    to_main_queue = four_board_file_writer_process["to_main_queue"]
    incoming_data_queue = four_board_file_writer_process["board_queues"][0][0]

    if test_data_packet == SIMPLE_BETA_2_CONSTRUCT_DATA_FROM_ALL_WELLS:
        file_writer_process.set_beta_2_mode()

    # add data packets
    num_packets_to_send = 3  # send arbitrary number of packets
    for _ in range(num_packets_to_send):
        incoming_data_queue.put_nowait(test_data_packet)
    confirm_queue_is_eventually_of_size(incoming_data_queue, num_packets_to_send)
    # set to 0 to speed up test
    file_writer_process._minimum_iteration_duration_seconds = 0  # pylint: disable=protected-access
    # get performance metrics dict
    invoke_process_run_and_check_errors(
        file_writer_process,
        num_iterations=FILE_WRITER_PERFOMANCE_LOGGING_NUM_CYCLES,
        perform_setup_before_loop=True,
    )
    confirm_queue_is_eventually_of_size(to_main_queue, 1)
    actual = to_main_queue.get(timeout=QUEUE_CHECK_TIMEOUT_SECONDS)
    actual = actual["message"]
    # make sure recording metrics not present
    assert "num_recorded_data_points_metrics" not in actual
    assert "recording_duration_metrics" not in actual

    # Tanner (6/1/21): avoid BrokenPipeErrors
    drain_queue(four_board_file_writer_process["board_queues"][0][1])

@pytest.mark.slow
@pytest.mark.timeout(200)
def test_FileWriterProcess__does_not_log_percent_use_metrics_in_first_logging_cycle(
    four_board_file_writer_process,
):
    file_writer_process = four_board_file_writer_process["fw_process"]
    to_main_queue = four_board_file_writer_process["to_main_queue"]

    # set to 0 to speed up test
    file_writer_process._minimum_iteration_duration_seconds = 0  # pylint: disable=protected-access

    invoke_process_run_and_check_errors(
        file_writer_process,
        num_iterations=FILE_WRITER_PERFOMANCE_LOGGING_NUM_CYCLES,
        perform_setup_before_loop=True,
    )
    confirm_queue_is_eventually_of_size(to_main_queue, 1)

    actual = to_main_queue.get(timeout=QUEUE_CHECK_TIMEOUT_SECONDS)
    actual = actual["message"]
    assert "percent_use_metrics" not in actual

@pytest.mark.slow
@pytest.mark.parametrize(
    "test_beta_version,test_description",
    [
        (1, "logs correctly with beta 1 data"),
        (2, "logs correctly with beta 2 data"),
    ],
)
def test_FileWriterProcess__logs_metrics_of_data_recording_correctly(
    test_beta_version, test_description, four_board_file_writer_process, mocker
):
    file_writer_process = four_board_file_writer_process["fw_process"]
    board_queues = four_board_file_writer_process["board_queues"]
    to_main_queue = four_board_file_writer_process["to_main_queue"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]

    # set to 0 to speed up test
    file_writer_process._minimum_iteration_duration_seconds = 0  # pylint: disable=protected-access

    num_packets_to_send = 5  # arbitrary value
    if test_beta_version == 1:
        start_recording_command = copy.deepcopy(GENERIC_BETA_1_START_RECORDING_COMMAND)
        data_packet = copy.deepcopy(SIMPLE_BETA_1_CONSTRUCT_DATA_FROM_WELL_0)
        num_points_per_packet = data_packet["data"].shape[1]
    else:
        file_writer_process.set_beta_2_mode()
        start_recording_command = copy.deepcopy(GENERIC_BETA_2_START_RECORDING_COMMAND)
        data_packet = copy.deepcopy(SIMPLE_BETA_2_CONSTRUCT_DATA_FROM_ALL_WELLS)
        num_points_per_packet = data_packet["time_indices"].shape[0]

    start_recording_command["metadata_to_copy_onto_main_file_attributes"][START_RECORDING_TIME_INDEX_UUID] = 0
    put_object_into_queue_and_raise_error_if_eventually_still_empty(start_recording_command, from_main_queue)
    invoke_process_run_and_check_errors(file_writer_process, perform_setup_before_loop=True)
    # Tanner (9/10/20): remove start_recording confirmation
    to_main_queue.get(timeout=QUEUE_CHECK_TIMEOUT_SECONDS)

    num_points_list = [num_points_per_packet] * num_packets_to_send
    for _ in range(num_packets_to_send):
        board_queues[0][0].put_nowait(data_packet)

    confirm_queue_is_eventually_of_size(board_queues[0][0], num_packets_to_send)
    expected_recording_durations = list(range(num_packets_to_send))
    perf_counter_vals = [
        0 if i % 2 == 0 else expected_recording_durations[i // 2] for i in range(num_packets_to_send * 2)
    ]
    mocker.patch.object(time, "perf_counter", autospec=True, side_effect=perf_counter_vals)

    invoke_process_run_and_check_errors(
        file_writer_process, num_iterations=FILE_WRITER_PERFOMANCE_LOGGING_NUM_CYCLES
    )
    confirm_queue_is_eventually_empty(board_queues[0][0])

    actual = to_main_queue.get(timeout=QUEUE_CHECK_TIMEOUT_SECONDS)
    actual = actual["message"]
    assert (
        "num_recorded_data_points_metrics" in actual
    ), f"Message did not contain key: 'num_recorded_data_points_metrics', Full message dict: {actual}"
    assert actual["num_recorded_data_points_metrics"] == {
        "max": max(num_points_list),
        "min": min(num_points_list),
        "stdev": round(stdev(num_points_list), 6),
        "mean": round(sum(num_points_list) / len(num_points_list), 6),
    }
    assert actual["recording_duration_metrics"] == {
        "max": max(expected_recording_durations),
        "min": min(expected_recording_durations),
        "stdev": round(stdev(expected_recording_durations), 6),
        "mean": round(sum(expected_recording_durations) / len(expected_recording_durations), 6),
    }

    # Tanner (3/8/21): Prevent BrokenPipeErrors
    drain_queue(board_queues[0][1])

def test_FileWriterProcess_teardown_after_loop__sets_teardown_complete_event(
    four_board_file_writer_process,
    mocker,
):
    fw_process = four_board_file_writer_process["fw_process"]

    fw_process.soft_stop()
    fw_process.run(perform_setup_before_loop=False, num_iterations=1)

    assert fw_process.is_teardown_complete() is True

@freeze_time("2020-07-20 15:09:22.654321")
def test_FileWriterProcess_teardown_after_loop__puts_teardown_log_message_into_queue(
    four_board_file_writer_process,
):
    fw_process = four_board_file_writer_process["fw_process"]
    to_main_queue = four_board_file_writer_process["to_main_queue"]

    fw_process.soft_stop()
    fw_process.run(perform_setup_before_loop=False, num_iterations=1)
    confirm_queue_is_eventually_of_size(to_main_queue, 1)

    actual = to_main_queue.get(timeout=QUEUE_CHECK_TIMEOUT_SECONDS)
    assert actual["message"] == "File Writer Process beginning teardown at 2020-07-20 15:09:22.654321"

def test_FileWriterProcess_teardown_after_loop__does_not_call_close_all_files__when_not_recording(
    four_board_file_writer_process, mocker
):
    fw_process = four_board_file_writer_process["fw_process"]

    spied_close_all_files = mocker.spy(fw_process, "close_all_files")

    fw_process.soft_stop()
    fw_process.run(perform_setup_before_loop=False, num_iterations=1)

    spied_close_all_files.assert_not_called()

@pytest.mark.parametrize(
    "test_start_recording_command,test_description",
    [
        (GENERIC_BETA_1_START_RECORDING_COMMAND, "calls close with beta 1 files"),
        (GENERIC_BETA_2_START_RECORDING_COMMAND, "calls close with beta 2 files"),
    ],
)
def test_FileWriterProcess_teardown_after_loop__calls_close_all_files__when_still_recording(
    test_start_recording_command, test_description, four_board_file_writer_process, mocker
):
    fw_process = four_board_file_writer_process["fw_process"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]

    if test_start_recording_command == GENERIC_BETA_2_START_RECORDING_COMMAND:
        fw_process.set_beta_2_mode()

    spied_close_all_files = mocker.spy(fw_process, "close_all_files")
    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        test_start_recording_command, from_main_queue
    )

    fw_process.soft_stop()
    fw_process.run(perform_setup_before_loop=False, num_iterations=1)

    spied_close_all_files.assert_called_once()

@pytest.mark.parametrize(
    "test_start_recording_command,test_description",
    [
        (GENERIC_BETA_1_START_RECORDING_COMMAND, "calls close with beta 1 files"),
        (GENERIC_BETA_2_START_RECORDING_COMMAND, "calls close with beta 2 files"),
    ],
)
def test_FileWriterProcess_hard_stop__calls_close_all_files__when_still_recording(
    test_start_recording_command, test_description, four_board_file_writer_process, mocker
):
    fw_process = four_board_file_writer_process["fw_process"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]

    if test_start_recording_command == GENERIC_BETA_2_START_RECORDING_COMMAND:
        fw_process.set_beta_2_mode()

    spied_close_all_files = mocker.spy(fw_process, "close_all_files")
    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        test_start_recording_command, from_main_queue
    )
    fw_process.run(
        perform_setup_before_loop=False,
        num_iterations=1,
        perform_teardown_after_loop=False,
    )
    assert spied_close_all_files.call_count == 0  # confirm precondition
    fw_process.hard_stop()

    spied_close_all_files.assert_called_once()

def test_FileWriterProcess_hard_stop__closes_all_beta_1_files_after_stop_recording_before_all_files_are_finalized__and_files_can_be_opened_after_process_stops(
    four_board_file_writer_process, mocker
):
    expected_timestamp = "2020_02_09_190935"
    expected_barcode = GENERIC_BETA_1_START_RECORDING_COMMAND["metadata_to_copy_onto_main_file_attributes"][
        PLATE_BARCODE_UUID
    ]

    fw_process = four_board_file_writer_process["fw_process"]
    board_queues = four_board_file_writer_process["board_queues"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]
    tmp_dir = four_board_file_writer_process["file_dir"]

    spied_close_all_files = mocker.spy(fw_process, "close_all_files")

    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        GENERIC_BETA_1_START_RECORDING_COMMAND, from_main_queue
    )
    invoke_process_run_and_check_errors(fw_process)

    # fill files with data
    start_timepoint = GENERIC_BETA_1_START_RECORDING_COMMAND["timepoint_to_begin_recording_at"]
    test_data = np.array([[start_timepoint], [0]], dtype=np.int32)
    for i in range(24):
        tissue_data_packet = {
            "well_index": i,
            "is_reference_sensor": False,
            "data": test_data,
        }
        board_queues[0][0].put_nowait(tissue_data_packet)
    for i in range(6):
        ref_data_packet = {
            "reference_for_wells": REF_INDEX_TO_24_WELL_INDEX[i],
            "is_reference_sensor": True,
            "data": test_data,
        }
        board_queues[0][0].put_nowait(ref_data_packet)
    confirm_queue_is_eventually_of_size(board_queues[0][0], 30)

    # set to 0 to speed up test
    fw_process._minimum_iteration_duration_seconds = 0  # pylint: disable=protected-access
    invoke_process_run_and_check_errors(fw_process, num_iterations=30)
    confirm_queue_is_eventually_empty(board_queues[0][0])

    stop_recording_command = copy.deepcopy(GENERIC_STOP_RECORDING_COMMAND)
    put_object_into_queue_and_raise_error_if_eventually_still_empty(stop_recording_command, from_main_queue)
    invoke_process_run_and_check_errors(fw_process)

    assert spied_close_all_files.call_count == 0  # confirm precondition
    fw_process.hard_stop()
    spied_close_all_files.assert_called_once()

    # confirm files can be opened and files contains at least one piece of metadata and the correct tissue data
    for row_idx in range(4):
        for col_idx in range(6):
            well_name = WELL_DEF_24.get_well_name_from_row_and_column(row_idx, col_idx)
            with h5py.File(
                os.path.join(
                    tmp_dir,
                    f"{expected_barcode}__{expected_timestamp}",
                    f"{expected_barcode}__{expected_timestamp}__{well_name}.h5",
                ),
                "r",
            ) as this_file:
                assert (
                    str(START_RECORDING_TIME_INDEX_UUID) in this_file.attrs
                ), f"START_RECORDING_TIME_INDEX_UUID missing for Well {well_name}"
                assert get_tissue_dataset_from_file(this_file).shape == (
                    test_data.shape[1],
                ), f"Incorrect tissue data shape for Well {well_name}"

def test_FileWriterProcess_hard_stop__closes_all_beta_2_files_after_stop_recording_before_all_files_are_finalized__and_files_can_be_opened_after_process_stops(
    four_board_file_writer_process, mocker
):
    expected_timestamp = "2020_02_09_190359"
    expected_barcode = GENERIC_BETA_2_START_RECORDING_COMMAND["metadata_to_copy_onto_main_file_attributes"][
        PLATE_BARCODE_UUID
    ]

    fw_process = four_board_file_writer_process["fw_process"]
    fw_process.set_beta_2_mode()
    board_queues = four_board_file_writer_process["board_queues"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]
    tmp_dir = four_board_file_writer_process["file_dir"]

    spied_close_all_files = mocker.spy(fw_process, "close_all_files")

    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        GENERIC_BETA_2_START_RECORDING_COMMAND, from_main_queue
    )
    invoke_process_run_and_check_errors(fw_process)

    # fill files with data
    test_num_data_points = 50
    start_timepoint = GENERIC_BETA_2_START_RECORDING_COMMAND["timepoint_to_begin_recording_at"]
    test_data = np.zeros(test_num_data_points, dtype=np.int16)
    data_packet = {
        "data_type": "magnetometer",
        "time_indices": np.arange(start_timepoint, start_timepoint + test_num_data_points, dtype=np.uint64),
        "is_first_packet_of_stream": False,
    }
    for well_idx in range(24):
        channel_dict = {
            "time_offsets": np.zeros((GENERIC_NUM_SENSORS_ENABLED, test_num_data_points), dtype=np.uint16),
            SERIAL_COMM_SENSOR_AXIS_LOOKUP_TABLE["A"]["X"]: test_data,
            SERIAL_COMM_SENSOR_AXIS_LOOKUP_TABLE["C"]["Z"]: test_data,
        }
        data_packet[well_idx] = channel_dict
    board_queues[0][0].put_nowait(data_packet)
    confirm_queue_is_eventually_of_size(board_queues[0][0], 1)

    invoke_process_run_and_check_errors(fw_process)
    confirm_queue_is_eventually_empty(board_queues[0][0])

    stop_recording_command = copy.deepcopy(GENERIC_STOP_RECORDING_COMMAND)
    put_object_into_queue_and_raise_error_if_eventually_still_empty(stop_recording_command, from_main_queue)
    invoke_process_run_and_check_errors(fw_process)

    assert spied_close_all_files.call_count == 0  # confirm precondition
    fw_process.hard_stop()
    spied_close_all_files.assert_called_once()

    # confirm files can be opened and files contains at least one piece of metadata and the correct tissue data
    for row_idx in range(4):
        for col_idx in range(6):
            well_name = WELL_DEF_24.get_well_name_from_row_and_column(row_idx, col_idx)
            with h5py.File(
                os.path.join(
                    tmp_dir,
                    f"{expected_barcode}__{expected_timestamp}",
                    f"{expected_barcode}__{expected_timestamp}__{well_name}.h5",
                ),
                "r",
            ) as this_file:
                assert (
                    str(START_RECORDING_TIME_INDEX_UUID) in this_file.attrs
                ), f"START_RECORDING_TIME_INDEX_UUID missing for Well {well_name}"
                assert get_time_index_dataset_from_file(this_file).shape == (
                    test_num_data_points,
                ), f"Incorrect time index data shape for Well {well_name}"
                assert get_time_offset_dataset_from_file(this_file).shape == (
                    GENERIC_NUM_SENSORS_ENABLED,
                    test_num_data_points,
                ), f"Incorrect time offset data shape for Well {well_name}"
                assert get_tissue_dataset_from_file(this_file).shape == (
                    GENERIC_NUM_CHANNELS_ENABLED,
                    test_num_data_points,
                ), f"Incorrect tissue data shape for Well {well_name}"

def test_FileWriterProcess__ignores_commands_from_main_while_finalizing_beta_1_files_after_stop_recording(
    four_board_file_writer_process, mocker
):
    fw_process = four_board_file_writer_process["fw_process"]
    board_queues = four_board_file_writer_process["board_queues"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]

    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        GENERIC_BETA_1_START_RECORDING_COMMAND, from_main_queue
    )
    invoke_process_run_and_check_errors(fw_process)

    # fill files with data
    start_timepoint = GENERIC_BETA_1_START_RECORDING_COMMAND["timepoint_to_begin_recording_at"]
    first_data = np.array([[start_timepoint], [0]], dtype=np.int32)
    for i in range(24):
        tissue_data_packet = {
            "well_index": i,
            "is_reference_sensor": False,
            "data": first_data,
        }
        board_queues[0][0].put_nowait(tissue_data_packet)
    for i in range(6):
        ref_data_packet = {
            "reference_for_wells": REF_INDEX_TO_24_WELL_INDEX[i],
            "is_reference_sensor": True,
            "data": first_data,
        }
        board_queues[0][0].put_nowait(ref_data_packet)
    confirm_queue_is_eventually_of_size(board_queues[0][0], 30)

    # set to 0 to speed up test
    fw_process._minimum_iteration_duration_seconds = 0  # pylint: disable=protected-access
    invoke_process_run_and_check_errors(fw_process, num_iterations=30)
    confirm_queue_is_eventually_empty(board_queues[0][0])

    stop_recording_command = copy.deepcopy(GENERIC_STOP_RECORDING_COMMAND)
    put_object_into_queue_and_raise_error_if_eventually_still_empty(stop_recording_command, from_main_queue)
    invoke_process_run_and_check_errors(fw_process)

    # check that command is ignored # Tanner (1/12/21): no particular reason this command needs to be update_directory, but it's easy to test if this gets processed
    expected_new_dir = "dummy_dir"
    update_dir_command = {
        "command": "update_directory",
        "new_directory": expected_new_dir,
    }
    put_object_into_queue_and_raise_error_if_eventually_still_empty(update_dir_command, from_main_queue)
    invoke_process_run_and_check_errors(fw_process)
    confirm_queue_is_eventually_of_size(from_main_queue, 1)

    # add data past stop point so files will be finalized
    stop_timepoint = GENERIC_STOP_RECORDING_COMMAND["timepoint_to_stop_recording_at"]
    last_data = np.array([[stop_timepoint], [0]], dtype=np.int32)
    for i in range(24):
        final_tissue_data_packet = {
            "well_index": i,
            "is_reference_sensor": False,
            "data": last_data,
        }
        board_queues[0][0].put_nowait(final_tissue_data_packet)
    for i in range(6):
        final_ref_data_packet = {
            "reference_for_wells": REF_INDEX_TO_24_WELL_INDEX[i],
            "is_reference_sensor": True,
            "data": last_data,
        }
        board_queues[0][0].put_nowait(final_ref_data_packet)
    confirm_queue_is_eventually_of_size(board_queues[0][0], 30)

    invoke_process_run_and_check_errors(fw_process, num_iterations=30)
    confirm_queue_is_eventually_empty(board_queues[0][0])
    # check command is still ignored
    confirm_queue_is_eventually_of_size(from_main_queue, 1)

    # now all files should be finalized, confirm command is now processed
    invoke_process_run_and_check_errors(fw_process)
    confirm_queue_is_eventually_empty(from_main_queue)
    assert fw_process.get_file_directory() == expected_new_dir

    # Tanner (3/8/21): Prevent BrokenPipeErrors
    drain_queue(board_queues[0][1])

def test_FileWriterProcess__ignores_commands_from_main_while_finalizing_beta_2_files_after_stop_recording(
    four_board_file_writer_process, mocker
):
    fw_process = four_board_file_writer_process["fw_process"]
    fw_process.set_beta_2_mode()
    board_queues = four_board_file_writer_process["board_queues"]
    from_main_queue = four_board_file_writer_process["from_main_queue"]

    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        GENERIC_BETA_2_START_RECORDING_COMMAND, from_main_queue
    )
    invoke_process_run_and_check_errors(fw_process)

    # fill files with data
    num_data_points = 100
    start_timepoint = GENERIC_BETA_2_START_RECORDING_COMMAND["timepoint_to_begin_recording_at"]
    data_packet = {
        "data_type": "magnetometer",
        "time_indices": np.arange(start_timepoint, start_timepoint + num_data_points, dtype=np.uint64),
        "is_first_packet_of_stream": False,
    }
    for well_idx in range(24):
        channel_dict = {
            "time_offsets": np.zeros((GENERIC_NUM_SENSORS_ENABLED, num_data_points), dtype=np.uint16),
            SERIAL_COMM_SENSOR_AXIS_LOOKUP_TABLE["A"]["X"]: np.zeros(num_data_points, dtype=np.int16),
            SERIAL_COMM_SENSOR_AXIS_LOOKUP_TABLE["C"]["Z"]: np.zeros(num_data_points, dtype=np.int16),
        }
        data_packet[well_idx] = channel_dict
    board_queues[0][0].put_nowait(data_packet)
    confirm_queue_is_eventually_of_size(board_queues[0][0], 1)
    invoke_process_run_and_check_errors(fw_process)
    confirm_queue_is_eventually_empty(board_queues[0][0])

    stop_recording_command = copy.deepcopy(GENERIC_STOP_RECORDING_COMMAND)
    put_object_into_queue_and_raise_error_if_eventually_still_empty(stop_recording_command, from_main_queue)
    invoke_process_run_and_check_errors(fw_process)

    # check that command is ignored # Tanner (1/12/21): no particular reason this command needs to be update_directory, but it's easy to test if this gets processed
    expected_new_dir = "dummy_dir"
    update_dir_command = {
        "command": "update_directory",
        "new_directory": expected_new_dir,
    }
    put_object_into_queue_and_raise_error_if_eventually_still_empty(update_dir_command, from_main_queue)
    invoke_process_run_and_check_errors(fw_process)
    confirm_queue_is_eventually_of_size(from_main_queue, 1)

    # add data past stop point so files will be finalized
    stop_timepoint = GENERIC_STOP_RECORDING_COMMAND["timepoint_to_stop_recording_at"]
    final_data_packet = copy.deepcopy(data_packet)
    final_data_packet["time_indices"] = np.arange(
        stop_timepoint, stop_timepoint + num_data_points, dtype=np.uint64
    )
    board_queues[0][0].put_nowait(final_data_packet)
    confirm_queue_is_eventually_of_size(board_queues[0][0], 1)
    invoke_process_run_and_check_errors(fw_process)
    confirm_queue_is_eventually_empty(board_queues[0][0])
    # check command is still ignored
    confirm_queue_is_eventually_of_size(from_main_queue, 1)

    # now all files should be finalized, confirm command is now processed
    invoke_process_run_and_check_errors(fw_process)
    confirm_queue_is_eventually_empty(from_main_queue)
    assert fw_process.get_file_directory() == expected_new_dir

    # Tanner (3/8/21): Prevent BrokenPipeErrors
    drain_queue(board_queues[0][1])

@pytest.mark.slow
@pytest.mark.timeout(10)
@pytest.mark.parametrize(
    "test_start_recording_command,test_description",
    [
        (GENERIC_BETA_1_START_RECORDING_COMMAND, "tears down correctly with beta 1 files"),
        (GENERIC_BETA_2_START_RECORDING_COMMAND, "tears down correctly with beta 2 files"),
    ],
)
def test_FileWriterProcess_teardown_after_loop__can_teardown_process_while_recording__and_log_stop_recording_message(
    test_start_recording_command, test_description, running_four_board_file_writer_process, mocker
):
    fw_process = running_four_board_file_writer_process["fw_process"]
    to_main_queue = running_four_board_file_writer_process["to_main_queue"]
    from_main_queue = running_four_board_file_writer_process["from_main_queue"]

    if test_start_recording_command == GENERIC_BETA_2_START_RECORDING_COMMAND:
        fw_process.set_beta_2_mode()

    put_object_into_queue_and_raise_error_if_eventually_still_empty(
        test_start_recording_command, from_main_queue
    )

    fw_process.soft_stop()
    fw_process.join()

    queue_items = drain_queue(to_main_queue)

    actual = queue_items[-1]
    assert (
        actual["message"]
        == "Data is still be written to file. Stopping recording and closing files to complete teardown"
    )

def test_MantarrayH5FileCreator__sets_file_name_and_userblock_size():
    with tempfile.TemporaryDirectory() as tmp_dir:
        expected_filename = os.path.join(tmp_dir, "myfile.h5")
        test_file = MantarrayH5FileCreator(expected_filename)
        assert test_file.userblock_size == 512
        assert test_file.filename == expected_filename
        test_file.close()  # Eli (8/11/20): always make sure to explicitly close the files or tests can fail on windows

max_stars_repo_name: CuriBio/mantarray-desktop-app

max_issues_repo_name: CuriBio/mantarray-desktop-app

max_forks_repo_name: CuriBio/mantarray-desktop-app


text:
# Copyright 2017 Google Inc.
#
# 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
#
#     https://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.
# ==============================================================================
"""Simplest FFN model, as described in https://arxiv.org/abs/1611.00421."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import tensorflow as tf

from .. import model

# Note: this model was originally trained with conv3d layers initialized with
# TruncatedNormalInitializedVariable with stddev = 0.01.
def _predict_object_mask(input_patches, input_seed, depth=9, is_training=True, adabn=False):
  """Computes single-object mask prediction."""

  if not is_training:
    if adabn:
        bn_decay = 0.95
    else:
        bn_decay = 1.0
  else:
    bn_decay=0.95
  train_bn = is_training
  in_k = 12

  if input_patches.get_shape().as_list()[-1] == 2:
      image = tf.expand_dims(input_patches[:,:,:,:,0], axis=4)
      membrane = tf.expand_dims(input_patches[:,:,:,:,1], axis=4)
      image_k = in_k-1
  else:
      image = input_patches
      image_k = in_k

  x = tf.contrib.layers.conv3d(image,
                                 scope='conv0_a',
                                 num_outputs=image_k,
                                 kernel_size=(1, 12, 12),
                                 padding='SAME')
  if input_patches.get_shape().as_list()[-1] == 2:
      print('FFN-hgru-v5: using membrane as input')
      x = tf.concat([x, membrane], axis=4)

  from .layers.recurrent import htd_cnn
  with tf.variable_scope('htd_net'):
      hgru_net = htd_cnn.hGRU(var_scope='htd_net',
                              timesteps=8,
                              dtype=tf.float32,
                              use_3d=True,
                              train=is_training,
                              train_bn=train_bn,
                              bn_decay=bn_decay,
                              in_k=in_k,
                              hgru1_fsiz=[1, 9, 9],
                              hgru2_fsiz=[3, 5, 5],
                              hgru_td_fsiz=[1, 1, 1],
                              hgru_h1_nl=tf.nn.tanh,
                              hgru_h2_nl=tf.nn.tanh,
                              hgru_bistream_weights='shared',
                              hgru_symmetric_weights=True,
                              hgru_soft_coefficients=True,
                              belly_up_td=True,
                              ds_fsiz_list=[[2, 5, 5], [2, 4, 4]],
                              ds_conv_repeat=2,
                              ds_k_list=[18, 24],
                              ds_pool_list=[[1, 2, 2], [2, 2, 2]],
                              ds_stride_list=[[1, 2, 2], [2, 2, 2]],
                              use_dsus_skip=True)
      net = hgru_net.build(x, ffn_seed=input_seed)

  finalbn_param_initializer = {
      'moving_mean': tf.constant_initializer(0., dtype=tf.float32),
      'moving_variance': tf.constant_initializer(1., dtype=tf.float32),
      'gamma': tf.constant_initializer(0.1, dtype=tf.float32)
  }
  net = tf.nn.relu(net)
  net = tf.contrib.layers.batch_norm(
      inputs=net,
      scale=True,
      center=True,
      fused=True,
      renorm=False,
      decay=bn_decay,
      param_initializers=finalbn_param_initializer,
      is_training=train_bn)
  logits = tf.contrib.layers.conv3d(net,
                                    scope='conv_lom1',
                                    num_outputs=in_k,
                                    kernel_size=(1, 1, 1),
                                    activation_fn=None)
  logits = tf.contrib.layers.batch_norm(
      inputs=logits,
      scale=True,
      center=False,
      fused=True,
      renorm=False,
      decay=bn_decay,
      param_initializers=finalbn_param_initializer,
      is_training=train_bn)
  logits = tf.nn.relu(logits)
  logits = tf.contrib.layers.conv3d(logits,
                                    scope='conv_lom2',
                                    num_outputs=1,
                                    kernel_size=(1, 1, 1),
                                    activation_fn=None)
  import numpy as np
  extras = 0
  hgru_w = 0
  ff_fb = 0
  for x in tf.trainable_variables():
      prod = np.prod(x.get_shape().as_list())
      if ('hgru' in x.name):
          if ('w' in x.name):
              hgru_w += prod/2
          elif ('mlp' in x.name):
              hgru_w += prod
          else:
              print(x.name + ' '+ str(prod))
              extras += prod
      elif ('ff' in x.name) | ('fb' in x.name) | ('conv0' in x.name) | ('conv_lom' in x.name):
          if ('weight' in x.name):
              ff_fb += prod
          else:
              print(x.name + ' ' + str(prod))
              extras += prod
      else:
          print(x.name + ' ' + str(prod))
          extras += prod
  hgru_w = int(hgru_w)
  print('>>>>>>>>>>>>>>>>>>>>>>TRAINABLE VARS: ' + 'horizontal('+str(hgru_w)+') vertical('+str(ff_fb)+') extras('+str(extras)+')')
  print('>>>>>>>>>>>>>>>>>>>>>>TRAINABLE VARS: ' + 'total(' + str(hgru_w+ff_fb+extras) + ')')
  print('>>>>>>>>>>>>>>>>>>>>>>BN-TRAIN: ' + str(train_bn))
  return logits

class ConvStack3DFFNModel(model.FFNModel):
  dim = 3

  def __init__(self, with_membrane=False, fov_size=None, deltas=None, batch_size=None, depth=9, is_training=True, adabn=False, reuse=False, tag='', TA=None):
    super(ConvStack3DFFNModel, self).__init__(deltas, batch_size, with_membrane, validation_mode=not(is_training), tag=tag)

    self.set_uniform_io_size(fov_size)
    self.depth = depth
    self.reuse=reuse
    self.TA=TA
    self.is_training=is_training
    self.adabn=adabn

  def define_tf_graph(self):
    self.show_center_slice(self.input_seed)

    if self.input_patches is None:
      self.input_patches = tf.placeholder(
          tf.float32, [1] + list(self.input_image_size[::-1]) +[1],
          name='patches')

    with tf.variable_scope('seed_update', reuse=self.reuse):
      logit_update = _predict_object_mask(self.input_patches, self.input_seed,
                                          depth=self.depth, is_training=self.is_training, adabn=self.adabn)

    logit_seed = self.update_seed(self.input_seed, logit_update)

    # Make predictions available, both as probabilities and logits.
    self.logits = logit_seed
    self.logistic = tf.sigmoid(logit_seed)

    if self.labels is not None:
      self.set_up_sigmoid_pixelwise_loss(logit_seed)
      if self.TA is None:
        self.set_up_optimizer(max_gradient_entry_mag=0.0)
      else:
        self.set_up_optimizer(max_gradient_entry_mag=0.0, TA=self.TA)
      self.show_center_slice(logit_seed)
      self.show_center_slice(self.labels, sigmoid=False)
      self.add_summaries()

    # ADABN: Add only non-bn vars to saver
    var_list = tf.global_variables()
    moving_ops_names = ['moving_mean', 'moving_variance']
    # var_list = [
    #       x for x in var_list
    #       if x.name.split('/')[-1].split(':')[0] + ':'
    #       not in moving_ops_names]
    # self.saver = tf.train.Saver(
    #       var_list=var_list,
    #       keep_checkpoint_every_n_hours=100)
    # ADABN: Define bn-var initializer to reset moments every iteration
    moment_list = [x for x in tf.global_variables() if (moving_ops_names[0] in x.name) | (moving_ops_names[1] in x.name)]
    self.moment_list = None #moment_list
    self.ada_initializer = tf.variables_initializer( var_list=moment_list)

    self.fgru_moment_list = [x for x in moment_list if 'recurrent' in x.name]
    self.fgru_ada_initializer = tf.variables_initializer(var_list=self.fgru_moment_list)
    self.ext_moment_list =  [x for x in moment_list if 'recurrent' not in x.name]
    self.ext_ada_initializer = tf.variables_initializer(var_list=self.ext_moment_list)

    self.saver = tf.train.Saver(keep_checkpoint_every_n_hours=1)

ext: py

reason: import numpy


text:
#!/usr/bin/env python
# -*- coding: utf-8 -*-

#-----------------------------------------------------------------------------
# Copyright (c) 2018-2021, NeXpy Development Team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file COPYING, distributed with this software.
#-----------------------------------------------------------------------------

"""
This module contains a set of customized widgets both for dialogs and plot objects.
"""
import math
import warnings

import numpy as np

from .pyqt import QtCore, QtGui, QtWidgets

from matplotlib import cbook, colors
from matplotlib.patches import Circle, Ellipse, Polygon, Rectangle

from .utils import (boundaries, find_nearest, format_float, get_color,
                    natural_sort, report_error)

warnings.filterwarnings("ignore", category=cbook.mplDeprecation)

bold_font = QtGui.QFont()
bold_font.setBold(True)

class NXStack(QtWidgets.QWidget):
    """Widget containing a stack of widgets selected by a dropdown menu.

    Attributes
    ----------
    layout : QtWidgets.QVBoxLayout
        Layout of the entire stack.
    stack : QtWidgets.QStackedWidget
        Widget containing the stacked widgets.
    box : QtWidgets.QComboBox
        Pull-down menu containing the stack options.
    """
    def __init__(self, labels, widgets, parent=None):
        """Initialize the widget stack.

        Parameters
        ----------
        labels : list of str
            List of labels to be used in the QComboBox.
        widgets : list of QWidgets
            List of QWidgets to be stacked.
        parent : QObject, optional
            Parent of the NXStack instance (the default is None).
        """
        super(NXStack, self).__init__(parent=parent)
        self.layout = QtWidgets.QVBoxLayout()
        self.stack = QtWidgets.QStackedWidget(self)
        self.widgets = dict(zip(labels, widgets))
        self.box = NXComboBox(slot=self.stack.setCurrentIndex, items=labels)
        for widget in widgets:
            self.stack.addWidget(widget)
        self.layout.addWidget(self.box)
        self.layout.addWidget(self.stack)
        self.layout.addStretch()
        self.setLayout(self.layout)

    def add(self, label, widget):
        """Add a widget to the stack.

        Parameters
        ----------
        label : str
            Label used to select the widget in the QComboBox
        widget : QtWidgets.QWidget
            Widget to be added to the stack
        """
        self.box.addItem(label)
        self.stack.addWidget(widget)

    def remove(self, label):
        if label in self.widgets:
            self.stack.removeWidget(self.widgets[label])
            del self.widgets[label]
        self.box.remove(label)

class NXSortModel(QtCore.QSortFilterProxyModel):

    def __init__(self, parent=None):
        super(NXSortModel, self).__init__(parent=parent)

    def lessThan(self, left, right):
        left_text = self.sourceModel().itemFromIndex(left).text()
        right_text = self.sourceModel().itemFromIndex(right).text()
        return natural_sort(left_text) < natural_sort(right_text)

class NXScrollArea(QtWidgets.QScrollArea):
    """Scroll area embedding a widget."""

    def __init__(self, widget=None, horizontal=False, parent=None):
        """Initialize the scroll area.

        Parameters
        ----------
        widget : QtWidgets.QWidget
            Widget contained within the scroll area.
        horizontal : bool
            True if a horizontal scroll bar is enabled, default False.
        """
        super(NXScrollArea, self).__init__(parent=parent)
        if widget:
            self.setWidget(widget)
        if not horizontal:
            self.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarAlwaysOff)
        self.setWidgetResizable(True)
        self.setSizePolicy(QtWidgets.QSizePolicy.Expanding,
                           QtWidgets.QSizePolicy.Expanding)

    def setWidget(self, widget):
        super(NXScrollArea, self).setWidget(widget)
        widget.setMinimumWidth(widget.sizeHint().width() +
                               self.verticalScrollBar().sizeHint().width())

class NXLabel(QtWidgets.QLabel):
    """A text label.

    This is being subclassed from the PyQt QLabel class because of a bug in
    recent versions of PyQt5 (>11) that requires the box to be repainted
    after any programmatic changes.
    """

    def __init__(self, text=None, parent=None, bold=False, width=None,
                 align='left'):
        """Initialize the edit window and optionally set the alignment

        Parameters
        ----------
        text : str, optional
            The default text.
        parent : QWidget
            Parent of the NXLineEdit box.
        bold : bool, optional
            True if the label text is bold, default False.
        width : int, optional
            Fixed width of label.
        align : 'left', 'center', 'right'
            Alignment of text.
        """
        super(NXLabel, self).__init__(parent=parent)
        if text:
            self.setText(text)
        if bold:
            self.setFont(bold_font)
        if width:
            self.setFixedWidth(width)
        if align == 'left':
            self.setAlignment(QtCore.Qt.AlignLeft | QtCore.Qt.AlignVCenter)
        elif align == 'center':
            self.setAlignment(QtCore.Qt.AlignHCenter | QtCore.Qt.AlignVCenter)
        elif align == 'right':
            self.setAlignment(QtCore.Qt.AlignRight | QtCore.Qt.AlignVCenter)

    def setText(self, text):
        """Function to set the text in the box.

        Parameters
        ----------
        text : str
            Text to replace the text box contents.
        """
        super(NXLabel, self).setText(str(text))
        self.repaint()

class NXLineEdit(QtWidgets.QLineEdit):
    """An editable text box.

    This is being subclassed from the PyQt QLineEdit class because of a bug in
    recent versions of PyQt5 (>11) that requires the box to be repainted
    after any programmatic changes.
    """

    def __init__(self, text=None, parent=None, slot=None, readonly=False,
                 width=None, align='left'):
        """Initialize the edit window and optionally set the alignment

        Parameters
        ----------
        text : str, optional
            The default text.
        parent : QWidget
            Parent of the NXLineEdit box.
        slot: func, optional
            Slot to be used for editingFinished signals.
        right : bool, optional
            If True, make the box text right-aligned.
        """
        super(NXLineEdit, self).__init__(parent=parent)
        if slot:
            self.editingFinished.connect(slot)
        if text is not None:
            self.setText(text)
        if readonly:
            self.setReadOnly(True)
        if width:
            self.setFixedWidth(width)
        if align == 'left':
            self.setAlignment(QtCore.Qt.AlignLeft | QtCore.Qt.AlignVCenter)
        elif align == 'center':
            self.setAlignment(QtCore.Qt.AlignHCenter | QtCore.Qt.AlignVCenter)
        elif align == 'right':
            self.setAlignment(QtCore.Qt.AlignRight | QtCore.Qt.AlignVCenter)

    def setText(self, text):
        """Function to set the text in the box.

        Parameters
        ----------
        text : str
            Text to replace the text box contents.
        """
        super(NXLineEdit, self).setText(str(text))
        self.repaint()

class NXTextBox(NXLineEdit):
    """Subclass of NXLineEdit with floating point values."""

    def value(self):
        """Return the text box value as a floating point number.

        Returns
        -------
        float
            Value of text box converted to a floating point number
        """
        return float(str(self.text()))

    def setValue(self, value):
        """Set the value of the text box string formatted as a float.

        Parameters
        ----------
        value : str or int or float
            Text box value to be formatted as a float
        """
        self.setText(str(float('%.4g' % value)))

class NXPlainTextEdit(QtWidgets.QPlainTextEdit):
    """An editable text window."""

    def __init__(self, text=None, wrap=True, parent=None):
        super(NXPlainTextEdit, self).__init__(parent=parent)
        self.setFont(QtGui.QFont('Courier'))
        if not wrap:
            self.setLineWrapMode(QtWidgets.QPlainTextEdit.NoWrap)
        if text:
            self.setPlainText(text)
        self.setFocusPolicy(QtCore.Qt.StrongFocus)

    def __repr__(self):
        return 'NXPlainTextEdit()'

    def setPlainText(self, text):
        """Function to set the text in the window.

        Parameters
        ----------
        text : str
            Text to replace the text box contents.
        """
        super(NXPlainTextEdit, self).setPlainText(str(text))
        self.repaint()

    def get_text(self, tab_spaces=4):
        """Return the text contained in the edit window.

        Parameters
        ----------
        tab_spaces : int, optional
            Number of spaces to replace tabs (default is 4). If set to 0, tab
            characters are not replaced.

        Returns
        -------
        str
            Current text in the edit window.
        """
        text = self.document().toPlainText().strip()
        if tab_spaces > 0:
            return text.replace('\t', tab_spaces*' ')
        else:
            return text + '\n'

class NXMessageBox(QtWidgets.QMessageBox):
    """A scrollable message box"""

    def __init__(self, title, text, *args, **kwargs):
        super(NXMessageBox, self).__init__(*args, **kwargs)
        scroll = NXScrollArea(parent=self)
        self.content = QtWidgets.QWidget()
        scroll.setWidget(self.content)
        scroll.setWidgetResizable(True)
        layout = QtWidgets.QVBoxLayout(self.content)
        layout.addWidget(NXLabel(title, bold=True))
        layout.addWidget(NXLabel(text, self))
        self.layout().addWidget(scroll, 0, 0, 1, self.layout().columnCount())
        self.setStyleSheet("QScrollArea{min-width:300 px; min-height: 400px}")

class NXComboBox(QtWidgets.QComboBox):
    """Dropdown menu for selecting a set of options."""

    def __init__(self, slot=None, items=[], default=None):
        """Initialize the dropdown menu with an initial list of items

        Parameters
        ----------
        slot : func, optional
            A function to be called when a selection is made
        items : list of str, optional
            A list of options to initialize the dropdown menu
        default : str, optional
            The option to be set as default when the menu is initialized
        """
        super(NXComboBox, self).__init__()
        self.setSizeAdjustPolicy(QtWidgets.QComboBox.AdjustToContents)
        self.setFocusPolicy(QtCore.Qt.StrongFocus)
        self.setMinimumWidth(80)
        if items:
            self.addItems([str(item) for item in items])
            if default:
                self.setCurrentIndex(self.findText(default))
        if slot:
            self.activated.connect(slot)

    def __iter__(self):
        """Implement key iteration."""
        return self.items().__iter__()

    def __next__(self):
        """Implements key iteration."""
        return self.items().__next__()

    def __contains__(self, item):
        """True if the item is one of the options."""
        return item in self.items()

    def keyPressEvent(self, event):
        """Function to enable the use of cursor keys to make selections.

        `Up` and `Down` keys are used to select options in the dropdown menu.
        `Left` and `Right` keys ar used to expand the dropdown menu to
        display the options.

        Parameters
        ----------
        event : QtCore.QEvent
            Keypress event that triggered the function.
        """
        if (event.key() == QtCore.Qt.Key_Up or
            event.key() == QtCore.Qt.Key_Down):
            super(NXComboBox, self).keyPressEvent(event)
        elif (event.key() == QtCore.Qt.Key_Right or
              event.key() == QtCore.Qt.Key_Left):
            self.showPopup()
        else:
            self.parent().keyPressEvent(event)

    def findText(self, value, **kwargs):
        """Function to return the index of a text value.

        This is needed since h5py now returns byte strings, which will trigger
        ValueErrors unless they are converted to unicode strings.

        Parameters
        ----------
        value :
            Searched value.

        Returns
        -------
        int
            Index of the searched value.
        """
        if isinstance(value, bytes):
            value = value.decode('utf-8')
        return super(NXComboBox, self).findText(str(value), **kwargs)

    def add(self, *items):
        """Add items to the list of options.

        Parameters
        ----------
        *items : list of str
            List of options to be added to the dropdown menu.
        """
        for item in items:
            if item not in self:
                self.addItem(str(item))

    def insert(self, idx, item):
        """Insert item at the specified index.

        Parameters
        ----------
        item : str or int
            List of options to be added to the dropdown menu.
        idx : int
            Index of position before which to insert item
        """
        if item not in self:
            self.insertItem(idx, str(item))

    def remove(self, item):
        """Remove item from the list of options.

        Parameters
        ----------
        item : str or int
            Option to be removed from the dropdown menu.
        """
        if str(item) in self:
            self.removeItem(self.findText(str(item)))

    def items(self):
        """Return a list of the dropdown menu options.

        Returns
        -------
        list of str
            The options currently listed in the dropdown menu
        """
        return [self.itemText(idx) for idx in range(self.count())]

    def sort(self):
        """Sorts the box items in alphabetical order."""
        self.model().sort(0)

    def select(self, item):
        """Select the option matching the text.

        Parameters
        ----------
        item : str
            The option to be selected in the dropdown menu.
        """
        self.setCurrentIndex(self.findText(str(item)))
        self.repaint()

    @property
    def selected(self):
        """Return the currently selected option.

        Returns
        -------
        str
            Currently selected option in the dropdown menu.
        """
        return self.currentText()

class NXCheckBox(QtWidgets.QCheckBox):
    """A checkbox with associated label and slot function."""

    def __init__(self, label=None, slot=None, checked=False):
        """Initialize the checkbox.

        Parameters
        ----------
        label : str, optional
            Text describing the checkbox.
        slot : func, optional
            Function to be called when the checkbox state is changed.
        checked : bool, optional
            Initial checkbox state (the default is False).
        """
        super(NXCheckBox, self).__init__(label)
        self.setFocusPolicy(QtCore.Qt.StrongFocus)
        self.setChecked(checked)
        if slot:
            self.stateChanged.connect(slot)

    def keyPressEvent(self, event):
        """Function to enable the use of cursor keys to change the state.

        `Up` and `Down` keys are used to toggle the checkbox state.

        Parameters
        ----------
        event : QtCore.QEvent
            Keypress event that triggered the function.
        """
        if (event.key() == QtCore.Qt.Key_Up or
            event.key() == QtCore.Qt.Key_Down):
            if self.isChecked():
                self.setCheckState(QtCore.Qt.Unchecked)
            else:
                self.setCheckState(QtCore.Qt.Checked)
        else:
            self.parent().keyPressEvent(event)

class NXPushButton(QtWidgets.QPushButton):
    """A button with associated label and slot function."""

    def __init__(self, label, slot, checkable=False, width=None, parent=None):
        """Initialize button

        Parameters
        ----------
        label : str
            Text describing the button
        slot : func
            Function to be called when the button is pressed
        parent : QObject, optional
            Parent of button.
        """
        super(NXPushButton, self).__init__(label, parent=parent)
        self.setFocusPolicy(QtCore.Qt.StrongFocus)
        self.setDefault(False)
        self.setAutoDefault(False)
        self.clicked.connect(slot)
        if checkable:
            self.setCheckable(True)
        if width:
            self.setFixedWidth(width)

    def keyPressEvent(self, event):
        """Function to enable the use of keys to press the button.

        `Return`, Enter`, and `Space` keys activate the slot function.

        Parameters
        ----------
        event : QtCore.QEvent
            Keypress event that triggered the function.
        """
        if (event.key() == QtCore.Qt.Key_Return or
            event.key() == QtCore.Qt.Key_Enter or
            event.key() == QtCore.Qt.Key_Space):
            self.clicked.emit()
        else:
            self.parent().keyPressEvent(event)

class NXColorButton(QtWidgets.QPushButton):
    """Push button for selecting colors."""

    colorChanged = QtCore.Signal(QtGui.QColor)

    def __init__(self, parent=None):
        super(NXColorButton, self).__init__(parent=parent)
        self.setFixedWidth(18)
        self.setStyleSheet("width:18px; height:18px; "
                           "margin: 0px; border: 0px; padding: 0px;"
                           "background-color: white")
        self.setIconSize(QtCore.QSize(12, 12))
        self.clicked.connect(self.choose_color)
        self._color = QtGui.QColor()

    def choose_color(self):
        color = QtWidgets.QColorDialog.getColor(self._color, self.parentWidget())
        if color.isValid():
            self.set_color(color)

    def get_color(self):
        return self._color

    @QtCore.Slot(QtGui.QColor)
    def set_color(self, color):
        if color != self._color:
            self._color = color
            self.colorChanged.emit(self._color)
            pixmap = QtGui.QPixmap(self.iconSize())
            pixmap.fill(color)
            self.setIcon(QtGui.QIcon(pixmap))
            self.repaint()

    color = QtCore.Property(QtGui.QColor, get_color, set_color)

class NXColorBox(QtWidgets.QWidget):
    """Text box and color square for selecting colors.

    This utilizes the ColorButton class in the formlayout package.

    Attributes
    ----------
    layout : QHBoxLayout
        Layout containing the text and color boxes.
    box : NXLineEdit
        Text box containing the string representation of the color.
    button : QPushbutton
        Color button consisting of a colored icon.
    """

    def __init__(self, color='#ffffff', label=None, width=None, parent=None):
        """Initialize the text and color box.

        The selected color can be changed by entering a valid text string or
        by selecting the color using the standard system GUI.

        Valid text strings are HTML hex strings or standard Matplotlib colors.

        Parameters
        ----------
        color : str, optional
            Initial color (the default is '#ffffff', which represents 'white')
        parent : QObject, optional
            Parent of the color box.
        """
        super(NXColorBox, self).__init__(parent=parent)
        self.color_text = get_color(color)
        color = self.qcolor(self.color_text)
        self.layout = QtWidgets.QHBoxLayout()
        self.layout.setContentsMargins(0, 0, 0, 0)
        if label:
            self.layout.addStretch()
            self.layout.addWidget(NXLabel(label))
        self.textbox = NXLineEdit(self.color_text,
                                  parent=parent, slot=self.update_color,
                                  width=width, align='right')
        self.layout.addWidget(self.textbox)
        self.button = NXColorButton(parent=parent)
        self.button.color = color
        self.button.colorChanged.connect(self.update_text)
        self.layout.addWidget(self.button)
        self.setLayout(self.layout)
        self.update_color()

    def update_color(self):
        """Set the button color following a change to the text box."""
        try:
            color = self.qcolor(get_color(self.textbox.text()))
            if color.isValid():
                self.button.color = color
                self.color_text = self.textbox.text()
        except ValueError as error:
            report_error('Invalid color', error)
            self.textbox.setText(self.color_text)

    def update_text(self, color):
        """Set the text box string following a change to the color button."""
        self.color_text = colors.to_hex(color.getRgbF())
        self.textbox.setText(self.color_text)

    def qcolor(self, text):
        """Create a QColor from a Matplotlib color."""
        qcolor = QtGui.QColor()
        text = get_color(text)
        if text.startswith('#') and len(text)==7:
            correct = '#0123456789abcdef'
            for char in text:
                if char.lower() not in correct:
                    return qcolor
        elif text not in list(QtGui.QColor.colorNames()):
            return qcolor
        qcolor.setNamedColor(text)
        return qcolor

class NXSpinBox(QtWidgets.QSpinBox):
    """Subclass of QSpinBox with floating values.

    Parameters
    ----------
    slot : function
        PyQt slot triggered by changing values
    data : array-like, optional
        Values of data to be adjusted by the spin box.

    Attributes
    ----------
    data : array-like
        Data values.
    validator : QDoubleValidator
        Function to ensure only floating point values are entered.
    old_value : float
        Previously stored value.
    diff : float
        Difference between maximum and minimum values when the box is
        locked.
    pause : bool
        Used when playing a movie with changing z-values.
    """
    def __init__(self, slot=None, data=None):
        super(NXSpinBox, self).__init__()
        self.data = data
        self.validator = QtGui.QDoubleValidator()
        self.old_value = None
        self.diff = None
        self.pause = False
        if slot:
            self.valueChanged.connect(slot)
            self.editingFinished.connect(slot)
        self.setAlignment(QtCore.Qt.AlignRight)
        self.setFixedWidth(100)
        self.setKeyboardTracking(False)
        self.setAccelerated(False)
        self.app = QtWidgets.QApplication.instance()

    def value(self):
        """Return the value of the spin box.

        Returns
        -------
        float
            Floating point number defined by the spin box value
        """
        if self.data is not None:
            return float(self.centers[self.index])
        else:
            return 0.0

    @property
    def centers(self):
        """The values of the data points based on bin centers.

        Returns
        -------
        array-like
            Data points set by the spin box
        """
        if self.data is None:
            return None
        elif self.reversed:
            return self.data[::-1]
        else:
            return self.data

    @property
    def boundaries(self):
        if self.data is None:
            return None
        else:
            return boundaries(self.centers, self.data.shape[0])

    @property
    def index(self):
        """Return the current index of the spin box."""
        return super(NXSpinBox, self).value()

    @property
    def reversed(self):
        """Return `True` if the data are in reverse order."""
        if self.data[-1] < self.data[0]:
            return True
        else:
            return False

    def setValue(self, value):
        super(NXSpinBox, self).setValue(self.valueFromText(value))
        self.repaint()

    def valueFromText(self, text):
        return self.indexFromValue(float(str(text)))

    def textFromValue(self, value):
        try:
            return format_float(float('%.4g' % self.centers[value]))
        except:
            return ''

    def valueFromIndex(self, idx):
        if idx < 0:
            return self.centers[0]
        elif idx > self.maximum():
            return self.centers[-1]
        else:
            return self.centers[idx]

    def indexFromValue(self, value):
        return (np.abs(self.centers - value)).argmin()

    def minBoundaryValue(self, idx):
        if idx <= 0:
            return self.boundaries[0]
        elif idx >= len(self.centers) - 1:
            return self.boundaries[-2]
        else:
            return self.boundaries[idx]

    def maxBoundaryValue(self, idx):
        if idx <= 0:
            return self.boundaries[1]
        elif idx >= len(self.centers) - 1:
            return self.boundaries[-1]
        else:
            return self.boundaries[idx+1]

    def validate(self, input_value, pos):
        return self.validator.validate(input_value, pos)

    @property
    def tolerance(self):
        return self.diff / 100.0

    def stepBy(self, steps):
        self.pause = False
        if self.diff:
            value = self.value() + steps * self.diff
            if (value <= self.centers[-1] + self.tolerance) and \
               (value - self.diff >= self.centers[0] - self.tolerance):
                self.setValue(value)
            else:
                self.pause = True
        else:
            if self.index + steps <= self.maximum() and \
               self.index + steps >= 0:
                super(NXSpinBox, self).stepBy(steps)
            else:
                self.pause = True

    def timerEvent(self, event):
        self.app.processEvents()
        if self.app.mouseButtons() & QtCore.Qt.LeftButton:
            super(NXSpinBox, self).timerEvent(event)

class NXDoubleSpinBox(QtWidgets.QDoubleSpinBox):
    """Subclass of QDoubleSpinBox.

    Parameters
    ----------
    slot : function
        PyQt slot triggered by changing values

    Attributes
    ----------
    validator : QDoubleValidator
        Function to ensure only floating point values are entered.
    old_value : float
        Previously stored value.
    diff : float
        Difference between maximum and minimum values when the box is
        locked.
    """
    def __init__(self, slot=None, editing=None):
        super(NXDoubleSpinBox, self).__init__()
        self.validator = QtGui.QDoubleValidator()
        self.validator.setRange(-np.inf, np.inf)
        self.validator.setDecimals(1000)
        self.old_value = None
        self.diff = None
        if slot and editing:
            self.valueChanged.connect(slot)
            self.editingFinished.connect(editing)
        elif slot:
            self.valueChanged.connect(slot)
            self.editingFinished.connect(slot)
        self.setAlignment(QtCore.Qt.AlignRight)
        self.setFixedWidth(100)
        self.setKeyboardTracking(False)
        self.setDecimals(2)
        self.steps = np.array([1, 2, 5, 10])
        self.app = QtWidgets.QApplication.instance()

    def validate(self, input_value, position):
        return self.validator.validate(input_value, position)

    def setSingleStep(self, value):
        value = abs(value)
        if value == 0:
            stepsize = 0.01
        else:
            digits = math.floor(math.log10(value))
            multiplier = 10**digits
            stepsize = find_nearest(self.steps, value/multiplier) * multiplier
        super(NXDoubleSpinBox, self).setSingleStep(stepsize)

    def stepBy(self, steps):
        if self.diff:
            self.setValue(self.value() + steps * self.diff)
        else:
            super(NXDoubleSpinBox, self).stepBy(steps)
        self.old_value = self.text()

    def valueFromText(self, text):
        value = float(text)
        if value > self.maximum():
            self.setMaximum(value)
        elif value < self.minimum():
            self.setMinimum(value)
        return value

    def textFromValue(self, value):
        if value > 1e6:
            return format_float(value)
        else:
            return format_float(value, width=8)

    def setValue(self, value):
        if value == 0:
            self.setDecimals(2)
        else:
            digits = math.floor(math.log10(abs(value)))
            if digits < 0:
                self.setDecimals(-digits)
            else:
                self.setDecimals(2)
        if value > self.maximum():
            self.setMaximum(value)
        elif value < self.minimum():
            self.setMinimum(value)
        super(NXDoubleSpinBox, self).setValue(value)
        self.repaint()

    def timerEvent(self, event):
        self.app.processEvents()
        if self.app.mouseButtons() & QtCore.Qt.LeftButton:
            super(NXDoubleSpinBox, self).timerEvent(event)

class NXSlider(QtWidgets.QSlider):
    """Subclass of QSlider.

    Parameters
    ----------
    slot : function
        PyQt slot triggered by changing values
    move : bool
        True if the slot is triggered by moving the slider. Otherwise,
        it is only triggered on release.
    """
    def __init__(self, slot=None, move=True, inverse=False):
        super(NXSlider, self).__init__(QtCore.Qt.Horizontal)
        self.setFocusPolicy(QtCore.Qt.NoFocus)
        self.setMinimumWidth(100)
        self.setRange(0, 100)
        self.setSingleStep(5)
        self.setTracking(True)
        self.inverse = inverse
        if self.inverse:
            self.setInvertedAppearance(True)
            self.setValue(100)
        else:
            self.setInvertedAppearance(False)
            self.setValue(0)
        if slot:
            self.sliderReleased.connect(slot)
            if move:
                self.sliderMoved.connect(slot)

    def value(self):
        _value = super(NXSlider, self).value()
        if self.inverse:
            return self.maximum() - _value
        else:
            return _value

    def setValue(self, value):
        if self.inverse:
            super(NXSlider, self).setValue(self.maximum() - value)
        else:
            super(NXSlider, self).setValue(value)

class NXpatch(object):
    """Class for a draggable shape on the NXPlotView canvas."""
    lock = None

    def __init__(self, shape, border_tol=0.1, resize=True, plotview=None):
        if plotview:
            self.plotview = plotview
        else:
            from .plotview import get_plotview
            self.plotview = get_plotview()
        self.canvas = self.plotview.canvas
        self.shape = shape
        self.border_tol = border_tol
        self.press = None
        self.background = None
        self.allow_resize = resize
        self.plotview.ax.add_patch(self.shape)

    def __getattr__(self, name):
        """Return Matplotlib attributes if not defined in the class."""
        return getattr(self.shape, name)

    def connect(self):
        'connect to all the events we need'
        self.plotview.deactivate()
        self.cidpress = self.canvas.mpl_connect(
            'button_press_event', self.on_press)
        self.cidrelease = self.canvas.mpl_connect(
            'button_release_event', self.on_release)
        self.cidmotion = self.canvas.mpl_connect(
            'motion_notify_event', self.on_motion)

    def is_inside(self, event):
        if event.inaxes != self.shape.axes:
            return False
        contains, _ = self.shape.contains(event)
        if contains:
            return True
        else:
            return False

    def initialize(self, xp, yp):
        """Function to be overridden by shape sub-class."""

    def update(self, x, y):
        """Function to be overridden by shape sub-class."""

    def on_press(self, event):
        """Store coordinates on button press if over the object."""
        if not self.is_inside(event):
            self.press = None
            return
        self.press = self.initialize(event.xdata, event.ydata)
        self.canvas.draw()

    def on_motion(self, event):
        """Move the object if motion activated over the object."""
        if self.press is None:
            return
        if event.inaxes != self.shape.axes:
            return
        self.update(event.xdata, event.ydata)
        self.canvas.draw()

    def on_release(self, event):
        """Reset the data when the button is released."""
        if self.press is None:
            return
        self.press = None
        self.canvas.draw()

    def disconnect(self):
        """Disconnect all the stored connection ids."""
        self.canvas.mpl_disconnect(self.cidpress)
        self.canvas.mpl_disconnect(self.cidrelease)
        self.canvas.mpl_disconnect(self.cidmotion)
        self.plotview.activate()

    def remove(self):
        if self in self.plotview.shapes:
            self.plotview.shapes.remove(self)
        self.shape.remove()
        self.plotview.draw()

    def set_facecolor(self, color):
        self.shape.set_facecolor(color)
        self.plotview.draw()

    def set_edgecolor(self, color):
        self.shape.set_edgecolor(color)
        self.plotview.draw()

    def set_color(self, color):
        self.shape.set_facecolor(color)
        self.shape.set_edgecolor(color)
        self.plotview.draw()

    def set_alpha(self, alpha):
        self.shape.set_alpha(alpha)
        self.plotview.draw()

    def set_linestyle(self, linestyle):
        self.shape.set_linestyle(linestyle)
        self.plotview.draw()

    def set_linewidth(self, linewidth):
        self.shape.set_linewidth(linewidth)
        self.plotview.draw()

class NXcircle(NXpatch):

    def __init__(self, x, y, r, border_tol=0.1, resize=True, plotview=None,
                 **opts):
        x, y, r = float(x), float(y), float(r)
        shape = Ellipse((x,y), 2*r, 2*r, **opts)
        if 'linewidth' not in opts:
            shape.set_linewidth(1.0)
        if 'color' not in opts and 'facecolor' not in opts:
            shape.set_facecolor('r')
        super(NXcircle, self).__init__(shape, border_tol, resize, plotview)
        self.shape.set_label('Circle')
        self.circle = self.shape
        self.circle.height = self.height

    def __repr__(self):
        x, y = self.circle.center
        r = abs(self.circle.width) / 2
        return 'NXcircle(%g, %g, %g)' % (x, y, r)

    @property
    def transform(self):
        return self.plotview.ax.transData.transform

    @property
    def inverse_transform(self):
        return self.plotview.ax.transData.inverted().transform

    @property
    def center(self):
        return self.circle.center

    @property
    def radius(self):
        return abs(self.circle.width) / 2.0

    @property
    def width(self):
        return abs(self.circle.width)

    @property
    def height(self):
        return 2 * (self.inverse_transform((0,self.pixel_radius)) -
                    self.inverse_transform((0,0)))[1]

    @property
    def pixel_radius(self):
        return (self.transform((self.radius,0)) - self.transform((0,0)))[0]

    def pixel_shift(self, x, y, x0, y0):
        return tuple(self.transform((x,y)) - self.transform((x0,y0)))

    def radius_shift(self, x, y, xp, yp, x0, y0):
        xt, yt = self.pixel_shift(x, y, x0, y0)
        r = np.sqrt(xt**2 + yt**2)
        xt, yt = self.pixel_shift(xp, yp, x0, y0)
        r0 = np.sqrt(xt**2 + yt**2)
        return (self.inverse_transform((r,0)) -
                self.inverse_transform((r0,0)))[0]

    def set_center(self, x, y):
        self.circle.center = x, y
        self.plotview.draw()

    def set_radius(self, radius):
        self.circle.width = 2.0 * radius
        self.circle.height = self.height
        self.plotview.draw()

    def initialize(self, xp, yp):
        x0, y0 = self.circle.center
        w0, h0 = self.width, self.height
        xt, yt = self.pixel_shift(xp, yp, x0, y0)
        rt = self.pixel_radius
        if (self.allow_resize and
            (np.sqrt(xt**2 + yt**2) > rt * (1-self.border_tol))):
            expand = True
        else:
            expand = False
        return x0, y0, w0, h0, xp, yp, expand

    def update(self, x, y):
        x0, y0, w0, h0, xp, yp, expand = self.press
        if expand:
            self.circle.width = self.width + self.radius_shift(x, y, xp, yp, x0, y0)
            self.circle.height = self.height
        else:
            self.circle.center = (x0+x-xp, y0+y-yp)

class NXellipse(NXpatch):

    def __init__(self, x, y, dx, dy, border_tol=0.2, resize=True, plotview=None,
                 **opts):
        shape = Ellipse((float(x),float(y)), dx, dy, **opts)
        if 'linewidth' not in opts:
            shape.set_linewidth(1.0)
        if 'color' not in opts and 'facecolor' not in opts:
            shape.set_facecolor('r')
        super(NXellipse, self).__init__(shape, border_tol, resize, plotview)
        self.shape.set_label('Ellipse')
        self.ellipse = self.shape

    def __repr__(self):
        x, y = self.ellipse.center
        w, h = self.ellipse.width, self.ellipse.height
        return 'NXellipse(%g, %g, %g, %g)' % (x, y, w, h)

    @property
    def center(self):
        return self.ellipse.center

    @property
    def width(self):
        return self.ellipse.width

    @property
    def height(self):
        return self.ellipse.height

    def set_center(self, x, y):
        self.ellipse.set_center((x, y))
        self.plotview.draw()

    def set_width(self, width):
        self.ellipse.width = width
        self.plotview.draw()

    def set_height(self, height):
        self.ellipse.height = height
        self.plotview.draw()

    def initialize(self, xp, yp):
        x0, y0 = self.ellipse.center
        w0, h0 = self.ellipse.width, self.ellipse.height
        bt = self.border_tol
        if (self.allow_resize and
            ((abs(x0-xp) < bt*w0 and abs(y0+np.true_divide(h0,2)-yp) < bt*h0) or
             (abs(x0-xp) < bt*w0 and abs(y0-np.true_divide(h0,2)-yp) < bt*h0) or
             (abs(y0-yp) < bt*h0 and abs(x0+np.true_divide(w0,2)-xp) < bt*w0) or
             (abs(y0-yp) < bt*h0 and abs(x0-np.true_divide(w0,2)-xp) < bt*w0))):
            expand = True
        else:
            expand = False
        return x0, y0, w0, h0, xp, yp, expand

    def update(self, x, y):
        x0, y0, w0, h0, xp, yp, expand = self.press
        dx, dy = (x-xp, y-yp)
        bt = self.border_tol
        if expand:
            if (abs(x0-xp) < bt*w0 and abs(y0+np.true_divide(h0,2)-yp) < bt*h0):
                self.ellipse.height = h0 + dy
            elif (abs(x0-xp) < bt*w0 and abs(y0-np.true_divide(h0,2)-yp) < bt*h0):
                self.ellipse.height = h0 - dy
            elif (abs(y0-yp) < bt*h0 and abs(x0+np.true_divide(w0,2)-xp) < bt*w0):
                self.ellipse.width = w0 + dx
            elif (abs(y0-yp) < bt*h0 and abs(x0-np.true_divide(w0,2)-xp) < bt*w0):
                self.ellipse.width = w0 - dx
        else:
            self.ellipse.set_center((x0+dx, y0+dy))

class NXrectangle(NXpatch):

    def __init__(self, x, y, dx, dy, border_tol=0.1, resize=True, plotview=None,
                 **opts):
        shape = Rectangle((float(x),float(y)), float(dx), float(dy), **opts)
        if 'linewidth' not in opts:
            shape.set_linewidth(1.0)
        if 'color' not in opts and 'facecolor' not in opts:
            shape.set_facecolor('r')
        super(NXrectangle, self).__init__(shape, border_tol, resize, plotview)
        self.shape.set_label('Rectangle')
        self.rectangle = self.shape

    def __repr__(self):
        x, y = self.rectangle.xy
        w, h = self.rectangle.get_width(), self.rectangle.get_height()
        return 'NXrectangle(%g, %g, %s, %s)' % (x, y, w, h)

    @property
    def width(self):
        return self.rectangle.get_width()

    @property
    def height(self):
        return self.rectangle.get_height()

    @property
    def xy(self):
        return self.rectangle.xy

    def set_bounds(self, x, y, dx, dy):
        self.rectangle.set_bounds(x, y, dx, dy)
        self.plotview.draw()

    def set_left(self, left):
        self.rectangle.set_x(left)
        self.plotview.draw()

    def set_right(self, right):
        self.rectangle.set_x(right - self.rectangle.get_width())
        self.plotview.draw()

    def set_bottom(self, bottom):
        self.rectangle.set_y(bottom)
        self.plotview.draw()

    def set_top(self, top):
        self.rectangle.set_y(top - self.rectangle.get_height())
        self.plotview.draw()

    def set_width(self, width):
        self.rectangle.set_width(width)
        self.plotview.draw()

    def set_height(self, height):
        self.rectangle.set_height(height)
        self.plotview.draw()

    def initialize(self, xp, yp):
        x0, y0 = self.rectangle.xy
        w0, h0 = self.rectangle.get_width(), self.rectangle.get_height()
        bt = self.border_tol
        if (self.allow_resize and
            (abs(x0+np.true_divide(w0,2)-xp)>np.true_divide(w0,2)-bt*w0 or
             abs(y0+np.true_divide(h0,2)-yp)>np.true_divide(h0,2)-bt*h0)):
            expand = True
        else:
            expand = False
        return x0, y0, w0, h0, xp, yp, expand

    def update(self, x, y):
        x0, y0, w0, h0, xp, yp, expand = self.press
        dx, dy = (x-xp, y-yp)
        bt = self.border_tol
        if expand:
            if abs(x0 - xp) < bt * w0:
                self.rectangle.set_x(x0+dx)
                self.rectangle.set_width(w0-dx)
            elif abs(x0 + w0 - xp) < bt * w0:
                self.rectangle.set_width(w0+dx)
            elif abs(y0 - yp) < bt * h0:
                self.rectangle.set_y(y0+dy)
                self.rectangle.set_height(h0-dy)
            elif abs(y0 + h0 - yp) < bt * h0:
                self.rectangle.set_height(h0+dy)
        else:
            self.rectangle.set_x(x0+dx)
            self.rectangle.set_y(y0+dy)

class NXpolygon(NXpatch):

    def __init__(self, xy, closed=True, plotview=None, **opts):
        shape = Polygon(xy, closed, **opts)
        if 'linewidth' not in opts:
            shape.set_linewidth(1.0)
        if 'color' not in opts and 'facecolor' not in opts:
            shape.set_facecolor('r')
        super(NXpolygon, self).__init__(shape, resize=False, plotview=plotview)
        self.shape.set_label('Polygon')
        self.polygon = self.shape

    def __repr__(self):
        xy = self.polygon.xy
        v = xy.shape[0] - 1
        return 'NXpolygon(%g, %g, vertices=%s)' % (xy[0][0], xy[0][1], v)

    @property
    def xy(self):
        return self.polygon.xy

    def initialize(self, xp, yp):
        xy0 = self.polygon.xy
        return xy0, xp, yp

    def update(self, x, y):
        xy0, xp, yp = self.press
        dxy = (x-xp, y-yp)
        self.polygon.set_xy(xy0+dxy)

max_stars_repo_name: nexpy/nexpy

max_issues_repo_name: nexpy/nexpy

max_forks_repo_name: nexpy/nexpy