Communication Protocol

def abort(self):
    """Abort any currently running script or measurement and wait for completion.

    This method sends the abort signal to the instrument. If a script was
    still running, it will wait for it to complete.
    Note that this could take a while, depending on the measurement that
    was running.
    """
    _ = self.flush()

    try:
        response = self.query('Z')
    except CommProtocolError as exc:
        if exc.error_code != '0006':
            raise

        time.sleep(0.1)
    else:
        if response == 'Z\n':
            _ = self.read_until('\n\n')

def close(self) -> None:
    """Close device connection."""
    self._device.Close()

def flush(self):
    """Send a blank line to the device and read its response.

    This does not modify the write buffer. It sends
    an empty command and reads whatever response follows.

    Returns
    -------
    str
        The response from the device after sending `'\\n'`.
    """
    self.write('\n')
    return self.read_until('\n')

def get_communication_capabilities(self) -> tuple[str, ...]:
    """Retrieve which communication commands are available on this instrument.

    Returns a set of commands part of the communication protocol that
    are supported by the device's firmware.

    Returns
    -------
    tuple[str, ...]
        Tuple of supported commands for the instrument.
    """
    response = self.query('CC')
    return parse_capabilities(response, mapping=COMMUNICATION_CAPABILITIES)

def get_methodscript_capabilities(self) -> tuple[str, ...]:
    """Retrieve which MethodSCRIPT features are available on this instrument.

    Returns a set of command names that are licensed
    and supported by the connected instrument's hardware and firmware.

    Returns
    -------
    tuple[str, ...]
        Tuple of available MethodSCRIPT command names.
    """

    response = self.query('CM')
    return parse_capabilities(response, mapping=METHODSCRIPT_CAPABILITIES)

def lines(
    self,
    timeout: float | None = None,
    delay: float | None = None,
) -> Generator[str, None, None]:
    """Yield response chunks until a timeout occurs or no more data arrives.

    This is a generator that continuously reads from the device buffer,
    yielding chunks as they arrive.
    It stops when the elapsed time between responses exceeds `timeout`.

    Parameters
    ----------
    timeout : float, optional
        Maximum total time (in seconds) between responses.
        Defaults to `self.timeout`.
    delay : float, optional
        Pause (in seconds) between read attempts when no data is available.
        Defaults to `self.delay`.

    Yields
    ------
    response : str
        Response chunks as they arrive from the device.
    """
    delay = delay or self.delay
    timeout = timeout or self.timeout

    deadline = time.monotonic() + timeout

    while True:
        response = self.read()

        if response:
            match = ERROR_PATTERN.match(response)

            if match:
                error_code = match.group(1) + match.group(2).strip()
                raise CommProtocolError(error_code=error_code)

            yield response
            deadline = time.monotonic() + timeout

        if time.monotonic() > deadline:
            raise TimeoutError('Timed out waiting for response')

        if not response:
            time.sleep(delay)

def open(self) -> None:
    """Open device connection."""
    self._device.Open()

def query(
    self,
    command: str,
    end: str | None = None,
    delay: float | None = None,
) -> str:
    """Send a command and return its response in a single call.

    This is the primary method for interactive communication with the
    instrument. It writes the command to the device, waits for completion,
    reads the full response, and returns it as a string.

    For commands that run for a long time (e.g. scripts), this
    method will block until the script completes or times out.

    Parameters
    ----------
    command : str
        The command to send (e.g., 'i' to get the serial number).
        If `command` does not end with `'\\n'`, one is automatically added.
    end : str, optional
        The termination character(s) that mark the end of the response.

        If `None`, a lookup table determines the appropriate terminator
        based on the command. Most commands use `'\\n'`. Commands with
        variable-length responses (e.g. scripts) use `'\\n\\n'`.
    delay : float, optional
        Pause (in seconds) between read attempts. Defaults to `self.delay`.

    Returns
    -------
    response : str
        The complete response from the device.
    """
    delay = delay or self.delay

    if not end:
        try:
            func, *_ = command.split(maxsplit=1)
        except ValueError:
            func = ''

        end = NEWLINE_TERMINATORS.get(func, '\n')

    if not command.endswith('\n'):
        command = f'{command}\n'

    self.write(command)

    prefix = command[0]

    response = self.wait_until(prefix)

    if not response.endswith(end):
        response += self.read_until(end=end, delay=delay)

    return response.removeprefix(prefix).removeprefix('\n').removesuffix('\n')

def read(self) -> str:
    """Read the next available chunk from the instrument's input buffer.

    This method does not block. It returns immediately with whatever
    data is currently available in the buffer.

    Returns
    -------
    response : str
        The next response chunk, or an empty string ('') if the buffer
        contains no data. Each read is recorded in `self.history` for
        debugging and inspection.
    """
    response = self._device.Read()

    if response:
        self.history.append(response)

    return response

