picoparser 26.3.7

Parsing PicoScenes .csi file and convert it to numpy .npy file with multithreaded parsing.

PicoParser.py

Parsing PicoScenes .csi file and convert it to numpy .npy file with multithreaded parsing. Provides faster parsing with significantly lower memory usage compared to PicoScenes-Python-Toolbox.

An example using the libpico dynamic link library.

Usage

Download and install from PyPI:

pip install picoparser

or install using local .whl file:

pip install picoparser-***-py3-none-***.whl

• Replace *** with the corresponding version in the releases (Or build yourself).

Example:

from picoparser import PicoParser

# Always use context manager
with PicoParser(filePath, 4) as parser:

  # Iterate each frame for further processing. Each frame is parsed one by one at iteration.
  for x in parser.iterFrames():
    print(x.standardHeader.addr1)

  # Frames are queued and sent to thread pool for multithreaded parsing. Consume more resources than iter methods.
  for x in parser.getFrames():
    print(x.standardHeader.addr1)

  # Convert the whole file to numpy ndarray. Only works on single pair of RX and TX NIC with consistent signal configuration.
  tstampNdarray, csiNdarray, magNdarray, phaseNdarray = parser.getNdarray(
    True,
    True,
    True,
    True,
  )

PicoParser's available methods:

def __init__(self, filePath: Path, nWorker: int = 1):
  """
  Initialize PicoParser with file path and number of workers.

  Args:
    filePath: Path to the PicoScenes .csi file.
    nWorker: Desired number of workers for multithreaded parsing. Greater than os.cpu_count() will be ignored.
  """

def iterFrameIndices(self) -> Iterator[tuple[int, int]]:
  """
  Yield frame start offsets and lengths from the mapped file.

  Yields:
    Tuples of frame start index and length.
  """

def iterFramesRaw(self) -> Iterator[memoryview]:
  """
  Yield memoryview slices for each frame in the memory mapped file.

  Yields:
    A view of the bytes for frames.
  """

def iterFrames(self, interp: bool = False) -> Iterator[PicoParserFrame]:
  """
  Yield frame data. Each frame is parsed one by one at iteration, resulting in low memory consumption.

  Args:
    interp: Whether to apply interpolation along subcarrier.

  Yields:
    Parsed frames.
  """

def getFramesByIndices(
  self,
  frameIndices: Iterable[tuple[int, int]],
  interp: bool = False,
) -> Iterator[PicoParserFrame]:
  """
  Yield frame data with multithreaded parsing for provided indices. Frames are queued and sent to thread pool for parsing. Consume more resources than iter methods.

  Args:
    frameIndices: Frame start offsets and lengths.
    interp: Whether to apply interpolation along subcarrier.

  Yields:
    Parsed frames.
  """

def getFrames(self, interp: bool = False) -> Iterator[PicoParserFrame]:
  """
  Yield frame data with multithreaded parsing. All frames are queued and sent to thread pool for parsing. Consume more resources than iter methods.

  Args:
    interp: Whether to apply interpolation along subcarrier.

  Yields:
    Parsed frames.
  """

def getNdarray(
  self,
  enableTs: bool,
  enableCsi: bool,
  enableMag: bool,
  enablePhase: bool,
  interp: bool = False,
) -> tuple[
  np.ndarray | None, np.ndarray | None, np.ndarray | None, np.ndarray | None
]:
  """
  Return the whole file's ndarrays according to requested data types. Only works on single pair of RX and TX NIC with consistent signal configuration.

  Args:
    enableTs: Include timestamp data if True.
    enableCsi: Include CSI data if True.
    enableMag: Include magnitude data if True.
    enablePhase: Include phase data if True.
    interp: Whether to apply interpolation along subcarrier.

  Returns:
    Ndarray for timestamp, CSI, magnitude, and phase. None for non-requested component.
  """

Data structure of PicoParserFrame:

@dataclass
class PicoParserFrame:
  standardHeader: StandardHeader
  rxSBasic: RxSBasic
  rxExtraInfo: RxExtraInfo
  csi: Csi

@dataclass
class Ieee80211MacFrameHeaderControlField:
  version: int
  type: int
  subtype: int
  toDS: int
  fromDS: int
  moreFrags: int
  retry: int
  powerMgmt: int
  more: int
  protect: int
  order: int

@dataclass
class StandardHeader:
  controlField: Ieee80211MacFrameHeaderControlField
  addr1: np.ndarray
  addr2: np.ndarray
  addr3: np.ndarray
  frag: int
  seq: int

@dataclass
class RxSBasic:
  deviceType: int
  tstamp: int
  systemTime: int
  centerFreq: int
  controlFreq: int
  cbw: int
  packetFormat: int
  pktCbw: int
  guardInterval: int
  mcs: int
  numSTS: int
  numESS: int
  numRx: int
  noiseFloor: int
  rssi: int

@dataclass
class RxExtraInfo:
  featureCode: int
  length: int
  version: int
  macAddrRom: np.ndarray
  macAddrCur: np.ndarray
  channelSelect: int
  bmode: int
  evm: np.ndarray
  txChainMask: int
  rxChainMask: int
  txPower: int
  cf: int
  txTsf: int
  lastHwTxTsf: int
  channelFlags: int
  txNess: int
  tuningPolicy: int
  pllRate: int
  pllRefdiv: int
  pllClockSelect: int
  agc: int
  antSelect: np.ndarray
  samplingRate: int
  cfo: int
  sfo: int

@dataclass
class Csi:
  deviceType: int
  firmwareVersion: int
  packetFormat: int
  cbw: int
  carrierFreq: int
  samplingRate: int
  subcarrierBandwidth: int
  antSelect: int
  subcarrierOffset: int
  nTones: int
  nTx: int
  nRx: int
  nEss: int
  nCsi: int
  subcarrierIndices: np.ndarray
  csi: np.ndarray
  magnitude: np.ndarray
  phase: np.ndarray

Build

1. Build libpico according its README.md.

2. Move libpico.so (libpico.dll on Windows) to src/picoparser/_native/.

3. Build wheels with python -m build.

Dependencies

• libpico
• numpy

If you find this helpful

Please cite my works.

License

Licensed under the AGPLv3.