canboat 9.1.0

// (C) 2009-2026, Kees Verruijt, Harlingen, The Netherlands.

//! `canboat convert` — translate a capture between formats.
//!
//! Every supported input (the canboat ASCII line formats, plus the
//! binary `.pcap`/`.pcap.gz`/`.nif` capture containers) is normalised
//! to a [`RawFrame`](crate::engine::RawFrame) stream, then emitted as:
//!
//! - `plain` — canboat PLAIN/FAST lines (a pure frame→frame shunt via
//!   [`crate::io::copy`]); subsumes `nif2analyzer` and format
//!   normalisation.
//! - `json` / `text` — fully decoded records via the analyzer decode
//!   pipeline; this is what the `analyzer` binary does.

use std::fs::File;
use std::io::{self, BufRead, BufReader, Write};
use std::path::{Path, PathBuf};

use anyhow::{Context, Result};

use crate::engine::format::{Framing, InputFormat};
use crate::engine::output::{GeoFormat, JsonOptions, TextOptions, write_json, write_text};
use crate::engine::{BusProtocol, FramePacketType, RawFrame, Reassembled, Reassembler};
use crate::io::{
    BstD0Reader, BstReader, EblReader, EblWriter, FrameReader, FrameWriter, LineFrameReader,
    MxpgnWriter, PlainWriter, TextLineWriter, UsbCanAReader, analyze, container, copy,
};

/// Output format for `convert --to`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)]
enum OutFormat {
    /// canboat PLAIN/FAST lines (raw frames, no decode).
    Plain,
    /// One decoded JSON object per record.
    Json,
    /// canboat human-readable text, one line per record.
    Text,
    /// RAW ASCII (Actisense's name; the Yacht Devices RAW format): one raw
    /// CAN frame per line, `<time> R <canid> <hex>…`.
    #[value(name = "raw-ascii", alias = "ydwg02")]
    Ydwg02,
    /// Actisense N2K ASCII: one reassembled message per line, `A<time> …`.
    #[value(name = "n2k-ascii", alias = "actisense")]
    Actisense,
    /// Actisense `.ebl` binary log (raw frames).
    #[value(name = "actisense-ebl", alias = "ebl")]
    ActisenseEbl,
    /// ShipModul MiniPlex `$MXPGN` sentences, one CAN frame each.
    Mxpgn,
    /// `$MXPGN` sentences with the data in bus order, as an Actisense
    /// gateway sends them.
    #[value(name = "actisense-mxpgn")]
    ActisenseMxpgn,
    /// SeaSmart `$PCDIN` sentences, one whole message each.
    #[value(name = "seasmart")]
    SeaSmart,
}

impl OutFormat {
    /// Formats that re-encode raw frames (no PGN decode). Everything
    /// else (`json`/`text`) runs the analyzer decode pipeline.
    fn is_frame_level(self) -> bool {
        matches!(
            self,
            OutFormat::Plain
                | OutFormat::Ydwg02
                | OutFormat::Actisense
                | OutFormat::ActisenseEbl
                | OutFormat::Mxpgn
                | OutFormat::ActisenseMxpgn
                | OutFormat::SeaSmart
        )
    }
}