def read_until(
    self,
    end: str = '\n',
    delay: float | None = None,
) -> str:
    """Read lines from the device until a termination sequence is found.

    Continuously reads responses and concatenates them until
    `end` appears.

    Parameters
    ----------
    end : str
        The termination sequence that marks the end of the response.
        Most commands use '\n'. Scripts and variable-length responses
        typically use '\n\n'.
    delay : float, optional
        Pause (in seconds) between read attempts. Defaults to `self.delay`.

    Returns
    -------
    response : str
        Accumulated response up to and including the termination sequence.

    Raises
    ------
    MethodScriptRuntimeError
        If the device returns an error response during reading.
    """
    buffer: list[str] = []

    for line in self.lines(delay=delay):
        if line:
            buffer.append(line)

        if line.endswith(end):
            break

    response = ''.join(buffer)

    return response

def run_methodscript(self, script: str) -> str:
    """Load and execute a MethodSCRIPT on the instrument.

    MethodSCRIPTs are scripts that run directly on the PalmSens device,
    This method uploads the script, runs it to completion,
    and returns any output produced by the script.

    See the MethodSCRIPT documentation for more information.

    Parameters
    ----------
    script : str
        The MethodSCRIPT to run. The entire script must end
        with exactly one newline ('\n').

    Returns
    -------
    str
        Output produced by the running script, if any.
    """
    script = script.rstrip('\n')

    return self.query(f'e\n{script}\n\n')

def wait_until(self, prefix: str, timeout: float | None = None) -> str:
    """Wait until a response line starting with `prefix` arrives.

    When you send a command, the device echoes back its first character
    before returning the actual result. Use this method to wait for
    that echo.

    Parameters
    ----------
    prefix : str
        The first character of an expected response line. This usually
        matches the command abbreviation (e.g., 'E' for a measurement start).
    timeout : float, optional
        Maximum time (in seconds) to wait before raising `TimeoutError`.
        Defaults to `self.timeout`.

    Returns
    -------
    response : str
        The response chunk that starts with `prefix`, including
        its termination character.

    Raises
    ------
    TimeoutError
        If no matching line arrives within the timeout period.
    """
    timeout = timeout or self.timeout

    deadline = time.monotonic() + timeout

    for response in self.lines():
        if response.startswith(prefix):
            return response

        if time.monotonic() > deadline:
            break

    raise TimeoutError(f'Timed out waiting for response starting with {prefix!r}')

def write(self, data: str):
    """Write a command or data to the instrument.

    Parameters
    ----------
    data : str
        Command or data to send. To submit a command for execution,
        append a newline character (`'\\n'`) to the end of the string.
    """
    self._device.Write(data)

async def abort(self):
    """Abort any currently running script or measurement and wait for completion.

    This method sends the abort signal to the instrument. If a script was
    still running, it will wait for it to complete.
    Note that this could take a while, depending on the measurement that
    was running.
    """
    _ = await self.flush()

    try:
        response = await self.query('Z')
    except CommProtocolError as exc:
        if exc.error_code != '0006':
            raise

        await asyncio.sleep(0.1)
    else:
        if response == 'Z\n':
            _ = await self.read_until('\n\n')

async def close(self) -> None:
    """Close device connection."""
    await self.instrument._close_async()

async def flush(self):
    """Send a blank line to the device and read its response.

    This does not modify the write buffer. It sends
    an empty command and reads whatever response follows.

    Returns
    -------
    str
        The response from the device after sending `'\\n'`.
    """
    await self.write('\n')
    return await self.read_until('\n')

async def get_communication_capabilities(self) -> tuple[str, ...]:
    """Retrieve which communication commands are available on this instrument.

    Returns a set of commands part of the communication protocol that
    are supported by the device's firmware.

    Returns
    -------
    tuple[str, ...]
        Set of supported commands for the instrument.
    """
    response = await self.query('CC')
    return parse_capabilities(response, mapping=COMMUNICATION_CAPABILITIES)

async def get_methodscript_capabilities(self) -> tuple[str, ...]:
    """Retrieve which MethodSCRIPT features are available on this instrument.

    Returns a set of command names that are licensed
    and supported by the connected instrument's hardware and firmware.

    Returns
    -------
    tuple[str, ...]
        Set of available MethodSCRIPT command names.
    """

    response = await self.query('CM')
    return parse_capabilities(response, mapping=METHODSCRIPT_CAPABILITIES)

async def lines(
    self,
    timeout: float | None = None,
    delay: float | None = None,
) -> AsyncIterator[str]:
    """Yield response chunks until a timeout occurs or no more data arrives.

    This is an async generator that continuously reads from the device buffer,
    yielding chunks as they arrive.
    It stops when the elapsed time between responses exceeds `timeout`.

    Parameters
    ----------
    timeout : float, optional
        Maximum total time (in seconds) between responses.
        Defaults to `self.timeout`.
    delay : float, optional
        Pause (in seconds) between read attempts when no data is available.
        Defaults to `self.delay`.

    Yields
    ------
    response : str
        Response chunks as they arrive from the device.
    """
    delay = delay or self.delay
    timeout = timeout or self.timeout

    deadline = time.monotonic() + timeout

    while True:
        response = await self.read()

        if response:
            match = ERROR_PATTERN.match(response)

            if match:
                error_code = match.group(1) + match.group(2).strip()
                raise CommProtocolError(error_code=error_code)

            yield response
            deadline = time.monotonic() + timeout

        if time.monotonic() > deadline:
            raise TimeoutError('Timed out waiting for response')

        if not response:
            await asyncio.sleep(delay)

