lasprs 0.14.2

use super::*;
use crate::{
    daq::daqconfig::CouplingMode,
    measurement::{CPSThread, MeasurementType, MeasurementWriter, SharedMeasurement},
    ps::CPSSettings,
    rt::SimpleClipDetector,
    siggen::SourceDescriptor,
    *,
};
use anyhow::{Error, Result, anyhow, bail};
use crossbeam::{atomic::AtomicCell, channel::Receiver};
use hdf5_metno::{
    Dataset, File, H5Type, dataset, datatype,
    types::{VarLenArray, VarLenUnicode},
};
use parking_lot::{Mutex, RwLock};
use reinterpret::reinterpret_slice;
use std::{
    borrow::Cow,
    fmt::Display,
    path::{Path, PathBuf},
    str::FromStr,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering::Relaxed},
    },
    time::Duration,
};
use streamdata::*;
use streammgr::*;
use streammsg::{InStreamMsg, PreCaptureBuffer};
use strum::EnumMessage;
use triple_buffer::triple_buffer;
use uuid::Uuid;

/// The stabilization time for IEPE sensors. The stream should run at least for
/// this time, before a recording can be started. That is, when IEPE is enabled
/// for some sensors in the stream.
const IEPE_STABILIZATION_TIME: Duration = Duration::from_millis(6000);

#[cfg_attr(
    feature = "python-bindings",
    gen_stub_pyclass,
    pyclass(get_all, from_py_object)
)]
#[derive(Clone, Debug)]
/// Finished recording
pub struct FinishedRecording {
    /// Whether there occured a signal clipped during the recording.
    pub clipped: bool,

    /// Possibly, the recording finished, but an error happened, due to
    /// which the data is invalid.
    pub error: Option<String>,

    /// The measurement that was recorded. If the user pressed cancel, this is
    /// None. If an error occured, this is also None.
    pub measurement: Option<SharedMeasurement>,
}
impl PartialEq for FinishedRecording {
    fn eq(&self, other: &Self) -> bool {
        self.clipped == other.clipped && self.error == other.error
    }
}

/// Status of a recording
#[cfg_attr(
    feature = "python-bindings",
    gen_stub_pyclass_complex_enum,
    pyclass(get_all, eq, from_py_object)
)]
#[derive(Clone, Debug, PartialEq)]
pub enum RecordStatus {
    /// Not yet started, waiting for first msg
    Idle {},

    /// Waiting for start delay to be processed.
    Waiting {},

    /// Recording in progress
    Recording {
        /// Whether a clip has already happened during recording
        clipped: bool,
        /// The amount of time that is currently recorded
        recorded: Flt,
        /// The percentage done [0 to 100%]. Stays at 0% for infinite duration
        /// recordings
        pct_done: Flt,
    },

    /// Recording finished
    Finished {
        /// Data describing the finished recording.
        finishedrecording: FinishedRecording,
    },
}

/// Settings used to start a recording.
#[derive(Clone)]
#[cfg_attr(
    feature = "python-bindings",
    gen_stub_pyclass,
    pyclass(get_all, from_py_object)
)]
pub struct RecordSettings {
    /// Full file path and file name of the file (including extension).
    filepath: PathBuf,

    /// The recording time. Set to None to perform indefinite recording
    pub duration: Option<Duration>,

    /// The delay to wait before adding data
    pub startDelay: Option<Duration>,

    /// Settings for computing Cross-Power Spectra during the recording.
    pub cps_settings: Option<CPSSettings>,

    /// Measurement type
    pub measurementType: MeasurementType,

    /// When this flag is set, the signal generator is controlled automatically
    /// by the recording process. Before starting, the signal generator is
    /// unmuted, and after finishing, it is muted again.
    pub record_controls_siggen_mute: bool,

    /// List of reference measurement UUID's, that are 'some' kind of reference
    /// for the new measurement.
    pub reference_meas: Vec<SharedMeasurement>,
}

impl RecordSettings {
    /// Start a new recording with simple settings.
    ///
    /// # Arguments
    ///
    /// - `filename`: Name of file to record to - `duration`: How long recording
    ///   should be. None means record indefinitely.
    pub fn new_simple<U>(name: &str, duration: Option<U>) -> Result<RecordSettings>
    where
        U: Into<Duration> + Default,
    {
        RecordSettings::new(name, false, duration, None, None, None, false, None, vec![])
    }