/// Input line format for `convert --from`. When omitted, the format is
/// auto-detected from the first line. Mirrors the ASCII line formats
/// canboat itself recognises.
#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)]
enum FromFormat {
    /// canboat PLAIN / FAST — `<ts>,<prio>,<pgn>,<src>,<dst>,<len>,<hex>…`.
    #[value(name = "plain", alias = "fast")]
    Plain,
    /// PLAIN with single frames interleaved among coalesced fast-packets.
    #[value(name = "plain-mix-fast")]
    PlainMixFast,
    /// Actisense N2K ASCII: one reassembled message per line (`A<time> …`),
    /// from a W2K-1 or PRO-NDC-1E2K in N2K ASCII mode or Actisense's tools.
    #[value(name = "n2k-ascii", alias = "actisense", alias = "actisense-ascii")]
    Actisense,
    /// RAW ASCII: one raw CAN frame per line (`<time> R/T <canid> <hex>…`),
    /// from a Yacht Devices YDWG-02 / YDEN-02 / YDNU-02, or a W2K-1 or
    /// PRO-NDC-1E2K in RAW ASCII mode.
    #[value(name = "raw-ascii", alias = "ydwg02", alias = "yden")]
    Ydwg02,
    /// Digital Yacht iKonvert (`!PDGY,<pgn>,…`).
    #[value(name = "ikonvert")]
    Ikonvert,
    /// Analyzer JSON records, re-encoded to wire frames (camelCase and
    /// spaced-name shapes; `-nv` values are written back verbatim).
    #[value(name = "json")]
    Json,
    /// Airmar (`<ts> - <pgn> <canid> <hex>…`).
    #[value(name = "airmar")]
    Airmar,
    /// `$PCDIN,…` sentences (Chetco SeaSmart.Net, Quark-elec, Yacht Devices).
    #[value(name = "seasmart", alias = "chetco")]
    SeaSmart,
    /// `$MXPGN,…` sentences as a ShipModul MiniPlex sends them, one CAN
    /// frame each; other NMEA 0183 sentences in the stream are skipped.
    #[value(name = "mxpgn")]
    Mxpgn,
    /// `$MXPGN,…` sentences as an Actisense gateway sends them (a W2K-2's
    /// CANDIRECT data server in NMEA 0183): the data in bus order.
    #[value(name = "actisense-mxpgn")]
    ActisenseMxpgn,
    /// Garmin CSV (relative ms-since-boot timestamps).
    #[value(name = "garmin", alias = "garmin-csv")]
    Garmin,
    /// Garmin CSV2 (absolute timestamps + `Processed PGN` column).
    #[value(name = "garmin-csv2")]
    GarminCsv2,
    /// SocketCAN `candump` text (pretty `can0 <id> [len] <hex>…`,
    /// `-l`/`-L` log `(<epoch>) can0 <id>#<hex>` or Angstrom
    /// `<0x<id>> [len] <hex>…` lines).
    #[value(name = "candump")]
    Candump,
    /// Actisense `.ebl` binary log. Not a line format — decoded via the
    /// EBL frame reader, so it has no [`InputFormat`] mapping.
    #[value(name = "actisense-ebl", alias = "ebl")]
    ActisenseEbl,
    /// Actisense N2K Actisense (BST-D0): reassembled NMEA 2000 messages in
    /// BDTP framing, as a W2K-1 or PRO-NDC-1E2K sends them in Actisense
    /// mode. Binary too, with its own frame reader.
    #[value(name = "n2k-actisense")]
    BstD0,
    /// Actisense RAW Actisense (BST-95): raw CAN frames in BDTP framing, as
    /// a PRO-NDC-1E2K or W2K-1 sends them in "CAN Actisense" mode. Binary
    /// too; the reader takes any BST message that carries a frame.
    #[value(name = "raw-actisense")]
    Bst95,
    /// Waveshare USB-CAN-A: the 20-byte frames it sends on its serial port,
    /// as `cat /dev/ttyUSB0 > capture.bin` saves them. Binary too.
    #[value(name = "usb-can-a")]
    UsbCanA,
}

impl FromFormat {
    /// Map to the engine line-parser selector. `None` for the binary
    /// formats, which use their own frame readers.
    fn to_input_format(self) -> Option<InputFormat> {
        Some(match self {
            FromFormat::Plain => InputFormat::Plain,
            FromFormat::PlainMixFast => InputFormat::PlainMixFast,
            FromFormat::Actisense => InputFormat::ActisenseAscii,
            FromFormat::Ydwg02 => InputFormat::Ydwg02,
            FromFormat::Ikonvert => InputFormat::Ikonvert,
            FromFormat::Airmar => InputFormat::Airmar,
            FromFormat::SeaSmart => InputFormat::SeaSmart,
            FromFormat::Mxpgn => InputFormat::Mxpgn,
            FromFormat::ActisenseMxpgn => InputFormat::ActisenseMxpgn,
            FromFormat::Garmin => InputFormat::GarminCsv,
            FromFormat::GarminCsv2 => InputFormat::GarminCsv2,
            FromFormat::Candump => InputFormat::Candump,
            FromFormat::ActisenseEbl
            | FromFormat::BstD0
            | FromFormat::Bst95
            | FromFormat::UsbCanA
            | FromFormat::Json => {
                return None;
            }
        })
    }
}

