#include "acoustic_analyzer/io/mobile_phone_source.h"
#include <cstring>
#include <cmath>
#include <chrono>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#include <arpa/inet.h>

namespace acoustic {

MobilePhoneSource::MobilePhoneSource(int port, int sample_rate, float timeout_sec)
    : port_(port), sample_rate_(sample_rate), timeout_sec_(timeout_sec), sockfd_(-1) {}

MobilePhoneSource::~MobilePhoneSource() {
    close();
}

bool MobilePhoneSource::open() {
    sockfd_ = socket(AF_INET, SOCK_DGRAM, 0);
    if (sockfd_ < 0) return false;

    int opt = 1;
    setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port_);
    addr.sin_addr.s_addr = INADDR_ANY;

    if (bind(sockfd_, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) < 0) {
        ::close(sockfd_);
        sockfd_ = -1;
        return false;
    }

    // Set non-blocking
    int flags = fcntl(sockfd_, F_GETFL, 0);
    fcntl(sockfd_, F_SETFL, flags | O_NONBLOCK);

    running_ = true;
    recv_thread_ = std::thread(&MobilePhoneSource::receive_loop, this);
    return true;
}

void MobilePhoneSource::close() {
    running_ = false;
    if (recv_thread_.joinable()) recv_thread_.join();
    if (sockfd_ >= 0) {
        ::close(sockfd_);
        sockfd_ = -1;
    }
}

void MobilePhoneSource::receive_loop() {
    std::vector<float> packet_buf(2048);
    sockaddr_in client_addr{};
    socklen_t addr_len = sizeof(client_addr);
    auto start = std::chrono::steady_clock::now();

    while (running_) {
        ssize_t n = recvfrom(sockfd_, packet_buf.data(),
                             packet_buf.size() * sizeof(float),
                             0, reinterpret_cast<sockaddr*>(&client_addr), &addr_len);
        if (n > 0) {
            size_t samples = n / sizeof(float);
            std::lock_guard<std::mutex> lock(mutex_);
            for (size_t i = 0; i < samples; ++i) {
                buffer_.push(packet_buf[i]);
            }
            cv_.notify_one();
            start = std::chrono::steady_clock::now();
        } else {
            auto now = std::chrono::steady_clock::now();
            float elapsed = std::chrono::duration<float>(now - start).count();
            if (elapsed > timeout_sec_) {
                // Timeout: stop waiting
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        }
    }
}

size_t MobilePhoneSource::read(std::vector<std::vector<float>>& out, size_t max_samples) {
    std::unique_lock<std::mutex> lock(mutex_);
    if (buffer_.empty()) {
        cv_.wait_for(lock, std::chrono::milliseconds(100));
    }

    size_t available = std::min(max_samples, buffer_.size());
    out.resize(1);
    out[0].resize(available);
    for (size_t i = 0; i < available; ++i) {
        out[0][i] = buffer_.front();
        buffer_.pop();
    }
    return available;
}

} // namespace acoustic