    /// Create new record settings. Convenience wrapper to fill in fields in
    /// right form. Start delay is optional
    ///
    /// * args:
    ///
    /// - `filename`: Name of file to record to
    /// - `add_counter`: Add  a number to the file to make it unique
    /// - `duration`: How long recording should be. Zero means record indefinitely.
    /// - `startDelay`: Optional start delay.
    /// - `cps_settings`: Optional settings for computing Cross-Power Spectra
    /// - `measurementType`: Optional measurement type. Sets to generic if not provided.
    /// - `directory`: Directory for recording files, if None, the current
    /// - `reference_meas`: List of reference measurements to use for this recording.
    ///   working directory is used.
    ///
    ///
    // TODO: Create a builder pattern for this
    #[expect(clippy::too_many_arguments)]
    pub fn new<U>(
        name: &str,
        add_counter: bool,
        duration: Option<U>,
        startDelay: Option<U>,
        cps_settings: Option<CPSSettings>,
        measurementType: Option<MeasurementType>,
        record_controls_siggen_mute: bool,
        directory: Option<&Path>,
        reference_meas: Vec<SharedMeasurement>,
    ) -> Result<RecordSettings>
    where
        U: Into<Duration> + Default,
    {
        checkValidMeasurementName(name)?;

        // Create unique filename, check if directory exists and is writable, or
        // errors when file already exists
        let filepath = {
            let dir: PathBuf = directory
                .map(|d| d.to_owned())
                .unwrap_or_else(|| PathBuf::from("."));
            if !dir.exists() {
                bail!("Directory does not exist: {}", dir.display());
            }
            if !dir.is_dir() {
                bail!("{} is not a directory.", dir.display());
            }
            // Check if able to write inside directory
            let res = std::fs::metadata(&dir)?;
            if res.permissions().readonly() {
                bail!(
                    "Cannot create new measurement. The storage directory {} is read-only.",
                    dir.display()
                );
            }

            let mut filepath: PathBuf = dir.clone();
            filepath.push(name);
            filepath.set_extension(MEASUREMENT_EXTENSION);

            let name = measurementName(&filepath)?.to_string();

            if filepath.exists() {
                if !add_counter {
                    bail!(
                        "Measurement with name '{}' already exists. \
                         If names are to be identical, a suggenstion \
                         is to add a counter to the measurement names",
                        name
                    );
                } else {
                    let mut counter = 1;
                    'findnameloop: loop {
                        filepath.clear();
                        filepath.push(&dir);
                        let fn_numbered = format!("{name}_{counter:.2}");
                        filepath.push(&fn_numbered);
                        filepath.set_extension(MEASUREMENT_EXTENSION);
                        if !filepath.exists() {
                            break 'findnameloop;
                        }
                        if counter == 99 {
                            bail!(
                                "Maximum number of measurement files reached \
                                by adding counter. Please use a new and unique \
                                measurement name."
                            )
                        }
                        counter += 1;
                    }
                }
            }
            filepath
        };

        Ok(RecordSettings {
            filepath,
            duration: duration.map(|d| d.into()),
            startDelay: startDelay.map(|d| d.into()),
            cps_settings,
            measurementType: measurementType.unwrap_or_default(),
            record_controls_siggen_mute,
            reference_meas,
        })
    }

    /// Returns the name of the measurement, this is the file name only without the
    /// extension.
    pub fn measurementName(&self) -> String {
        self.filepath
            .file_name()
            .unwrap()
            .to_string_lossy()
            .into_owned()
    }
}

#[cfg(feature = "python-bindings")]
#[cfg_attr(feature = "python-bindings", gen_stub_pymethods, pymethods)]
impl RecordSettings {
    // I do not know why, but adding this signature results in compilation errors for the stub generation.
    // #[pyo3(signature=(name,add_counter=false, directory=None, duration=None, startDelay=None, cps_settings=None, measurementType=None, record_controls_siggen_mute=None, reference_meas=None))]
    #[allow(clippy::too_many_arguments)]
    #[new]
    fn new_py(
        name: &str,
        add_counter: bool,
        directory: Option<String>,
        duration: Option<Flt>,
        startDelay: Option<Flt>,
        cps_settings: Option<CPSSettings>,
        measurementType: Option<MeasurementType>,
        record_controls_siggen_mute: Option<bool>,
        reference_meas: Option<Vec<SharedMeasurement>>,
    ) -> PyResult<Self> {
        let startDelay = if let Some(delay) = startDelay {
            let delay = Duration::try_from_secs_f64(delay)
                .map_err(|e| anyhow! {"Invalid start delay specified: {e}"})?;
            Some(delay)
        } else {
            None
        };
        let record_controls_siggen_mute = record_controls_siggen_mute.unwrap_or(false);
        let duration = if let Some(duration) = duration {
            Some(
                Duration::try_from_secs_f64(duration)
                    .map_err(|e| anyhow! {"Invalid duration specified: {e}"})?,
            )
        } else {
            None
        };

        let dir = directory.as_ref().map(|d| Path::new(d.as_str()));
        let reference_meas = reference_meas.unwrap_or_default();
        Ok(RecordSettings::new(
            name,
            add_counter,
            duration,
            startDelay,
            cps_settings,
            measurementType,
            record_controls_siggen_mute,
            dir,
            reference_meas,
        )?)
    }