/// Long help for `canboat convert --help`. Kept next to the parser and
/// wired into the subcommand from `main.rs` (clap takes the subcommand's
/// short `about` from the enum variant's doc comment, so `long_about`
/// has to be attached there).
pub const LONG_ABOUT: &str = "\
Convert a capture between formats: any supported input → PLAIN, JSON, or text.

Input line formats (auto-detected from the first line, or forced with --from):
  plain, plain-mix-fast, n2k-ascii, raw-ascii, ikonvert, airmar, seasmart,
  mxpgn, garmin, garmin-csv2, candump; actisense-mxpgn only with --from.
  (n2k-ascii and raw-ascii are Actisense's names; actisense and ydwg02 still
  work.) mxpgn is ShipModul's $MXPGN, the data last byte first;
  actisense-mxpgn the same sentences with the data in bus order, as an
  Actisense gateway (a W2K-2) sends them. Nothing in a stream tells them
  apart, so $MXPGN is read as ShipModul's unless --from says otherwise.

Binary input formats (forced with --from; a .ebl file is recognised by name):
  actisense-ebl  Actisense .ebl binary log
  n2k-actisense  Actisense N2K Actisense (BST-D0): a W2K-1 or PRO-NDC-1E2K in Actisense mode
  raw-actisense  Actisense RAW Actisense (BST-95): a PRO-NDC-1E2K or W2K-1 in CAN Actisense mode

Container files (unwrapped automatically by file extension):
  .pcap / .pcap.gz   libpcap SocketCAN capture (link-type 227)
  .nif               Navico Information File (filtered + raw capture groups)

Output formats (--to, default json):
  plain          canboat PLAIN/FAST lines (raw frames, no decode)
  json           one decoded JSON object per record
  text           canboat human-readable text, one line per record
  raw-ascii      RAW ASCII lines, as a YDWG-02 sends them (raw frames)
  n2k-ascii      Actisense N2K ASCII lines (reassembled messages)
  actisense-ebl  Actisense .ebl binary log (raw frames)
  mxpgn          $MXPGN sentences, as a ShipModul MiniPlex sends them (raw frames)
  actisense-mxpgn  $MXPGN sentences, as an Actisense gateway sends them (raw frames)
  seasmart       $PCDIN sentences, as a SeaSmart gateway takes them (whole messages)

