naps.cores.stream package

Submodules

naps.cores.stream.buffer module

class naps.cores.stream.buffer.StreamBuffer(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

Basically a 1 deep Stream FIFO. Can be used to improve timing or to make outputs compliant with the Stream contract

elaborate(platform)

naps.cores.stream.buffer_test module

class naps.cores.stream.buffer_test.TestBuffer(methodName='runTest')

Bases: unittest.case.TestCase

test_basic()

test_output_properties()

naps.cores.stream.counter_source module

class naps.cores.stream.counter_source.CounterStreamSource(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

elaborate(platform)

naps.cores.stream.counter_source_test module

class naps.cores.stream.counter_source_test.CounterStreamSourceTest(methodName='runTest')

Bases: unittest.case.TestCase

test_stream_contract()

naps.cores.stream.debug module

class naps.cores.stream.debug.InflexibleSinkDebug(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

elaborate(platform)

class naps.cores.stream.debug.InflexibleSourceDebug(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

elaborate(platform)

class naps.cores.stream.debug.StreamInfo(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

efficiency_percent = driver_property

elaborate(platform)

stall_both_percent = driver_property

stall_sink_percent = driver_property

stall_source_percent = driver_property

naps.cores.stream.fifo module

naps.cores.stream.fifo.BufferedAsyncStreamFIFO(input, depth, i_domain='sync', o_domain='sync', exact_depth=False, **kwargs)

naps.cores.stream.fifo.BufferedSyncStreamFIFO(input: naps.stream.stream.Stream, depth, **kwargs)

naps.cores.stream.fifo.UnbufferedAsyncStreamFIFO(input, depth, i_domain='sync', o_domain='sync', exact_depth=False, **kwargs)

naps.cores.stream.fifo.UnbufferedSyncStreamFIFO(input: naps.stream.stream.Stream, depth, **kwargs)

naps.cores.stream.fifo_test module

class naps.cores.stream.fifo_test.TestFifo(methodName='runTest')

Bases: unittest.case.TestCase

check_fifo_basic(fifo_generator)

test_async_stream_fifo_buffered()

test_async_stream_fifo_buffered_output_properties()

test_async_stream_fifo_output_properties()

test_sim_async_stream_fifo()

test_sync_stream_fifo()

test_sync_stream_fifo_buffered()

test_sync_stream_fifo_buffered_output_properties()

test_sync_stream_fifo_output_properties()

naps.cores.stream.gearbox module

class naps.cores.stream.gearbox.SimpleStreamGearbox(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

elaborate(platform)

class naps.cores.stream.gearbox.StreamGearbox(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

Resize a Stream by ‘Gearing’ it up / down (changing the word rate)

elaborate(platform)

class naps.cores.stream.gearbox.StreamResizer(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

Simply resizing a Stream by truncating or zero extending the payload

elaborate(platform)

naps.cores.stream.gearbox_test module

class naps.cores.stream.gearbox_test.TestGearbox(methodName='runTest')

Bases: unittest.case.TestCase

test_dont_loose_data()

test_dont_loose_last_16_to_4()

test_dont_loose_last_8_to_4()

test_gearbox_12_to_48_to_64()

test_gearbox_3_to_7()

test_gearbox_48_to_12_last()

test_gearbox_4_to_8_last()

test_gearbox_7_to_3()

test_gearbox_8_to_4_last()

test_gearbox_automated_fifo()

test_output_stream_contract()

class naps.cores.stream.gearbox_test.TestSimpleGearbox(methodName='runTest')

Bases: unittest.case.TestCase

test_output_stream_contract()

test_simple_gearbox_48_to_12_last()

test_simple_gearbox_8_to_4_last()

test_simple_gearbox_dont_loose_last_16_to_4()

naps.cores.stream.metadata_wrapper module

class naps.cores.stream.metadata_wrapper.GenericMetadataWrapper(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

elaborate(platform)

class naps.cores.stream.metadata_wrapper.LastWrapper(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

wraps a core that transforms one BasicStream into another BasicStream (with variable latency) and applies the last signal correctly to its output

elaborate(platform)

naps.cores.stream.metadata_wrapper_test module

class naps.cores.stream.metadata_wrapper_test.GenericMetadataWrapperTest(methodName='runTest')

Bases: unittest.case.TestCase

test_core_output_stream_contract()

test_output_stream_contract()

test_randomized()

class naps.cores.stream.metadata_wrapper_test.LastWrapperContract(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

elaborate(platform)

class naps.cores.stream.metadata_wrapper_test.LastWrapperTest(methodName='runTest')

Bases: unittest.case.TestCase

test_core_output_stream_contract()

test_last_wrapper_contract()

test_long_fifo()

test_output_stream_contract()

test_randomized()

naps.cores.stream.repacking module

class naps.cores.stream.repacking.Repack12BitStream(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

Repacks a packed 12 bit little endian stream to a packed 12 bit big endian stream This core is probably not what you want unless you want to hand of your data to a computer. Then do this as your very last step because everything beyond this is pure confusion.

elaborate(platform)

naps.cores.stream.stream _memory_test module

naps.cores.stream.stream_memory module

class naps.cores.stream.stream_memory.StreamMemoryReader(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

elaborate(platform)

naps.cores.stream.tee module

class naps.cores.stream.tee.StreamCombiner(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

elaborate(platform)

class naps.cores.stream.tee.StreamSplitter(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

Routes a single N-bit stream to multiple (M) outputs which are each N/M bit wide. Internally uses a StreamTee.

elaborate(platform)

class naps.cores.stream.tee.StreamTee(*args, src_loc_at=0, **kwargs)

Bases: nmigen.hdl.ir.Elaboratable

Routes a single stream input to multiple outputs (like a tee shaped piece of pipe). To be able to fulfil the stream contract, every output has a StreamBuffer.

elaborate(platform)

get_output()

naps.cores.stream.tee_test module

class naps.cores.stream.tee_test.StreamTeeTest(methodName='runTest')

Bases: unittest.case.TestCase

test_stream_combiner_output_stream_contract()

test_tee_output_stream_contract()

Module contents