    /// Returns the name of the measurement - to be recorded.
    #[pyo3(name = "measurementName")]
    fn measurementName_py(&self) -> String {
        self.measurementName()
    }
}

/// This struct lets a recording run on a stream, waits till the first data arrives and records for a given period of time.
///
/// See [`RecordSettings`] for configuration options.

#[cfg_attr(feature = "python-bindings", gen_stub_pyclass, pyclass)]
pub struct Recording {
    // Recording settings
    settings: RecordSettings,

    // Stop the recording. This stops the thread
    stopThread: Arc<AtomicBool>,

    // Obtain status from thread.
    status: triple_buffer::Output<RecordStatus>,
}

impl Recording {
    /// Start a new recording
    ///
    /// # Arguments
    ///
    /// * setttings: The settings to use for the recording
    /// * smgr: Stream manager to use to start the recording
    ///
    pub fn new(settings: RecordSettings, mgr: &mut StreamMgr) -> Result<Recording> {
        let status = mgr.getStatus(StreamDirection::Input);
        match status {
            StreamStatus::NotRunning {} => bail!("Cannot record: no input stream is running."),
            StreamStatus::Running { start_time } => {
                // Check for IEPE stabilization
                let meta = mgr
                    .getStreamMetaData(StreamDirection::Input)
                    .expect("Stream is running, but no metadata available");
                let hasIEPE = meta
                    .channelInfo
                    .iter()
                    .any(|info| info.couplingMode == CouplingMode::ACWithIEPE);
                if hasIEPE && start_time.elapsed()? <= IEPE_STABILIZATION_TIME {
                    bail!(
                        "Cannot record yet: IEPE stabilization not complete. Please wait 1 minute after starting the stream before measuring."
                    );
                }
            }
            StreamStatus::Error { e } => bail!("Cannot record: {}", e),
        }

        // Detector for signal clipping
        let clipdetector = SimpleClipDetector::new(mgr);

        let stopThread = Arc::new(AtomicBool::new(false));
        let stopThread_clone = stopThread.clone();

        let settings_clone = settings.clone();

        let (mut status_tx, status_rx) = triple_buffer(&RecordStatus::Idle {});

        // Create new receiving channel.
        let rx = StreamHandler::new(mgr);
        // Source descriptor, if output stream is running
        let source = if mgr.isStreamRunningOK(StreamDirection::Output) {
            mgr.getSourceDescriptor()
        } else {
            SourceDescriptor::Silence {}
        };

        // The thread doing the actual work
        spawn(
            move || {
                let mut recording_closure = || {
                    // Handle to CPS thread
                    let cpsthread;
                    let meas;

                    // Start of scope in which the HDF5 file is open
                    {
                        let writer = MeasurementWriter::new(&settings.filepath)?;
                        let firstmsg = match rx.recv() {
                            Ok(msg) => msg,
                            Err(_) => bail!("Queue handle error"),
                        };

                        let (meta, precap_status) = match firstmsg {
                            InStreamMsg::StreamStarted(meta, precap) => (meta, precap),
                            _ => bail!("Recording failed. Missed stream metadata message."),
                        };

                        // Initialize the CPS thread if CPS settings are provided
                        cpsthread = settings
                            .cps_settings
                            .map(|settings| CPSThread::new(meta.samplerate, settings, None));

                        // Write measurement metadata to HDF5 file
                        let meas_meta = MeasurementMetadata::new_from_recording(
                            &meta,
                            Some(&source),
                            &settings,
                        );

                        // write metadata to HDF5 file
                        let mut writer = writer.write_meta(meas_meta)?;

                        let framesPerBlock = meta.framesPerBlock;
                        let mut wait_block_ctr = 0;

                        // Process pre-capture buffer from the StreamStarted message
                        let requires_precapture =
                            settings.measurementType.requiresPreCapturedData();

                        let mut incoming_ctr: usize = 0;
                        match (precap_status, requires_precapture) {
                            (PreCaptureBuffer::Loaded(precap_buf), true) => {
                                // Write pre-capture data to its own dataset
                                for block in precap_buf.into_iter() {
                                    let raw = block.getRaw().expect("No raw stream data");
                                    if writer.is_empty() {
                                        incoming_ctr = block.ctr;
                                    } else {
                                        incoming_ctr += 1;
                                    }
                                    writer.write_precapture(&raw)?;
                                    if block.ctr != incoming_ctr {
                                        eprintln!(
                                            "********** PRECAPTURE PACKAGES MISSED ***********"
                                        );
                                        bail!("Pre-capture data blocks missed.");
                                    }
                                }
                            }
                            (PreCaptureBuffer::NotLoaded, true) => {
                                bail!(
                                    "This measurement type requires pre-captured \
                                         data, but the stream has not been running \
                                         long enough. Please wait for at least {} seconds after starting the stream.",
                                    PRECAP_DURATION.as_secs(),
                                );
                            }
                            (_, false) => {
                                // Simply ignore the precapture buffer, we do not need it.
                            }
                        }

                        // Indicate we are ready to record!
                        if let Some(delay) = settings.startDelay {
                            status_tx.write(RecordStatus::Waiting {});
                            let startdelay_s = delay.as_micros() as Flt / 1e6;
                            wait_block_ctr =
                                (*meta.samplerate * startdelay_s / framesPerBlock as Flt) as u32;
                        } else {
                            status_tx.write(RecordStatus::Recording {
                                clipped: clipdetector.hasClipped(),
                                recorded: 0.,
                                pct_done: 0.,
                            });
                        }

                        'recloop: loop {
                            if stopThread.load(Relaxed) {
                                break 'recloop;
                            }
                            match rx.recv().unwrap() {
                                InStreamMsg::StreamError(e) => {
                                    bail!("Recording failed due to stream error: {}.", e)
                                }
                                InStreamMsg::StreamStarted(_, _) => {
                                    bail!("Stream started again.")
                                }
                                InStreamMsg::StreamStopped => {
                                    // Early stop. User stopped it.
                                    break 'recloop;
                                }
                                InStreamMsg::InStreamData(instreamdata) => {
                                    if wait_block_ctr > 0 {
                                        // We are still waiting — track ctr for continuity
                                        incoming_ctr = instreamdata.ctr;
                                        wait_block_ctr -= 1;
                                        if wait_block_ctr == 0 {
                                            status_tx.write(RecordStatus::Recording {
                                                clipped: clipdetector.hasClipped(),
                                                recorded: 0.,
                                                pct_done: 0.,
                                            });
                                        }
                                        continue 'recloop;
                                    }

                                    let raw = instreamdata.getRaw().expect("No raw stream data");
                                    if writer.is_empty() {
                                        // Initialize counter offset (no pre-capture was written)
                                        incoming_ctr = instreamdata.ctr;
                                    } else {
                                        incoming_ctr += 1;
                                    }
                                    writer.write(&raw)?;
                                    if instreamdata.ctr != incoming_ctr {
                                        eprintln!("********** PACKAGES MISSED ***********");
                                        bail!("Stream data blocks missed. Recording is invalid.")
                                    }
                                    if let Some(thread) = &cpsthread {
                                        // Do not apply sensitivity
                                        let fd = instreamdata.getFloatData(false);
                                        thread.push(fd);
                                    }

                                    // Total recorded time includes pre-capture frames
                                    let recorded_time = Duration::from_millis(
                                        ((1000 * writer.frames_recorded()) as Flt
                                            / *meta.samplerate)
                                            as u64,
                                    );

                                    if let Some(duration) = settings.duration
                                        && recorded_time >= duration
                                    {
                                        break 'recloop;
                                    }
                                    status_tx.write(RecordStatus::Recording {
                                        clipped: clipdetector.hasClipped(),
                                        recorded: recorded_time.as_secs_f64() as Flt,
                                        pct_done: if let Some(duration) = settings.duration {
                                            100. * recorded_time.as_millis() as Flt
                                                / duration.as_millis() as Flt
                                        } else {
                                            0.
                                        },
                                    });
                                }
                            }
                        } // end of 'recloop

                        if writer.is_empty() {
                            bail!("Recording stopped before any data is stored.");
                        }
                        let clipped = clipdetector.hasClipped();
                        meas = writer.finish(clipped)?;
                    } // End of scope in which the file is open for writing.

                    if let Some(thread) = cpsthread {
                        // Add thread to list of cpsthreads in measurement
                        thread.stopHere();
                        meas.write().add_cpsthread(thread);
                    }

                    Ok(meas)
                };
                match recording_closure() {
                    Err(e) => {
                        status_tx.write(RecordStatus::Finished {
                            finishedrecording: FinishedRecording {
                                clipped: clipdetector.hasClipped(),
                                error: Some(format!("{e}")),
                                measurement: None,
                            },
                        });
                    }
                    Ok(measurement) => {
                        status_tx.write(RecordStatus::Finished {
                            finishedrecording: FinishedRecording {
                                clipped: clipdetector.hasClipped(),
                                error: None,
                                measurement: Some(measurement),
                            },
                        });
                    } // End of thread
                }
            },
            ThreadPriority::Low,
        );