Decoded-output shape (json/text only):
  --id STYLE     spaces | camel (default) | uppercamel
  --units SYSTEM si (default) | metric
  --wrap         nest each JSON record in {\"<pgnId>\":{…}} (off by default)
  --no-banner    drop the leading {\"version\":…,\"units\":…} line (json only)

  Decoded output defaults to flat records with camelCase keys and strict
  SI units — the shape machine consumers want. For canboat C's historical
  output pass --id spaces --units metric. The old --camel / --upper-camel
  / --si flags still work but are deprecated (--camel and --upper-camel
  imply --wrap, as they do in canboat C).";

#[derive(Debug, clap::Args)]
pub struct Args {
    /// Input file. A `.pcap` / `.pcap.gz` / `.nif` container is
    /// unwrapped automatically. Omit (or pass `-`) to read stdin.
    #[arg(long, value_name = "PATH")]
    file: Option<PathBuf>,

    /// Input file as a positional argument — an alternative spelling of
    /// `--file` (e.g. `canboat convert capture.nif`).
    #[arg(value_name = "FILE", conflicts_with = "file")]
    file_pos: Option<PathBuf>,

    /// Force the input line format instead of auto-detecting it from
    /// the first line. Ignored for `.pcap`/`.nif` containers.
    #[arg(long, value_enum, value_name = "FORMAT")]
    from: Option<FromFormat>,

    /// Output format.
    #[arg(long, value_enum, default_value_t = OutFormat::Json)]
    to: OutFormat,

    /// Accept $PCDIN and $MXPGN sentences whose checksum is wrong.
    #[arg(long)]
    ignore_checksum: bool,

    /// Filter: only emit frames with this source address.
    #[arg(long, value_name = "N")]
    src: Option<u8>,

    /// Filter: only emit frames with this destination address.
    #[arg(long, value_name = "N")]
    dst: Option<u8>,

    /// Filter: only emit frames with this PGN.
    #[arg(long, value_name = "PGN")]
    pgn: Option<u32>,

    /// `.nif` containers only: emit just the `raw` full-traffic
    /// captures, skipping the filtered device roster.
    #[arg(long, conflicts_with = "filtered_only")]
    raw_only: bool,

    /// `.nif` containers only: emit just the `filtered` device-roster
    /// captures.
    #[arg(long)]
    filtered_only: bool,

    // ----- decoded-output shape: the same knobs the `analyzer` alias
    // exposes, so `convert --to json` can be diffed against canboat C.
    /// Emit lookup values as `{"value":N,"name":"..."}` instead of the
    /// bare name string. Matches canboat's `-nv`. JSON only.
    #[arg(long)]
    nv: bool,

    /// Include unavailable fields — `null` in JSON, `Unknown` in text —
    /// instead of omitting them. Matches canboat's `-empty`.
    #[arg(long)]
    empty: bool,

    /// Annotate every field with its raw bytes and bits. Matches
    /// canboat's `-debug`.
    #[arg(long)]
    debug: bool,

    /// Identifier spelling (`--id`) and unit system (`--units`), plus
    /// their deprecated canboat C spellings. Defaults: camelCase + SI.
    #[command(flatten)]
    shape: crate::cli::ShapeArgs,

    /// Decode-time device quirks, off by default (`--quirk
    /// gps-rollover`). Not for replaying pre-2019 captures.
    #[command(flatten)]
    quirk: crate::cli::QuirkArgs,

    /// Suppress the leading `{"version":…,"units":…}` banner that
    /// `--to json` emits. Use when the output is diffed as a pure
    /// record stream; leave it on when piping into `n2kd`, which reads
    /// the unit system off that line.
    #[arg(long)]
    no_banner: bool,

    /// `--protocol j1939` decodes against the J1939 schema
    /// (`database/j1939/pgns/`) instead of NMEA 2000 — the counterpart
    /// of running the C `analyzer-j1939`. For plain J1939 buses
    /// (engines, gensets); the ISO PGNs exist in both tables.
    #[command(flatten)]
    protocol: crate::cli::protocol::ProtocolArgs,

    /// Lat/lon display format. Matches canboat's `-geo`.
    #[arg(long, value_name = "FMT", default_value = "dd")]
    geo: GeoArg,
}

/// `--geo` values, mirroring canboat's `-geo {dd|dm|dms}`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)]
enum GeoArg {
    /// Decimal degrees, `dd.dddddd`.
    Dd,
    /// Degrees and decimal minutes, `dd.mm.mmm`.
    Dm,
    /// Degrees, minutes, seconds, `dd.mm.sss`.
    Dms,
}

impl From<GeoArg> for GeoFormat {
    fn from(g: GeoArg) -> Self {
        match g {
            GeoArg::Dd => GeoFormat::Dd,
            GeoArg::Dm => GeoFormat::Dm,
            GeoArg::Dms => GeoFormat::Dms,
        }
    }
}

impl Args {
    /// The input path: `--file` or the positional form, whichever was
    /// given. `None` means stdin.
    fn input(&self) -> Option<&Path> {
        self.file.as_deref().or(self.file_pos.as_deref())
    }

    /// Whether the input is stdin: none given, or `-`.
    fn reads_stdin(&self) -> bool {
        self.input().is_none_or(|p| p.as_os_str() == "-")
    }

    /// Which capture groups of a `.nif` to unwrap.
    fn container_opts(&self) -> container::Options {
        container::Options {
            filtered: !self.raw_only,
            raw: !self.filtered_only,
        }
    }
}

pub fn run(args: Args) -> Result<()> {
    // Surface decode-path diagnostics (e.g. a stream mislabeled
    // `# format=FAST` but carrying single frames) on stderr without
    // polluting the converted stdout. Warnings show by default; set
    // `RUST_LOG` for more. Ignore a double-init if a shim already set one.
    let _ = env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("warn"))
        .try_init();
    args.shape.warn_deprecated();
    args.quirk.apply();
    // Resolved before any I/O, so a refused `--protocol` never leaves a
    // half-written output behind, whichever path runs.
    let protocol = args.protocol.resolve()?;
    let forced = args.from.and_then(FromFormat::to_input_format);
    let binary = binary_input(&args);
    let live = args.reads_stdin() && crate::cli::stdin_is_live();
    let mut out = crate::cli::stdout(live);

    if args.to.is_frame_level() {
        convert_raw(&args, protocol, forced, binary, live, &mut out)
    } else {
        convert_decoded(&args, protocol, forced, binary, &mut out)
    }
}

/// A binary input, read by its own frame reader rather than a line parser.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Binary {
    /// An Actisense `.ebl` binary log.
    Ebl,
    /// Actisense BST-D0 messages.
    BstD0,
    /// Actisense BST messages; BST-95 raw CAN frames, say.
    Bst,
    /// Waveshare USB-CAN-A frames.
    UsbCanA,
}

