KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use fast_forward method in fMBT

Best Python code snippet using fMBT_python

test_kitchen.py

Source:test_kitchen.py

...84            f"was turned off for {SHORT_DELAY} minutes")85    class TestAfterDelay:86        def test_turn_water_heater_back_on(self, when_new, time_travel, assert_that):87            when_new.click_button()88            time_travel.fast_forward(SHORT_DELAY).minutes()89            assert_that(ID['bathroom']['water_heater']).was.turned_on()90        def test_send_notification(self, when_new, time_travel, assert_water_heater_notif_sent):91            when_new.click_button()92            time_travel.fast_forward(SHORT_DELAY).minutes()93            assert_water_heater_notif_sent("was turned back on")94class TestDoubleClickOnButton:95    def test_turn_off_water_heater(self, when_new, assert_that):96        when_new.click_button(type='double')97        assert_that(ID['bathroom']['water_heater']).was.turned_off()98    def test_send_notification(self, when_new, assert_water_heater_notif_sent):99        when_new.click_button(type='double')100        assert_water_heater_notif_sent(101            f"was turned off for {LONG_DELAY} minutes")102    class TestAfterShortDelay:103        def test_DOES_NOT_turn_water_heater_back_on(self, when_new, time_travel, assert_that):104            when_new.click_button(type='double')105            time_travel.fast_forward(SHORT_DELAY).minutes()106            assert_that(ID['bathroom']['water_heater']).was_not.turned_on()107        def test_DOES_NOT_send_notification(self, when_new, time_travel, assert_water_heater_notif_NOT_sent):108            when_new.click_button(type='double')109            time_travel.fast_forward(SHORT_DELAY).minutes()110            assert_water_heater_notif_NOT_sent("was turned back on")111    class TestAfterLongDelay:112        def test_turn_water_heater_back_on(self, when_new, time_travel, assert_that):113            when_new.click_button(type='double')114            time_travel.fast_forward(LONG_DELAY).minutes()115            assert_that(ID['bathroom']['water_heater']).was.turned_on()116        def test_send_notification(self, when_new, time_travel, assert_water_heater_notif_sent):117            when_new.click_button(type='double')118            time_travel.fast_forward(LONG_DELAY).minutes()119            assert_water_heater_notif_sent("was turned back on")120class TestClickCancellation:121    class TestSingleClick:122        def test_new_click_cancels_previous_one(self, when_new, time_travel, assert_that):123            #  T = 0min124            # FF = 0min125            time_travel.assert_current_time(0).minutes()126            when_new.click_button()127            #  T = 2min128            # FF = 2min129            time_travel.fast_forward(2).minutes()130            time_travel.assert_current_time(2).minutes()131            when_new.click_button()132            #  T = SHORT_DELAY133            # FF = SHORT_DELAY - 2min134            # Do NOT turn water heater back on yet!135            time_travel.fast_forward(SHORT_DELAY - 2).minutes()136            time_travel.assert_current_time(SHORT_DELAY).minutes()137            assert_that(ID['bathroom']['water_heater']).was_not.turned_on()138            #  T = SHORT_DELAY + 2min139            # FF = SHORT_DELAY + 2min - (2min + 8min)140            time_travel.fast_forward(SHORT_DELAY - 8).minutes()141            time_travel.assert_current_time(SHORT_DELAY + 2).minutes()142            assert_that(ID['bathroom']['water_heater']).was.turned_on()143        def test_multiple_clicks(self, when_new, time_travel, assert_that):144            # Given: 3 clicks, every 2 seconds145            when_new.click_button()146            time_travel.fast_forward(2).minutes()147            when_new.click_button()148            time_travel.fast_forward(2).minutes()149            when_new.click_button()150            time_travel.assert_current_time(4).minutes()151            # When 1/2:152            # Fast forwarding up until 1 min before reactivation153            # scheduled by last click154            time_travel.fast_forward(SHORT_DELAY - 1).minutes()155            # Then 1/2:156            # Water heater still not turned back on (first clicks ignored)157            assert_that(ID['bathroom']['water_heater']).was_not.turned_on()158            # When 2/2:159            # Fast forwarding after reactivation160            # scheduled by last click161            time_travel.fast_forward(SHORT_DELAY - 1).minutes()162            # Then 2/2:163            # Water heater still now turned back on164            assert_that(ID['bathroom']['water_heater']).was.turned_on()165    class TestDoubleClick:166        def test_multiple_clicks(self, when_new, time_travel, assert_that):167            # Given: 3 clicks, every 2 seconds168            when_new.click_button(type='double')169            time_travel.fast_forward(2).minutes()170            when_new.click_button(type='double')171            time_travel.fast_forward(2).minutes()172            when_new.click_button(type='double')173            time_travel.assert_current_time(4).minutes()174            # When 1/2:175            # Fast forwarding up until 1 min before reactivation176            # scheduled by last click177            time_travel.fast_forward(LONG_DELAY - 1).minutes()178            # Then 1/2:179            # Water heater still not turned back on (first clicks ignored)180            assert_that(ID['bathroom']['water_heater']).was_not.turned_on()181            # When 2/2:182            # Fast forwarding after reactivation183            # scheduled by last click184            time_travel.fast_forward(LONG_DELAY - 1).minutes()185            # Then 2/2:186            # Water heater still now turned back on187            assert_that(ID['bathroom']['water_heater']).was.turned_on()188    class TestMixedClicks:189        def test_short_then_long_keep_latest(self, when_new, time_travel, assert_that):190            when_new.click_button()191            time_travel.fast_forward(2).minutes()192            when_new.click_button(type='double')193            time_travel.fast_forward(LONG_DELAY - 1).minutes()194            assert_that(ID['bathroom']['water_heater']).was_not.turned_on()195            time_travel.fast_forward(1).minutes()196            assert_that(ID['bathroom']['water_heater']).was.turned_on()197        def test_long_then_short_keep_latest(self, when_new, time_travel, assert_that):198            when_new.click_button(type='double')199            time_travel.fast_forward(2).minutes()200            when_new.click_button()201            time_travel.fast_forward(SHORT_DELAY - 1).minutes()202            assert_that(ID['bathroom']['water_heater']).was_not.turned_on()203            time_travel.fast_forward(1).minutes()...