        Ok(Recording {
            settings: settings_clone,
            stopThread: stopThread_clone,
            status: status_rx,
        })
    }

    // Delete recording file, should be done when something went wrong (an error
    // occured), or when cancel() is called, or when recording object is dropped
    // while thread is still running.
    fn deleteFile(&self) {
        // File should not be un use anymore, as thread is joined.
        // In case of error, we try to delete the file
        if let Err(e) = std::fs::remove_file(&self.settings.filepath) {
            eprintln!("Recording failed, but file removal failed as well: {e}");
        }
    }

    /// Returns the measurement associated with this recording, if it has
    /// finished, and no error occured. If anything happened during recording,
    /// None is returned. Please use getStatus() to check if recording was
    /// successful.
    pub fn getMeasurement(&mut self) -> Option<SharedMeasurement> {
        if let RecordStatus::Finished { finishedrecording } = self.status() {
            finishedrecording.measurement
        } else {
            None
        }
    }
}

#[cfg_attr(feature = "python-bindings", gen_stub_pymethods, pymethods)]
impl Recording {
    #[cfg(feature = "python-bindings")]
    #[gen_stub(skip)]
    #[new]
    fn new_py(settings: RecordSettings, mgr: &mut StreamMgr) -> PyResult<Recording> {
        Ok(Recording::new(settings, mgr)?)
    }