impl Binary {
    /// A reader of `source`, a capture of a bus carrying `protocol`.
    fn reader<R: io::Read + 'static>(
        self,
        source: R,
        protocol: BusProtocol,
    ) -> Box<dyn FrameReader> {
        match self {
            Binary::Ebl => Box::new(EblReader::with_protocol(source, protocol)),
            Binary::BstD0 => Box::new(BstD0Reader::new(source)),
            Binary::Bst => Box::new(BstReader::with_protocol(source, protocol)),
            Binary::UsbCanA => Box::new(UsbCanAReader::with_protocol(source, protocol)),
        }
    }
}

/// The binary format of the input, if it is one: forced with `--from`,
/// or (unforced) a `.ebl` filename.
fn binary_input(args: &Args) -> Option<Binary> {
    match args.from {
        Some(FromFormat::ActisenseEbl) => Some(Binary::Ebl),
        Some(FromFormat::BstD0) => Some(Binary::BstD0),
        Some(FromFormat::Bst95) => Some(Binary::Bst),
        Some(FromFormat::UsbCanA) => Some(Binary::UsbCanA),
        Some(_) => None,
        None => args
            .input()
            .is_some_and(|p| p.extension().is_some_and(|e| e.eq_ignore_ascii_case("ebl")))
            .then_some(Binary::Ebl),
    }
}