async def open(self) -> None:
    """Open device connection."""
    await self.instrument._open_async()

async def query(
    self,
    command: str,
    end: str | None = None,
    delay: float | None = None,
) -> str:
    """Send a command and return its response in a single call.

    This is the primary method for interactive communication with the
    instrument. It writes the command to the device, waits for completion,
    reads the full response, and returns it as a string.

    For commands that run for a long time (e.g. scripts), this
    method will block until the script completes or times out.

    Parameters
    ----------
    command : str
        The command to send (e.g., 'i' to get the serial number).
        If `command` does not end with `'\\n'`, one is automatically added.
    end : str, optional
        The termination character(s) that mark the end of the response.

        If `None`, a lookup table determines the appropriate terminator
        based on the command. Most commands use `'\\n'`. Commands with
        variable-length responses (e.g. scripts) use `'\\n\\n'`.
    delay : float, optional
        Pause (in seconds) between read attempts. Defaults to `self.delay`.

    Returns
    -------
    response : str
        The complete response from the device, including termination characters.
    """
    delay = delay or self.delay

    if not end:
        try:
            func, *_ = command.split(maxsplit=1)
        except ValueError:
            func = ''

        end = NEWLINE_TERMINATORS.get(func, '\n')

    if not command.endswith('\n'):
        command = f'{command}\n'

    await self.write(command)

    prefix = command[0]

    response = await self.wait_until(prefix)

    if not response.endswith(end):
        response += await self.read_until(end=end, delay=delay)

    return response.removeprefix(prefix).removeprefix('\n').removesuffix('\n')

async def read(self) -> str:
    """Read the next available chunk from the instrument's input buffer.

    This method does not block. It returns immediately with whatever
    data is currently available in the buffer.

    Returns
    -------
    response : str
        The next response chunk, or an empty string ('') if the buffer
        contains no data. Each read is recorded in `self.history` for
        debugging and inspection.
    """
    response: str = await create_future(self._device.ReadAsync())

    if response:
        self.history.append(response)

    return response

async def read_until(
    self,
    end: str = '\n',
    delay: float | None = None,
) -> str:
    """Read lines from the device until a termination sequence is found.

    Continuously reads responses and concatenates them until
    `end` appears.

    Parameters
    ----------
    end : str
        The termination sequence that marks the end of the response.
        Most commands use '\n'. Scripts and variable-length responses
        typically use '\n\n'.
    delay : float, optional
        Pause (in seconds) between read attempts. Defaults to `self.delay`.

    Returns
    -------
    response : str
        Accumulated response up to and including the termination sequence.

    Raises
    ------
    MethodScriptRuntimeError
        If the device returns an error response during reading.
    """
    buffer: list[str] = []

    async for line in self.lines(delay=delay):
        if line:
            buffer.append(line)

        if line.endswith(end):
            break

    response = ''.join(buffer)

    return response

async def run_methodscript(self, script: str) -> str:
    """Load and execute a MethodSCRIPT on the instrument.

    MethodSCRIPTs are scripts that run directly on the PalmSens device,
    This method uploads the script, runs it to completion,
    and returns any output produced by the script.

    See the MethodSCRIPT documentation for more information.

    Parameters
    ----------
    script : str
        The MethodSCRIPT to run. The entire script must end
        with exactly one newline ('\n').

    Returns
    -------
    str
        Output produced by the running script, if any.
    """
    script = script.rstrip('\n')

    return await self.query(f'e\n{script}\n\n')

async def wait_until(self, prefix: str, timeout: float | None = None) -> str:
    """Wait until a response line starting with `prefix` arrives.

    When you send a command, the device echoes back its first character
    before returning the actual result. Use this method to wait for
    that echo.

    Parameters
    ----------
    prefix : str
        The first character of an expected response line. This usually
        matches the command abbreviation (e.g., 'E' for a measurement start).
    timeout : float, optional
        Maximum time (in seconds) to wait before raising `TimeoutError`.
        Defaults to `self.timeout`.

    Returns
    -------
    response : str
        The response chunk that starts with `prefix`, including
        its termination character.

    Raises
    ------
    TimeoutError
        If no matching line arrives within the timeout period.
    """
    timeout = timeout or self.timeout

    deadline = time.monotonic() + timeout

    async for response in self.lines():
        if response.startswith(prefix):
            return response

        if time.monotonic() > deadline:
            break

    raise TimeoutError(f'Timed out waiting for response starting with {prefix!r}')

async def write(self, data: str):
    """Write a command or data to the instrument.

    Parameters
    ----------
    data : str
        Command or data to send. To submit a command for execution,
        append a newline character (`'\\n'`) to the end of the string.
    """
    await create_future(self._device.WriteAsync(data))