    /// Get current record status
    pub fn status(&mut self) -> RecordStatus {
        self.status.read().clone()
    }

    #[cfg(feature = "python-bindings")]
    #[pyo3(name = "getMeasurement")]
    fn getMeasurement_py(&mut self) -> Option<SharedMeasurement> {
        if let RecordStatus::Finished { finishedrecording } = self.status() {
            finishedrecording.measurement
        } else {
            None
        }
    }

    /// Stop existing recording early.
    pub fn stop(&mut self) {
        // Stop thread and wait for message of finished
        self.stopThread.store(true, Relaxed);
        self.waitForThread();

        let RecordStatus::Finished {
            finishedrecording: FinishedRecording { error, .. },
        } = self.status()
        else {
            panic!("RecordStatus should be finished!");
        };
        if error.is_some() {
            // Delete file if there was an error
            self.deleteFile();
        }
    }

    /// Cancel recording. Deletes the recording file
    pub fn cancel(&mut self) {
        self.stopThread.store(true, Relaxed);
        self.waitForThread();
        self.deleteFile();
    }
    fn waitForThread(&mut self) {
        while !matches!(self.status(), RecordStatus::Finished { .. }) {
            std::thread::sleep(Duration::from_millis(10));
        }
    }
}

cfg_select! {
    any(feature = "python-bindings") => {
        use pyo3_stub_gen::type_info::{MemberInfo, PyClassInfo};
        pyo3_stub_gen::inventory::submit! {
            gen_methods_from_python! {
            r#"
            class Recording:
                def __init__(self, settings: RecordSettings, mgr: StreamMgr):
                        """ Initialize a Recording
                        """
            "#
            }
        }
    }
    _ => {}
}

impl Drop for Recording {
    fn drop(&mut self) {
        // If we enter here, stop() or cancel() has not been called. In that
        // case, we cleanup here by cancelling the recording and deleting the file.
        if !matches!(self.status(), RecordStatus::Finished { .. }) {
            self.stopThread.store(true, Relaxed);
            self.waitForThread();
            self.deleteFile();
        }
    }
}