/// Raw path: input frames → the selected frame-level format, no decode.
/// The no-filter case is exactly [`crate::io::copy`]; with filters, or
/// from a `live` input, whose every frame is flushed as it comes (a
/// binary format has no line end to flush at), we run the same pull loop.
/// The writer is chosen by `--to` (PLAIN / YDWG02 / Actisense ASCII /
/// Actisense EBL).
fn convert_raw<W: Write>(
    args: &Args,
    protocol: BusProtocol,
    forced: Option<InputFormat>,
    binary: Option<Binary>,
    live: bool,
    out: &mut W,
) -> Result<()> {
    let source = open_source(args.input(), args.container_opts())?;
    let mut reader: Box<dyn FrameReader> = if let Some(binary) = binary {
        binary.reader(source, protocol)
    } else if args.from == Some(FromFormat::Json) {
        // Bare physical values are read against whatever unit system the
        // input's banner declares; `--units` is only the assumption for
        // a bannerless stream. `-nv` raw values are unit-agnostic.
        Box::new(crate::json_input::JsonFrameReader::new(
            source,
            protocol.database(args.shape.units()),
        ))
    } else {
        let lines = match forced {
            Some(fmt) => LineFrameReader::with_format(source, fmt),
            None => LineFrameReader::new(source),
        }
        .protocol(protocol)
        .ignore_checksum(args.ignore_checksum);
        // MXPGN splits a fast-packet into its frames itself, and SeaSmart
        // carries it whole, so both need whole messages; the binary
        // readers above join them already.
        if matches!(
            args.to,
            OutFormat::Mxpgn | OutFormat::ActisenseMxpgn | OutFormat::SeaSmart
        ) {
            Box::new(Joined::new(lines, protocol))
        } else {
            Box::new(lines)
        }
    };
    let mut writer: Box<dyn FrameWriter + '_> = match args.to {
        OutFormat::Plain => Box::new(PlainWriter::new(out)),
        OutFormat::Ydwg02 => Box::new(TextLineWriter::ydwg02(out)),
        OutFormat::Actisense => Box::new(TextLineWriter::actisense(out)),
        OutFormat::ActisenseEbl => Box::new(EblWriter::new(out)),
        OutFormat::Mxpgn => Box::new(MxpgnWriter::new(out, protocol)),
        OutFormat::ActisenseMxpgn => Box::new(MxpgnWriter::actisense(out, protocol)),
        OutFormat::SeaSmart => Box::new(TextLineWriter::seasmart(out)),
        OutFormat::Json | OutFormat::Text => unreachable!("decoded formats routed elsewhere"),
    };

    if !args.has_filter() && !live {
        copy(&mut *reader, &mut *writer).context("converting frames")?;
        return Ok(());
    }
    while let Some(frame) = reader.read_frame().context("reading input")? {
        if args.keep(&frame) {
            writer.write_frame(&frame).context("writing frame")?;
            if live {
                writer.flush().context("flushing output")?;
            }
        }
    }
    writer.flush().context("flushing output")?;
    Ok(())
}

/// The frames of a line capture with each fast-packet joined into its
/// message, by the analyzer's rules: a line of more than 8 bytes, or a
/// `# format=` header of a coalesced format, means the whole stream is
/// coalesced already (but for PLAIN_MIX_FAST), as is every line of a
/// format whose lines are whole messages.
struct Joined<R: BufRead> {
    lines: LineFrameReader<R>,
    protocol: BusProtocol,
    reassembler: Reassembler,
    coalesced: bool,
}

impl<R: BufRead> Joined<R> {
    fn new(lines: LineFrameReader<R>, protocol: BusProtocol) -> Self {
        Self {
            lines,
            protocol,
            reassembler: Reassembler::new(),
            coalesced: false,
        }
    }
}

impl<R: BufRead> FrameReader for Joined<R> {
    fn read_frame(&mut self) -> io::Result<Option<RawFrame>> {
        while let Some(frame) = self.lines.read_frame()? {
            let format = self.lines.active_format();
            self.coalesced |= self.lines.header_coalesced()
                || (frame.data.len() > 8 && format != Some(InputFormat::PlainMixFast));
            let whole = self.coalesced || format.is_some_and(|f| f.framing() == Framing::Coalesced);
            let packet_type = if whole {
                FramePacketType::Other
            } else {
                self.protocol.packet_type(frame.pgn)
            };
            match self.reassembler.push(frame, packet_type) {
                Reassembled::PassThrough(f) | Reassembled::Complete(f) => return Ok(Some(f)),
                Reassembled::Partial => {}
                Reassembled::Error(e) => log::warn!("reassembly error: {e}"),
            }
        }
        Ok(None)
    }
}

