sim

Simulation helpers for streams.

recv_packet(ctx, stream, *, domain='sync', context=None) async

Receive a packet from a packetized stream.

Parameters:

Name	Type	Description	Default
ctx	SimulatorContext	Simulator context.	required
stream	Interface	Target stream.	required
domain		Clock domain.	'sync'
context		Context for looking up the clock domain.	None

Returns:

Type	Description
list[Any]	Packet contents.

Source code in katsuo/stream/sim.py

async def recv_packet(ctx: SimulatorContext, stream: stream.Interface, *, domain = 'sync', context = None) -> list[Any]:
    '''Receive a packet from a packetized stream.

    Args:
        ctx: Simulator context.
        stream: Target stream.
        domain: Clock domain.
        context: Context for looking up the clock domain.

    Returns:
        Packet contents.
    '''

    shape = stream.payload.shape()
    if not isinstance(shape, Packet):
        raise TypeError('recv_packet() can only be used on streams with Packet payloads')

    buf = []

    ctx.set(stream.ready, 1)

    while True:
        payload, = await ctx.tick(domain = domain, context = context).sample(stream.payload).until(stream.valid == 1)

        if shape.semantics.has_first and payload.first:
            buf.clear()

        if shape.semantics.has_end and payload.end:
            break

        buf.append(payload.data)

        if shape.semantics.has_last and payload.last:
            break

    ctx.set(stream.ready, 0)

    return buf

send_packet(ctx, stream, packet, *, domain='sync', context=None, omit_start=False, omit_end=False) async

Send a packet to a packetized stream.

Parameters:

Name	Type	Description	Default
ctx	SimulatorContext	Simulator context.	required
stream	Interface	Target stream.	required
packet	Iterable[Any]	Packet contents.	required
domain		Clock domain.	'sync'
context		Context for looking up the clock domain.	None
omit_start		Omit the start of packet indicator (FIRST).	False
omit_end		Omit the end of packet indicator (LAST/END).	False

Source code in katsuo/stream/sim.py

async def send_packet(ctx: SimulatorContext, stream: stream.Interface, packet: Iterable[Any], *, domain = 'sync', context = None, omit_start = False, omit_end = False):
    '''Send a packet to a packetized stream.

    Args:
        ctx: Simulator context.
        stream: Target stream.
        packet: Packet contents.
        domain: Clock domain.
        context: Context for looking up the clock domain.
        omit_start: Omit the start of packet indicator (FIRST).
        omit_end: Omit the end of packet indicator (LAST/END).
    '''

    shape = stream.payload.shape()
    if not isinstance(shape, Packet):
        raise TypeError('send_packet() can only be used on streams with Packet payloads')

    ctx.set(stream.valid, 1)

    for i, e in enumerate(packet):
        ctx.set(stream.payload.data, e)
        if shape.semantics.has_first:
            ctx.set(stream.payload.first, i == 0 and not omit_start)
        if shape.semantics.has_last:
            ctx.set(stream.payload.last, i == len(packet) - 1 and not omit_end)
        if shape.semantics.has_end:
            ctx.set(stream.payload.end, 0)

        await ctx.tick(domain = domain, context = context).until(stream.ready == 1)

    if shape.semantics.has_end and not omit_end:
        ctx.set(stream.payload.end, 1)
        await ctx.tick(domain = domain, context = context).until(stream.ready == 1)
        ctx.set(stream.payload.end, 0)

    ctx.set(stream.valid, 0)

stream_get(ctx, stream, *, domain='sync', context=None) async

Get a payload token from a stream.

Parameters:

Name	Type	Description	Default
ctx	SimulatorContext	Simulator context.	required
stream	Interface	Target stream.	required
domain		Clock domain.	'sync'
context		Context for looking up the clock domain.	None

Returns:

Type	Description
	Payload token.

Source code in katsuo/stream/sim.py

async def stream_get(ctx: SimulatorContext, stream: stream.Interface, *, domain = 'sync', context = None):
    '''Get a payload token from a stream.

    Args:
        ctx: Simulator context.
        stream: Target stream.
        domain: Clock domain.
        context: Context for looking up the clock domain.

    Returns:
        Payload token.
    '''

    ctx.set(stream.ready, 1)

    payload, = await ctx.tick(domain = domain, context = context).sample(stream.payload).until(stream.valid == 1)

    ctx.set(stream.ready, 0)
    return payload

stream_put(ctx, stream, payload, *, domain='sync', context=None) async

Put a payload token into a stream.

Parameters:

Name	Type	Description	Default
ctx	SimulatorContext	Simulator context.	required
stream	Interface	Target stream.	required
payload		Payload token.	required
domain		Clock domain.	'sync'
context		Context for looking up the clock domain.	None

Source code in katsuo/stream/sim.py

async def stream_put(ctx: SimulatorContext, stream: stream.Interface, payload, *, domain = 'sync', context = None):
    '''Put a payload token into a stream.

    Args:
        ctx: Simulator context.
        stream: Target stream.
        payload: Payload token.
        domain: Clock domain.
        context: Context for looking up the clock domain.
    '''

    ctx.set(stream.valid, 1)
    ctx.set(stream.payload, payload)

    await ctx.tick(domain = domain, context = context).until(stream.ready == 1)

    ctx.set(stream.valid, 0)