@classmethod
async def create(cls) -> Beamline:
    """
    Create and connect a Beamline from environment variables.

    Reads BCS_SERVER_ADDRESS and BCS_SERVER_PORT from the environment or
    a .env file, creates a BCSz server, and connects.

    Returns
    -------
    Beamline
        A connected, ready-to-use Beamline instance.

    Raises
    ------
    ConnectionError
        If the BCS server is unreachable or connection fails.
    """
    config = Connection()
    server = BCSz.BCSServer()
    buff = io.StringIO()
    with redirect_stdout(buff):
        await server.connect(**config.model_dump())
    return cls(server)

async def scan_from_dataframe(
    self,
    df: pd.DataFrame,
    ai_channels: list[str] | None = None,
    default_delay: float = 0.1,
    shutter: str = "Shutter Output",
    motor_timeout: float = 30.0,
    progress: bool = True,
    actuate_every: bool = False,
    writer: RunWriter | None = None,
    with_detector: bool = False,
) -> pd.DataFrame:
    """
    Execute a scan defined by a DataFrame.

    Each row defines one scan point: motor columns set motor positions,
    and an optional exposure column sets per-point acquisition time.

    Parameters
    ----------
    df : pd.DataFrame
        Scan definition. Motor columns must match valid motor names.
        An optional "exposure" (or "exp", "count_time") column sets
        per-point acquisition time.
    ai_channels : list[str] or None, optional
        AI channels to acquire at each point. If None, uses
        ["Photodiode", "TEY signal", "AI 3 Izero"].
    default_delay : float, optional
        Settle delay after each motor move in seconds (default: 0.1).
    shutter : str, optional
        DIO channel name for the shutter (default: "Shutter Output").
    motor_timeout : float, optional
        Maximum wait time for motor moves in seconds (default: 30.0).
    progress : bool, optional
        Show a tqdm progress bar (default: True).
    actuate_every : bool, optional
        If True, open/close the shutter per point. If False (default),
        open the shutter once for the entire scan.
    writer : RunWriter or None, optional
        If provided, scan data are persisted to the beamtime SQLite database
        via the writer. The caller is responsible for constructing and
        opening the writer before passing it here.
    with_detector : bool, optional
        If True, acquire a 2D detector image at each scan point using the
        beamline's AreaDetector. Requires a writer for image persistence.
        Shutter actuation is hardware-driven (default: False).

    Returns
    -------
    pd.DataFrame
        Results with columns: motor_position, channel_mean, channel_std,
        exposure, timestamp per row.

    Notes
    -----
    To abort a running scan from another Jupyter cell, call
    ``await bl.abort_scan()`` while the scan task is running.
    """
    scan_plan = ScanPlan.from_dataframe(
        df,
        ai_channels=ai_channels,
        default_delay=default_delay,
        shutter=shutter,
        actuate_every=actuate_every,
    )
    return await self._executor.execute_scan(
        scan_plan,
        progress=progress,
        writer=writer,
        detector=self.detector if with_detector else None,
    )

async def abort_scan(self) -> None:
    """
    Request an abort of the currently running scan.

    Sets the internal abort flag. The scan stops after the current
    point completes and returns partial results.

    Notes
    -----
    Safe to call even when no scan is running (no-op).
    From Jupyter: run the scan as an asyncio.Task and call this
    method from another cell while it executes.
    Programmatic: call from any async context.
    """
    await self._executor.abort()