外观
recording API
模块说明
text
Bounded, single-reader packet archives. Reading never connects or sends.SCHEMA
python
SCHEMA = 'neomega-packet-recording-v1'DEFAULT_SECONDS
python
DEFAULT_SECONDS = 30DEFAULT_MAX_EVENTS
python
DEFAULT_MAX_EVENTS = 1000DEFAULT_MAX_BYTES
python
DEFAULT_MAX_BYTES = 8 << 20open_recording
python
def open_recording(data_dir, *, prefix='packet-observation-')SDK 说明：
text
Create a private, exclusive archive directly in the Host data directory.PacketRecorder
python
class PacketRecorderSDK 说明：
text
Synchronous sink owned by an existing stream's only reader.
record() and note() return False when recording must stop. Check
limit_reason even on idle polls. Close the stream before finish(); pass any
stream/cancellation/close error to finish(error=...). No implicit reader,
subscription, background task or retry is created here.PacketRecorder.__init__
python
def __init__(
self,
output,
*,
session,
snapshot,
packet_catalog = (),
seconds = DEFAULT_SECONDS,
max_events = DEFAULT_MAX_EVENTS,
max_bytes = DEFAULT_MAX_BYTES,
)PacketRecorder.limit_reason
python
@property
def limit_reason(self)PacketRecorder.record
python
def record(self, event)PacketRecorder.note
python
def note(self, name, value)SDK 说明：
text
Save an explicit intent/baseline/receipt under the same byte budget.PacketRecorder.finish
python
def finish(self, reason='completed', *, error=None)SDK 说明：
text
Seal once, after actual stream cleanup. Incomplete results stay so.PacketRecorder.output
python
self.output公开实例属性，在 __init__ 中设置。
PacketRecorder.session
python
self.session公开实例属性，在 __init__ 中设置。
PacketRecorder.snapshot
python
self.snapshot公开实例属性，在 __init__ 中设置。
PacketRecorder.seconds
python
self.seconds公开实例属性，在 __init__ 中设置。
PacketRecorder.max_events
python
self.max_events公开实例属性，在 __init__ 中设置。
PacketRecorder.max_bytes
python
self.max_bytes公开实例属性，在 __init__ 中设置。
PacketRecorder.started
python
self.started公开实例属性，在 __init__ 中设置。
PacketRecorder.records
python
self.records公开实例属性，在 __init__ 中设置。
PacketRecorder.bytes_written
python
self.bytes_written公开实例属性，在 __init__ 中设置。
PacketRecorder.sequence
python
self.sequence公开实例属性，在 __init__、record 中设置。
PacketRecorder.revision
python
self.revision公开实例属性，在 __init__、record 中设置。
collect
python
async def collect(
ctx,
output,
packet_ids,
*,
max_events = DEFAULT_MAX_EVENTS,
seconds = DEFAULT_SECONDS,
max_bytes = DEFAULT_MAX_BYTES,
expected_names = None,
on_event = None,
)SDK 说明：
text
Collect selected packets with one bounded poll loop; callbacks share it.
on_event is an optional async callback, called after each archived event.
Local time limits never cancel an admitted IPC poll. Its bounded host wait
may finish just after the local limit; transport failures remain failures.PacketRecording
python
@dataclass
class PacketRecording字段（包含继承字段）：
PacketRecording.snapshot
python
snapshot: dict数据类字段；声明于 recording.PacketRecording。
PacketRecording.events
python
events: list数据类字段；声明于 recording.PacketRecording。
PacketRecording.notes
python
notes: list数据类字段；声明于 recording.PacketRecording。
PacketRecording.end
python
end: dict | None数据类字段；声明于 recording.PacketRecording。
PacketRecording.complete
python
complete: bool数据类字段；声明于 recording.PacketRecording。
PacketRecording.issues
python
issues: tuple数据类字段；声明于 recording.PacketRecording。
PacketRecording.rows
python
rows: list数据类字段；声明于 recording.PacketRecording。
read_recording
python
def read_recording(
source,
*,
max_bytes = DEFAULT_MAX_BYTES,
max_events = DEFAULT_MAX_EVENTS,
)SDK 说明：
text
Read a bounded archive offline. Invalid schema/footer raises ValueError.
Missing footer or a truncated final line returns complete=False. The caller
must check complete before using the same parsed events for a conclusion.
completeness only describes the archive; it never proves server coverage.