/// Decoded path: drive the analyzer pipeline and render each record as
/// JSON or text. Filters are pushed into the pipeline's `Config`.
fn convert_decoded<W: Write>(
    args: &Args,
    protocol: BusProtocol,
    forced: Option<InputFormat>,
    binary: Option<Binary>,
    out: &mut W,
) -> Result<()> {
    let json_opts = JsonOptions {
        include_empty: args.empty,
        name_value: args.nv,
        debug: args.debug,
        camel_case: args.shape.camel_case(),
        wrap: args.shape.wrap(),
        protocol,
    };
    let text_opts = TextOptions {
        show_unavailable: args.empty,
        debug: args.debug,
        geo: args.geo.into(),
    };
    let as_json = args.to == OutFormat::Json;

    // Lead a JSON stream with the same one-line banner `analyzer` emits
    // (version, commit, units, showLookupValues). It is a producer
    // contract, not decoration: `n2kd` reads `"units"` off it to pick
    // the schema it rebuilds each record against, and without it a
    // bannerless stream is assumed Metric — which would silently read
    // this converter's radians as degrees. Text output has no banner in
    // canboat C either.
    if as_json && !args.no_banner {
        writeln!(
            out,
            "{}",
            crate::build_info::version_banner(args.shape.is_si(), args.nv, protocol)
        )
        .context("writing JSON banner")?;
    }

    let cfg = analyze::Config {
        forced_format: forced,
        pgn_filter: args.pgn,
        src_filter: args.src,
        dst_filter: args.dst,
        suppress_startup_record: false,
        units: args.shape.units(),
        protocol,
        fixed_time: None,
        ignore_checksum: args.ignore_checksum,
    };

    let mut line = String::with_capacity(512);
    let mut sink_err: Option<io::Error> = None;
    let sink = |decoded: &crate::engine::DecodedPgn| {
        if sink_err.is_some() {
            return;
        }
        line.clear();
        if as_json {
            write_json(&mut line, decoded, &json_opts).expect("write to String");
        } else {
            write_text(&mut line, decoded, &text_opts).expect("write to String");
        }
        if let Err(e) = out
            .write_all(line.as_bytes())
            .and_then(|()| out.write_all(b"\n"))
        {
            sink_err = Some(e);
        }
    };

    let source = open_source(args.input(), args.container_opts())?;
    if binary.is_some() || args.from == Some(FromFormat::Json) {
        // Binary records and re-encoded JSON records are already
        // complete N2K messages, so skip the line-reader / reassembly /
        // coalesced-mode machinery entirely: pull each frame and decode
        // it directly, honouring the filters.
        // `db` decodes into the *output* unit system (`--units`). The
        // JSON reader tracks the *input's* separately, from its banner,
        // so `--from json` doubles as a unit converter: read a canboat C
        // `"units":"std"` stream, emit SI (or the other way round).
        let db = cfg.protocol.database(cfg.units);
        let mut reader: Box<dyn FrameReader> = if let Some(binary) = binary {
            binary.reader(source, cfg.protocol)
        } else {
            Box::new(crate::json_input::JsonFrameReader::new(source, db))
        };
        let mut sink = sink;
        while let Some(frame) = reader.read_frame().context("reading input")? {
            if cfg.pgn_filter.is_some_and(|w| frame.pgn != w)
                || cfg.src_filter.is_some_and(|w| frame.src != w)
                || cfg.dst_filter.is_some_and(|w| frame.dst != w)
            {
                continue;
            }
            if let Ok(decoded) = db.decode(&frame) {
                sink(&decoded);
            }
        }
    } else {
        analyze::decode_stream(source, &cfg, sink).context("decoding input")?;
    }
    if let Some(e) = sink_err {
        return Err(e).context("writing output");
    }
    Ok(())
}

/// Open the input as a canboat PLAIN-capable [`BufRead`]. A `.nif` /
/// `.pcap` container is unwrapped on the fly; `None` or `-` is stdin.
fn open_source(file: Option<&Path>, opts: container::Options) -> Result<Box<dyn BufRead>> {
    match file {
        Some(path) if path.as_os_str() != "-" => {
            if container::is_container(path) {
                container::plain_reader(path, opts)
                    .with_context(|| format!("opening {}", path.display()))
            } else {
                let f = File::open(path).with_context(|| format!("opening {}", path.display()))?;
                Ok(Box::new(BufReader::new(f)))
            }
        }
        // `Stdin::lock` yields a `'static` guard, so this is owned.
        _ => Ok(Box::new(io::stdin().lock())),
    }
}

impl Args {
    fn has_filter(&self) -> bool {
        self.src.is_some() || self.dst.is_some() || self.pgn.is_some()
    }

    /// True when `frame` passes the src/dst/pgn filters.
    fn keep(&self, frame: &RawFrame) -> bool {
        self.src.is_none_or(|s| frame.src == s)
            && self.dst.is_none_or(|d| frame.dst == d)
            && self.pgn.is_none_or(|p| frame.pgn == p)
    }
}