__init__.py

Source:__init__.py

1"""2# Fast-Forward Indexes3This is the reference implementation of [Fast-Forward indexes](https://arxiv.org/abs/2110.06051).4â  **Important**: As this library is still in its early stages, the API is subject to change! â 5# Features6* Efficient look-up-based computation of dense retrieval scores.7* Interpolation of sparse and dense scores.8* Passage- and document-level retrieval, including MaxP, FirstP and AverageP.9* *Early stopping* of interpolation based on approximated scores.10* Index compression via *sequential coalescing*.11# Installation12Install the package via `pip`:13```bash14pip install fast-forward-indexes15```16Alternatively, the package can be installed from source:17```bash18git clone https://github.com/mrjleo/fast-forward-indexes.git19cd fast-forward-indexes20python -m pip install .21```22After installing the package, simply import the library:23```python24import fast_forward25```26# Getting Started27Using a Fast-Forward index is as simple as providing a TREC run with sparse scores:28```python29from pathlib import Path30from fast_forward.encoder import TCTColBERTQueryEncoder31from fast_forward.index import InMemoryIndex, Mode32from fast_forward.ranking import Ranking33# choose a pre-trained query encoder34encoder = TCTColBERTQueryEncoder("castorini/tct_colbert-msmarco")35# load an index from disk into memory36index = InMemoryIndex.from_disk(Path("/path/to/index"), encoder, Mode.MAXP)37# load a sparse run (TREC format)38sparse_ranking = Ranking.from_file(Path("/path/to/sparse/run.tsv"))39# load all required queries40queries = {41    "q1": "query 1",42    "q2": "query 2",43    # ...44    "qn": "query n"45}46# compute the corresponding dense scores and interpolate47alpha = 0.248result = index.get_scores(49    sparse_ranking,50    queries,51    alpha=alpha,52    cutoff=10,53    early_stopping=True54)55# create a new TREC runfile with the interpolated ranking56result[alpha].save(Path("/path/to/interpolated/run.tsv"))57```58# Guides59This section contains guides for the most important features.60## Creating an Index61Creating a Fast-Forward index is simple. The following snippet illustrates how to create a `fast_forward.index.InMemoryIndex` object and add some representations to it:62```python63my_index = InMemoryIndex(my_encoder)64my_index.add(65    my_vectors, # three vectors in total66    doc_ids=["d1", "d1", "d2"],67    psg_ids=["d1_p1", "d1_p2", "d2_p1"]68)69```70Here, `my_vectors` is a Numpy array of shape `(3, dim)`. The first two vectors correspond to two passages of the document `d1`, the third vector corresponds to `d2`, which has only a single passage. It is also possible to provide either only document IDs or only passage IDs.71The index can then be saved to disk using `fast_forward.index.InMemoryIndex.save` and loaded using `fast_forward.index.InMemoryIndex.from_disk`.72## Using an Index73The usage of a Fast-Forward index (i.e. computing dense scores and interpolation) is currently handled by `fast_forward.index.Index.get_scores`. It requires a ranking (typically from a sparse retriever) and the corresponding queries and handles query encoding, scoring, interpolation and so on:74```python75sparse_ranking = Ranking.from_file(Path("/path/to/sparse/run.tsv"))76result = index.get_scores(77    sparse_ranking,78    queries,79    alpha=[0.0, 0.5, 1.0],80    cutoff=100081)82```83Here, `queries` is a simple dictionary mapping query IDs to actual queries to be encoded. In order to use *early stopping*, use `early_stopping=True`. Note that early stopping is usually only useful for small `cutoff` values.84## Retrieval Modes85Each index has a retrieval mode (`fast_forward.index.Mode`). The active mode influences the way scores are computed (`fast_forward.index.Index.get_scores`). For example, consider the [example index from earlier](#creating-an-index). Setting the mode to `fast_forward.index.Mode.PASSAGE` will cause the index to compute scores on the passage level and return passage IDs:86```python87my_index.mode = Mode.PASSAGE88```89Similarly, the index can return document IDs, where the score of a document computes as90* the highest score of its passages (`fast_forward.index.Mode.MAXP`),91* the score of the its first passage (`fast_forward.index.Mode.FIRSTP`) or92* the average score of all its passages (`fast_forward.index.Mode.AVEP`).93## Custom Query Encoders94Custom query encoders can easily be implemented. The preferred way to do this is by subclassing `fast_forward.encoder.QueryEncoder` and overriding the `fast_forward.encoder.QueryEncoder.encode` method. This allows the new query encoder to make use of batch encoding.95Alternatively, one can use the `fast_forward.encoder.LambdaQueryEncoder` class, which wraps a function that encodes a single query. The following example shows how to do this with a [Pyserini](https://github.com/castorini/pyserini) query encoder:96```python97pyserini_encoder = pyserini.dsearch.AnceQueryEncoder("castorini/ance-msmarco-passage")98my_encoder = LambdaQueryEncoder(pyserini_encoder.encode)99```100Note that this method is usually less efficient, as the queries are encoded one by one rather than in batches.101## Sequential Coalescing102The sequential coalescing algorithm is a compression technique for indexes with multiple representations per document. It is implemented as `fast_forward.index.create_coalesced_index`. Example usage:103```python104my_index = InMemoryIndex.from_disk(Path("/path/to/index"))105coalesced_index = InMemoryIndex(mode=Mode.MAXP)106create_coalesced_index(my_index, coalesced_index, 0.3)107```108# Examples109Example usage scripts can be found in the `examples` module.110## Creating a Fast-Forward Index from Pyserini111You can use this script to convert an existing dense [Pyserini](https://github.com/castorini/pyserini) index to the Fast-Forward format.112```bash113$ python -m fast_forward.examples.create_index_from_pyserini -h114```115```116usage: create_index_from_pyserini.py [-h] [--out_file OUT_FILE] INDEX117positional arguments:118  INDEX                Pyserini index name (pre-built) or path.119optional arguments:120  -h, --help           show this help message and exit121  --out_file OUT_FILE  Output file. (default: out)122```123For example, to convert the pre-built index `msmarco-doc-ance-maxp-bf`:124```bash125python -m fast_forward.examples.create_index_from_pyserini \\126    msmarco-doc-ance-maxp-bf \\127    --out_file my_index128```129## Computing and Interpolating Scores130This script loads an index into memory and performs re-ranking based on a sparse run.131The query encoder is loaded from the [Hugging Face Hub](https://huggingface.co/models).132```bash133$ python -m fast_forward.examples.interpolate -h134```135```136usage: interpolate.py [-h] [--cutoff CUTOFF] [--cutoff_result CUTOFF_RESULT]137                      [--alpha ALPHA [ALPHA ...]] [--early_stopping] [--target TARGET]138                      INDEX {maxp,avep,firstp,passage} ENCODER SPARSE_SCORES139                      QUERY_FILE140positional arguments:141  INDEX                 Fast-Forward index file.142  {maxp,avep,firstp,passage}143                        Retrieval mode.144  ENCODER               Pre-trained transformer encoder.145  SPARSE_SCORES         TREC runfile containing the scores of the sparse retriever.146  QUERY_FILE            Queries (tsv).147optional arguments:148  -h, --help            show this help message and exit149  --cutoff CUTOFF       Maximum number of sparse documents per query. (default: None)150  --cutoff_result CUTOFF_RESULT151                        Maximum number of documents per query in the final ranking.152                        (default: 1000)153  --alpha ALPHA [ALPHA ...]154                        Interpolation weight. (default: [0.0, 0.1, 0.2, 0.3, 0.4, 0.5,155                        0.6, 0.7, 0.8, 0.9, 1.0])156  --early_stopping      Use approximated early stopping. (default: False)157  --target TARGET       Output directory. (default: out)158```159The following example uses the pre-trained `castorini/tct_colbert-msmarco` query encoder.160We re-rank the top-`5000` documents for each query using an interpolation weight of `0.2`.161Finally, the top-`1000` documents are kept.162```bash163python -m fast_forward.examples.interpolate \\164    my_index \\165    maxp \\166    castorini/tct_colbert-msmarco \\167    my_sparse_run.tsv \\168    queries.tsv \\169    --cutoff 5000 \\170    --cutoff_result 1000 \\171    --target results \\172    --alpha 0.2173```...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.