[FIXED] TypeError: "value" parameter must be a scalar, dict or Series, but you passed a "Index"

Issue

df = df['Consequence Number'].fillna("CLS" + df.index.astype(str))

I did try the above solution along with iterating rows

 for index in priority_cls.iterrows():
     if priority_cls['Consequence Number'].isna() == True:
        priority_cls['Consequence Number'] = 'CLS00'+ str(priority_cls['ConsequenceNumber'].isnull().sum())

The error is as follows:

TypeError: "value" parameter must be a scalar, dict or Series, but you passed a "Index"

Solution

Convert your Index to_series:

df['Consequence Number'].fillna("CLS" + df.index.to_series().astype(str))

Example:

df = pd.DataFrame({'Consequence Number': ['A', 'B', pd.NA, pd.NA]})

df['out'] = (df['Consequence Number']
             .fillna("CLS" + df.index.to_series().astype(str))
            )

Output:

  Consequence Number   out
0                  A     A
1                  B     B
2               <NA>  CLS2
3               <NA>  CLS3

Answered By - mozway

Read More

[FIXED] Colab notebook opens browser tabs automatically

Issue

I have a colab notebook with code that opens multiple new tabs in the browser:

from IPython.display import Javascript
def open_tab(url):
    display(Javascript('window.open("{url}");'.format(url=url)))

Everything works fine when I use it with dataframe["urls"].apply(open_tab). However whenever I refresh/reopen the tab with colab notebook again, browser immediately reopens all the tabs that were opened in the last run.

That happens despite notebook being disconnected and runtime deleted previously, so no code is executed. Works the same on Chrome and Edge browsers. I suspect it might be something with browser caching and linking generated tabs with the tab of origin (colab notebook), but have no idea how to verify that :) I expected new tabs being opened only when the cell is executed.

Solution

The display() function creates an output object in the notebook of type display_data. For example consider the following code:

import pandas as pd
from IPython.display import Javascript
def open_tab(url):
    display(Javascript('window.open("{url}");'.format(url=url)))

df = pd.DataFrame(["https://stackoverflow.com", "https://google.com", "https://docs.python.org"], columns=["urls"])
df["urls"].apply(open_tab)

When you run this once and then download the notebook as .ipynb file you will see the following in the notebook source:

"cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 86
        },
        "id": "dKjZ-iZWw0Lc",
        "outputId": "30859861-e042-4912-9d08-f10740731ece"
      },
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "text/plain": [
              "<IPython.core.display.Javascript object>"
            ],
            "application/javascript": [
              "window.open(\"https://stackoverflow.com\");"
            ]
          },
          "metadata": {}
        },
        {
          "output_type": "display_data",
          "data": {
            "text/plain": [
              "<IPython.core.display.Javascript object>"
            ],
            "application/javascript": [
              "window.open(\"https://google.com\");"
            ]
          },
          "metadata": {}
        },
        {
          "output_type": "display_data",
          "data": {
            "text/plain": [
              "<IPython.core.display.Javascript object>"
            ],
            "application/javascript": [
              "window.open(\"https://docs.python.org\");"
            ]
          },
          "metadata": {}
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "0    None\n",
              "1    None\n",
              "2    None\n",
              "Name: urls, dtype: object"
            ]
          },
          "metadata": {},
          "execution_count": 3
        }
      ],
      "source": [
        "import pandas as pd\n",
        "from IPython.display import Javascript\n",
        "def open_tab(url):\n",
        "    display(Javascript('window.open(\"{url}\");'.format(url=url)))\n",
        "\n",
        "df = pd.DataFrame([\"https://stackoverflow.com\", \"https://google.com\", \"https://docs.python.org\"], columns=[\"urls\"])\n",
        "df[\"urls\"].apply(open_tab)"
      ]
    }
  ]

The output that you see at the bottom of the cell is there ("output_type": "execute_result"), and the source code, but also three "output_type": "display_data" objects, one for each url.

This type of output is also used for images, graphs, and rendered HTML, which you would normally most likely want to see when you reopen the notebook, even if you don't want to run the code again. Similar to those outputs, also the generated JavaScript is "shown again", i.e., executed.

To prevent this you can right-click on the cell that generates the JavaScript and choose "remove output", which will delete the output objects (while keeping the output of all other cells).

Answered By - Marijn

Read More

[FIXED] How to remove stop phrases/stop ngrams (multi-word strings) using pandas/sklearn?

Issue

I want to prevent certain phrases for creeping into my models. For example, I want to prevent 'red roses' from entering into my analysis. I understand how to add individual stop words as given in Adding words to scikit-learn's CountVectorizer's stop list by doing so:

from sklearn.feature_extraction import text
additional_stop_words=['red','roses']

However, this also results in other ngrams like 'red tulips' or 'blue roses' not being detected.

I am building a TfidfVectorizer as part of my model, and I realize the processing I need might have to be entered after this stage but I am not sure how to do this.

My eventual aim is to do topic modelling on a piece of text. Here is the piece of code (borrowed almost directly from https://de.dariah.eu/tatom/topic_model_python.html#index-0 ) that I am working on:

from sklearn import decomposition

from sklearn.feature_extraction import text
additional_stop_words = ['red', 'roses']

sw = text.ENGLISH_STOP_WORDS.union(additional_stop_words)
mod_vectorizer = text.TfidfVectorizer(
    ngram_range=(2,3),
    stop_words=sw,
    norm='l2',
    min_df=5
)

dtm = mod_vectorizer.fit_transform(df[col]).toarray()
vocab = np.array(mod_vectorizer.get_feature_names())
num_topics = 5
num_top_words = 5
m_clf = decomposition.LatentDirichletAllocation(
    n_topics=num_topics,
    random_state=1
)

doctopic = m_clf.fit_transform(dtm)
topic_words = []

for topic in m_clf.components_:
    word_idx = np.argsort(topic)[::-1][0:num_top_words]
    topic_words.append([vocab[i] for i in word_idx])

doctopic = doctopic / np.sum(doctopic, axis=1, keepdims=True)
for t in range(len(topic_words)):
    print("Topic {}: {}".format(t, ','.join(topic_words[t][:5])))

EDIT

Sample dataframe (I have tried to insert as many edge cases as possible), df:

   Content
0  I like red roses as much as I like blue tulips.
1  It would be quite unusual to see red tulips, but not RED ROSES
2  It is almost impossible to find blue roses
3  I like most red flowers, but roses are my favorite.
4  Could you buy me some red roses?
5  John loves the color red. Roses are Mary's favorite flowers.

Solution

TfidfVectorizer allows for a custom preprocessor. You can use this to make any needed adjustments.

For example, to remove all occurrences of consecutive "red" + "roses" tokens from your example corpus (case-insensitive), use:

import re
from sklearn.feature_extraction import text

cases = ["I like red roses as much as I like blue tulips.",
         "It would be quite unusual to see red tulips, but not RED ROSES",
         "It is almost impossible to find blue roses",
         "I like most red flowers, but roses are my favorite.",
         "Could you buy me some red roses?",
         "John loves the color red. Roses are Mary's favorite flowers."]

# remove_stop_phrases() is our custom preprocessing function.
def remove_stop_phrases(doc):
    # note: this regex considers "... red. Roses..." as fair game for removal.
    #       if that's not what you want, just use ["red roses"] instead.
    stop_phrases= ["red(\s?\\.?\s?)roses"]
    for phrase in stop_phrases:
        doc = re.sub(phrase, "", doc, flags=re.IGNORECASE)
    return doc

sw = text.ENGLISH_STOP_WORDS
mod_vectorizer = text.TfidfVectorizer(
    ngram_range=(2,3),
    stop_words=sw,
    norm='l2',
    min_df=1,
    preprocessor=remove_stop_phrases  # define our custom preprocessor
)

dtm = mod_vectorizer.fit_transform(cases).toarray()
vocab = np.array(mod_vectorizer.get_feature_names_out())

Now vocab has all red roses references removed.

print(sorted(vocab))

['Could buy',
 'It impossible',
 'It impossible blue',
 'It quite',
 'It quite unusual',
 'John loves',
 'John loves color',
 'Mary favorite',
 'Mary favorite flowers',
 'blue roses',
 'blue tulips',
 'color Mary',
 'color Mary favorite',
 'favorite flowers',
 'flowers roses',
 'flowers roses favorite',
 'impossible blue',
 'impossible blue roses',
 'like blue',
 'like blue tulips',
 'like like',
 'like like blue',
 'like red',
 'like red flowers',
 'loves color',
 'loves color Mary',
 'quite unusual',
 'quite unusual red',
 'red flowers',
 'red flowers roses',
 'red tulips',
 'roses favorite',
 'unusual red',
 'unusual red tulips']

UPDATE (per comment thread):

To pass in desired stop phrases along with custom stop words to a wrapper function, use:

desired_stop_phrases = ["red(\s?\\.?\s?)roses"]
desired_stop_words = ['Could', 'buy']

def wrapper(stop_words, stop_phrases):

    def remove_stop_phrases(doc):
        for phrase in stop_phrases:
            doc = re.sub(phrase, "", doc, flags=re.IGNORECASE)
        return doc
    
    sw = text.ENGLISH_STOP_WORDS.union(stop_words)
    mod_vectorizer = text.TfidfVectorizer(
        ngram_range=(2,3),
        stop_words=sw,
        norm='l2',
        min_df=1,
        preprocessor=remove_stop_phrases
    )
    
    dtm = mod_vectorizer.fit_transform(cases).toarray()
    vocab = np.array(mod_vectorizer.get_feature_names_out())
    
    return vocab

wrapper(desired_stop_words, desired_stop_phrases)

Answered By - andrew_reece

Read More

[FIXED] optuna parameter tuning for tweedie - Input contains infinity or a value too large for dtype('float32') error

Issue

I am trying to tune a XGBRegressor model and I am getting below error only when I try to use the parameter tuning flow:

Input contains infinity or a value too large for dtype('float32')

I do not get this error if I do not try to tune parameters.

I have ensured my data does not have any NaN or np.inf - I replace +/- np.inf with np.nan and replace all NaN with 0 later. Before training, I have changed all columns to np.float64 type.

I suspect during parameter tuning, the target value may be causing overflow with float32 - how to ensure sklearn/xgboost/optuna uses float64 instead of float32?

My training code is roughly following:

def __fit_new_model(
        df: pd.DataFrame, feature_cols: List[str], target_col: str, tuning_iterations: int
    ) -> XGBRegressor:
        from sklearn.model_selection import train_test_split
        from sklearn.metrics import mean_absolute_percentage_error

        if df.empty:
            raise AssertionError("input parameter error - empty input DataFrame")

        if not feature_cols:
            raise AssertionError("input parameter error - empty feature_cols")

        if not target_col:
            raise AssertionError("input parameter error - invalid target_col name")

        X, y = df[feature_cols], df[target_col]

        if X.isna().any().any() or y.isna().any():
            raise AssertionError("input data error - NaN values exist")

        X_train, X_validation, y_train, y_validation = train_test_split(X, y, test_size=0.20, random_state=42)

        regressor: XGBRegressor = XGBRegressor(
            random_state=42,
            tree_method="hist",
            n_estimators=100,
            early_stopping_rounds=100,
            objective="reg:tweedie",
            tweedie_variance_power=1.5,
            eval_metric=mean_absolute_percentage_error,
        )

        if tuning_iterations > 0:
            tuned_parameters: Dict[str, Any] = __get_tuned_model_parames(
                x_train=X_train,
                y_train=y_train,
                x_validation=X_validation,
                y_validation=y_validation,
                num_trials=tuning_iterations,
            )
            regressor = XGBRegressor(eval_metric=mean_absolute_percentage_error, **tuned_parameters)

        regressor.fit(X=X_train, y=y_train, eval_set=[(X_train, y_train), (X_validation, y_validation)], verbose=False)

        return regressor

def __get_tuned_model_parames(
        x_train: pd.DataFrame,
        y_train: pd.Series,
        x_validation: pd.DataFrame,
        y_validation: pd.Series,
        num_trials: int = 200,
    ) -> Dict[str, Any]:
        import optuna

        def objective(trial: optuna.trial.Trial):
            from sklearn.metrics import mean_absolute_percentage_error

            param = {
                "tree_method": "hist",
                "booster": trial.suggest_categorical("booster", ["gbtree", "dart"]),
                "lambda": trial.suggest_float("lambda", 1e-3, 10.0),
                "alpha": trial.suggest_float("alpha", 1e-3, 10.0),
                "colsample_bytree": trial.suggest_float("colsample_bytree", 0.05, 1.0),
                "subsample": trial.suggest_float("subsample", 0.05, 1.0),
                "learning_rate": trial.suggest_float("learning_rate", 1e-3, 0.1, log=True),
                "n_estimators": 100,
                "objective": trial.suggest_categorical(
                    # "objective", ["reg:tweedie", "reg:squarederror", "reg:squaredlogerror"]
                    "objective",
                    ["reg:tweedie"],
                ),
                "max_depth": trial.suggest_int("max_depth", 1, 12),
                "min_child_weight": trial.suggest_int("min_child_weight", 1, 20),
                "early_stopping_rounds": 100,
                "random_state": 42,
                "base_score": 0.5,
            }

            if param["objective"] == "reg:tweedie":
                param["tweedie_variance_power"] = trial.suggest_categorical(
                    "tweedie_variance_power", [1, 1.2, 1.3, 1.4, 1.5, 1.6, 1.7, 1.8, 1.9]
                )

            regressor_model: XGBRegressor = XGBRegressor(**param)
            regressor_model.fit(X=x_train, y=y_train, eval_set=[(x_validation, y_validation)], verbose=False)

            predictions = regressor_model.predict(x_validation)
            mape: float = mean_absolute_percentage_error(y_true=y_validation, y_pred=predictions)

            return mape

        sampler: optuna.samplers.TPESampler = optuna.samplers.TPESampler(seed=42)
        study_xgb = optuna.create_study(direction="minimize", sampler=sampler)
        optuna.logging.set_verbosity(optuna.logging.ERROR)

        study_xgb.optimize(lambda trial: objective(trial), n_trials=num_trials)
        model_params: Dict[str, Any] = study_xgb.best_params

        return model_params

Solution

The distribution may not be a Poisson one with my data and hence removing the value 1 from the choices for tweedie_variance_power resolved the issue. Maybe I made a typo earlier - instead of adding 1.1, I added 1.

The corresponding code section is now following:

            if param["objective"] == "reg:tweedie":
                param["tweedie_variance_power"] = trial.suggest_categorical(
                    "tweedie_variance_power",
                    [
                        1.1,
                        1.2,
                        1.3,
                        1.4,
                        1.5,
                        1.6,
                        1.7,
                        1.8,
                        1.9,
                    ],
                )

Answered By - soumeng78

Read More

[FIXED] How can I fix the metaclass conflict in django-rest-framework Serializer

Issue

I've written the following code in django and DRF:

class ServiceModelMetaClass(serializers.SerializerMetaclass, type):
    SERVICE_MODELS = {
        "email": EmailSubscriber,
        "push": PushSubscriber
    }

    def __call__(cls, *args, **kwargs):
        service = kwargs.get("data", {}).get("service")
        cls.Meta.subscriber_model = cls.SERVICE_MODELS.get(service)
        return super().__call__(*args, **kwargs)


class InterListsActionsSerializer(serializers.Serializer, metaclass=ServiceModelMetaClass):
    source_list_id = serializers.IntegerField()
    target_list_id = serializers.IntegerField()
    subscriber_ids = serializers.IntegerField(many=True, required=False)
    account_id = serializers.CharField()
    service = serializers.ChoiceField(choices=("email", "push"))

    class Meta:
        subscriber_model: Model = None

    def move(self):
        model = self.Meta.subscriber_model
        # Rest of the method code.

The purpose of this code is that this serializer might need doing operation on different models based on the service that the user wants to use. So I wrote this metaclass to prevent writing duplicate code and simply change the subscriber_model based on user's needs.
Now as you might know, serializers.Serializer uses a metaclass by its own, serializers.SerializerMetaclass. If I don't use this metaclass for creating my metaclass, it results in the following error:
TypeError: metaclass conflict: the metaclass of a derived class must be a (non-strict) subclass of the metaclasses of all its bases.
But when I try making my ServiceModelMetaClass metaclass to inherit from serializers.SerializerMetaclass it gives me this error:

   File "/project-root/segment_management/serializers/subscriber_list.py", line 33, in <module>
    class InterListsActionsSerializer(serializers.Serializer, metaclass=ServiceModelMetaClass):
  File "/project-root/segment_management/serializers/subscriber_list.py", line 36, in InterListsActionsSerializer
    subscriber_ids = serializers.IntegerField(many=True, required=False)
  File "/project-root/.venv/lib/python3.10/site-packages/rest_framework/fields.py", line 894, in __init__
    super().__init__(**kwargs)
TypeError: Field.__init__() got an unexpected keyword argument 'many'

What should I do to fix this problem or maybe a better alternative approach that keeps the code clean without using metaclass? Thanks in advance.

Solution

The problem is not the metaclass, or at least it is not the primary problem. The main problem is the many=True for your IntegerField. I looked it up, and apparently a field can not have many=True, only a Serializer can. Indeed, if we inspect the source code, we see ^[GitHub]:

def __new__(cls, *args, **kwargs):
    # We override this method in order to automatically create
    # `ListSerializer` classes instead when `many=True` is set.
    if kwargs.pop('many', False):
        return cls.many_init(*args, **kwargs)
    return super().__new__(cls, *args, **kwargs)

and the .many_init(…) essentially creates the field as a child field, and wraps it in a ListSerializer ^[GitHub]:

@classmethod def many_init(cls, *args, **kwargs):
    # …
    child_serializer = cls(*args, **kwargs)
    list_kwargs = {
        'child': child_serializer,
    }
    # …
    meta = getattr(cls, 'Meta', None)
    list_serializer_class = getattr(meta, 'list_serializer_class', ListSerializer)
    return list_serializer_class(*args, **list_kwargs)

But we can appy essentially the same trick with a field, by using a ListField ^[drf-doc]:

class InterListsActionsSerializer(
    serializers.Serializer, metaclass=ServiceModelMetaClass
):
    source_list_id = serializers.IntegerField()
    target_list_id = serializers.IntegerField()
    subscriber_ids = serializers.ListField(
        child=serializers.IntegerField(), required=False
    )
    account_id = serializers.CharField()
    service = serializers.ChoiceField(choices=('email', 'push'))

    class Meta:
        subscriber_model: Model = None

    def move(self):
        model = self.Meta.subscriber_model

Answered By - willeM_ Van Onsem

Read More

[FIXED] Mocking signals function in unit test

Issue

I'm using Django version 4.2.6.

I have a file, signals.py, that includes a pre-save handler for my entity:

def do_stuff(sender, instance, **kwargs):
    ... stuff...

pre_save.connect(do_stuff, sender=MyEntity)

... so that when an instance of MyEntity is either created or updated, the do_stuff() function is executed.

How do I mock do_stuff() in unit tests? I have a number of tests that create and update MyEntity instances, so do_stuff() is being called during the execution of the unit test. My implementation of do_stuff() is making a call to an external source amongst other functionality, so I want to mock it when the unit test runs.

I have tried declaring the following a part of my unit test:

@mock.patch("application.package.signals.do_stuff", autospec=True,)

... but when I execute the test, I can see that do_stuff() is still being executed.

Solution

As per bbayles' suggestion above, a solution was implemented where the code to call do_stuff() was only executed when it was deployed into an environment, rather than run locally or via unit tests.

I used os.environ to identify whether this was the case or not.

Answered By - GarlicBread

Read More

[FIXED] This version of ChromeDriver only supports Chrome version 121 Current browser version is 120.0.6099.227

Issue

import undetected_chromedriver as uc
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys

from bs4 import BeautifulSoup

from otp import get_otp

import time

options = webdriver.ChromeOptions()

# options.add_argument("--disable-dev-shm-usage")

# options.add_argument('--enable-javascript')
# options.add_argument('--headless')
# options.add_argument('--no-sandbox')
webdriver_service = Service(ChromeDriverManager().install())
webdriver_service.start()
driver = uc.Chrome(service=webdriver_service, options=options)

there is such a code. After 10 launches, I started getting this error -

selenium.common.exceptions.WebDriverException: Message: unknown error: cannot connect to chrome at 127.0.0.1:54585 from session not created: This version of ChromeDriver only supports Chrome version 121 Current browser version is 120.0.6099.227

Any thoughts pls?

Solution

Currently with undetected-chromedriver, you would need to update the version_main arg of uc.Chrome() every time a new version of Chrome comes out. That's not self-sustaining. Alternatively, there's https://github.com/seleniumbase/SeleniumBase with UC Mode, which is basically undetected-chromedriver with some improvements, such as automatically getting a version of chromedriver that is compatible with your version of Chrome. Set uc=True to activate UC Mode in a SeleniumBase script. Here's an example:

from seleniumbase import Driver

driver = Driver(uc=True)

There's some documentation about it here: https://github.com/seleniumbase/SeleniumBase/issues/2213

Here's a larger script:

from seleniumbase import Driver

driver = Driver(uc=True)
driver.get("https://nowsecure.nl/#relax")
# DO MORE STUFF
driver.quit()

For customizing the reconnect time when loading a URL that has bot detection services, you could swap the get() line with something like this:

driver.uc_open_with_reconnect("https://nowsecure.nl/#relax", reconnect_time=5)

(The reconnect_time is the wait time before chromedriver reconnects with Chrome. Before the time is up, a website cannot detect Selenium, but it also means that Selenium can't yet issue commands to Chrome.)

Answered By - Michael Mintz

Read More

[FIXED] Azure pipeline not finding Selenium tests

Issue

I am trying to set up an Azure pipeline which will run a Selenium test suite. The suite was build in Visual Studio 2022 with C# and Selenium. The suite runs fine from Visual Studio, command line and also Jenkins but we are looking to build it into a release pipeline and this is a test to see how this will work.

The test pipeline appears to work until we reach task - VSTest@2. At this point the process finds the build dll but does not find the tests which are in this file, below is the current code

trigger:
- master

pool:
  name: Azure Pipelines
  vmImage: 'windows-latest'
  demands: vstest

variables:
  solution: '**/Testapp.sln'

steps:
- task: NuGetToolInstaller@1

- task: NuGetCommand@2
  inputs:
    restoreSolution: '$(solution)'

- task: VSBuild@1
  inputs:
    solution: '$(solution)'
    msbuildArgs: '/p:DeployOnBuild=true /p:WebPublishMethod=Package /p:PackageAsSingleFile=true /p:SkipInvalidConfigurations=true /p:DesktopBuildPackageLocation="$(build.artifactStagingDirectory)\WebApp.zip" /p:DeployIisAppPath="Default Web Site"'

- task: VSTest@2
  displayName: 'Regression tests - warmup'
  inputs:
    testSelector: testAssemblies
    testAssemblyVer2: |
      D:\a\1\s\Testapp\bin\Debug\net6.0-windows\Testapp.dll
    testRunTitle: Warmup
    testFiltercriteria: 'TestCategory=TestappE2EWarmup'

I have also tried a few variations, thus

testAssemblyVer2: |
  D:\a\1\s\Testapp\Testapp.csproj

and

testFiltercriteria: 'Name=TestWarmup'

What we are getting at the moment is this

Microsoft (R) Test Execution Command Line Tool Version 17.3.0-preview-20220626-01 (x64)
Copyright (c) Microsoft Corporation.  All rights reserved.
vstest.console.exe "D:\a\1\s\Testapp\bin\Debug\net6.0-windows\Testapp.dll"
/TestCaseFilter:"Category=TestappE2EWarmup"
/Settings:"D:\a\_temp\qhskojhglda.tmp.runsettings"
/Logger:"trx"
/TestAdapterPath:"D:\a\1\s"
Starting test execution, please wait...
A total of 1 test files matched the specified pattern.
NUnit Adapter 4.0.0.0: Test execution started
Running selected tests in D:\a\1\s\Testapp\bin\Debug\net6.0-windows\Testapp.dll
   NUnit3TestExecutor discovered 0 of 0 NUnit test cases using Current Discovery mode, Explicit run
NUnit Adapter 4.0.0.0: Test execution complete
No test matches the given testcase filter `Category=TestappE2EWarmup` in D:\a\1\s\Testapp\bin\Debug\net6.0-windows\Testapp.dll

Now from what I understand, the VSTest@2 plugin should be able to find Selenium tests and test adapters such as NUnit as per the screenshot below:

This is a .NET6.0 project which is being built to replace a .NET Core 3.1 project. The old core code used to find the tests without any problem which is the code below.

- task: DotNetCoreCLI@2
  displayName: 'Regression tests - public'
  inputs:
    command: test
    projects: '$(regressionTests)'
    arguments: '--configuration $(BuildConfiguration) --filter "Category=TestappE2EWarmup"'

Where have I have gone wrong?

Solution

Thanks for the replies, I now having this working by switching from nunit to mstest in the project

<PackageReference Include="MSTest.TestAdapter" Version="2.2.10" />
<PackageReference Include="MSTest.TestFramework" Version="2.2.10" />
<PackageReference Include="SpecFlow.MsTest" Version="3.9.74" />

Now the tests are being found and they run

Answered By - Kev

Read More

[FIXED] Full page screenshot with Selenium Chrome driver in Python

Issue

AI insists that for Selenium 4 or later to take the full page screenshot with webdriver.Chrome(), you can use the 'full' argument in driver.get_screenshot_as_file("screenshot.png", full=True). However, I get error message

TypeError: WebDriver.get_screenshot_as_file() got an unexpected keyword argument 'full'

Please, help.

Solution

The error message does a pretty good job here: WebDriver.get_screenshot_as_file() does not have a keyword argument called 'full'.

Here is a method I have used in the past:

def save_screenshot():
    original_size = driver.get_window_size()
    required_width = driver.execute_script('return document.body.parentNode.scrollWidth')
    required_height = driver.execute_script('return document.body.parentNode.scrollHeight')
    driver.set_window_size(required_width, required_height)
    driver.find_element_by_tag_name('body').screenshot(path)
    driver.set_window_size(original_size['width'], original_size['height'])

Answered By - thetaco

Read More

[FIXED] I want to print the IMDB rating of a movie/series to the terminal after the completion of automation

Issue

I am using Google search for finding the element, as it seems easier to navigate than IMDB.

import selenium.webdriver as webdriver
print("This program finds the imdb rating of a movie or TV series!!!")

def get_results(search_term):
    url="https://www.google.com"
    browser=webdriver.Safari()
    browser.get(url)
    search_box= browser.find_element_by_id("lst-ib")
    search_box.send_keys(search_term)
    search_box.submit()
    links = browser.find_element_by_id("//div[@class = 'slp f']/text()")#this line is problematic, should i use xpath?how?
    print(links)

search_key=input("Enter the movie name : ")
get_results("what is the imdb rating of "+search_key)

and here is the error...

This program finds the imdb rating of a movie or TV series!!!
Enter the movie name : inception
Traceback (most recent call last):
  File "web1.py", line 15, in <module>
    get_results("what is the imdb rating of "+search_key)
  File "web1.py", line 10, in get_results
    links = browser.find_element_by_id("//div[@class = 'slp f']/text()")
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/selenium/webdriver/remote/webdriver.py", line 353, in find_element_by_id
    return self.find_element(by=By.ID, value=id_)
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/selenium/webdriver/remote/webdriver.py", line 957, in find_element
    'value': value})['value']
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/selenium/webdriver/remote/webdriver.py", line 314, in execute
    self.error_handler.check_response(response)
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/selenium/webdriver/remote/errorhandler.py", line 242, in check_response
    raise exception_class(message, screen, stacktrace)
selenium.common.exceptions.NoSuchElementException: Message: An element could not be located on the page using the given search parameters.

Please be specific. Thank you!

Solution

I have a little bit changed your code and it works. Added explicit wait and changed selector.

import selenium.webdriver as webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC


print("This program finds the imdb rating of a movie or TV series!!!")

def get_results(search_term):
    url = "https://www.google.com"
    browser = webdriver.Safari()
    browser.get(url)
    search_box= browser.find_element_by_id("lst-ib")
    search_box.send_keys(search_term)
    search_box.submit()
    rating = WebDriverWait(browser, 5).until(
    EC.presence_of_element_located((By.XPATH, '//*[@id="rso"]/div/div/div[1]/div/div/div[1]/div/div[2]')))
    print(rating.text)
    links = browser.find_elements_by_xpath('//h3/a')
    first_link = links[0].get_attribute('href')
    print(first_link)
    browser.quit()

search_key=input("Enter the movie name : ")
get_results("what is the imdb rating of "+search_key)

Hope, this will help you.

Answered By - Oleksandr Makarenko

Read More

[FIXED] How to check all checkboxes on a webpage with Selenium in Python (using a loop)

Issue

I am attempting to write a python program that will check all checkboxes on a given webpage. Currently, I only have the ability to select a single checkbox and am unsure why I am not able to iterate through the loop of elements containing checkboxes on this example page. Here is my code so far:

from selenium import webdriver
driver = webdriver.Firefox()
driver.maximize_window()
driver.get(
    "https://www.sugarcrm.com/request-demo/")

elements = []
elements = driver.find_elements_by_xpath("//*[@class='form-check-input']")
elemlength = len(elements)

for x in range(1, elemlength):
    if driver.find_element_by_xpath("//input[@type='checkbox']").is_selected():
        print("Already Selected")
    else:
        driver.find_element_by_xpath("//input[@type='checkbox']").click()

Solution

driver.find_element_by_xpath("//input[@type='checkbox']") will return the first checkbox it finds every time, so it will always be the same one.

You can use the list of elements returned by driver.find_elements_by_xpath("//*[@class='form-check-input']") instead.

elements = driver.find_elements_by_xpath("//*[@class='form-check-input']")
for element in elements:
    if element.is_selected():
        print("Already Selected")
    else:
        element.click()

Answered By - Henry

Read More

[FIXED] Python selenium keep browser open

Issue

I am using selenium to open some browser windows for marketing reasons. I simply open my marketing sources, login via selenium and start working.

The problem is, that after the code is executed selenium closes the window.

All solutions havent helped much so far.

I have 13 browser windows atm., which look like this:

def open_instagram():    
    try:
        # Connect
        chrome_options = webdriver.ChromeOptions()
        chrome_options.add_argument("--incognito")
        browser = webdriver.Chrome('drivers/chromedriver.exe', chrome_options=chrome_options)
        browser.set_window_size(1800, 900)
        browser.get("https://www.instagram.com/accounts/login/?hl=de")
        browser.find_element(By.NAME, 'username').send_keys('MYEMAIL', Keys.TAB, 'MYPW', Keys.ENTER)
    except Exception as e:
        print (e, 'Instagram')

open_instagram()

The closest solution which I found is adding this at the end of my script, but somehow it will only keep 5 windows open, than close 5 windows and open next 5 new windows:

while True:
    pass

I just want that selenium keeps all browser windows open, until I close them manually.

Solution

If you want chrome and chromedriver to stay open, you have to use the 'detach' option when starting chromedriver.

In your case add :

from selenium.webdriver.chrome.options import Options
chrome_options = Options()
chrome_options.add_experimental_option("detach", True)

before you pass in the options to the browser

browser = webdriver.Chrome('drivers/chromedriver.exe', options=chrome_options)

Or you can run the code in debug mode with breakpoint at the end and when it pauses 'kill' the program and take over the browser if you want to, but this works in IDE only.

Update: As of Jan. 24, chrome_options parameter when initializing Webdriver is now only "options".

Answered By - AnkDasCo

Read More

[FIXED] webscraping without pagination- no visible pages

Issue

I am trying to webscrape a podcast site, I received the first 10 items, but there are no pages in the bottom, thus I cannot loop pages, the scroll down button add new items, what should be the approach to get all the 400 hundreds items, thank you in advance

here is my code :

import time
import pandas as pd
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import Select
import time
import pandas as pd
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import Select
from selenium.webdriver.chrome.options import Options
options = Options()  # Initialize an instance of the Options class
options.headless = True  # True -> Headless mode activated
options.add_argument('window-size=1920x1080')  

website="https://www.osimhistoria.com/osimhistoria"
driver = webdriver.Chrome(options=options)
    #driver=webdriver.Chrome(path) dont need it 
driver.get(website)
    #driver.maximize_window()
#path='/backup_zachi/Python/webscrape/chromedriver'
driver = webdriver.Chrome()
#driver=webdriver.Chrome(path) dont need it 
driver.get(website)
driver.maximize_window()
container = driver.find_element(By.CLASS_NAME,'VM7gjN')
products = container.find_elements(By.XPATH,'//*[@id="comp-kgup2wfz"]')
for product in products:
    print(product.text)

Solution

u should scroll down until there are no more items to display. Here is a function u can use to scroll down. Then return to the top of the page and then scrape the data

from selenium import webdriver
import time

def scroll_to_bottom(driver, wait=5):
    """
    Scrolls to the bottom of the webpage.

    Parameters:
    driver (webdriver): The Selenium WebDriver instance.
    wait (int): The number of seconds to wait after reaching the bottom.
    """
    last_height = driver.execute_script("return document.body.scrollHeight")

    while True:
        # Scroll down to the bottom.
        driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")

        # Wait to load the page.
        time.sleep(wait)

        # Calculate new scroll height and compare with last scroll height.
        new_height = driver.execute_script("return document.body.scrollHeight")
        if new_height == last_height:
            break
        last_height = new_height

Answered By - Théo Pantecouteau

Read More

[FIXED] Launching the custom profile in Chrome for testing version

Issue

I am trying to run the following code, but I encounter a Driver's SessionNotCreatedException error.

Could you please have any ideaas how to specify a ChromeProfile when using Chrome For Testing?

Error Code: Exception has occurred: SessionNotCreatedException Message: session not created: Chrome failed to start: exited normally. (session not created: DevToolsActivePort file doesn't exist) (The process started from chrome location C:\Windows\System32\chrome\win64-120.0.6099.109\chrome-win64\chrome.exe is no longer running, so ChromeDriver is assuming that Chrome has crashed.)

My Code:

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import time

# set specify my profile
chrome_options = Options()
chrome_options.add_argument(r'user-data-dir=C:\Users\****\AppData\Local\Google\Chrome\User Data')
chrome_options.add_argument('profile-directory=Profile 05')

# set Chrome folder
chrome_options.binary_location = r'C:\Windows\System32\chrome\win64-120.0.6099.109\chrome-win64\chrome.exe'

# set ChromeDriver folder
driver_path = r'C:\Windows\System32\chromedriver\win64-120.0.6099.109\chromedriver-win64\chromedriver.exe'

# launch browser
driver = webdriver.Chrome(executable_path = driver_path, options=chrome_options)
# Open google
driver.get('https://www.google.co.jp/')

# waiting
time.sleep(3)

# close browser
driver.close()

I checked them below.. but seems not solved

• ・right folder pass
• ・could launch any website without specifying profile options. 　→it has launch Guest Mode. I want to launch with my profile.(not login)
• ・Im using same version(Chrome, Chrome Driver)

Solution

You can find the user data dir of Chrome for Testing for macOS and Linux here.

For Windows there's an extra layer of indirection, c.f. here.

Basically, change

chrome_options.add_argument(r'user-data-dir=C:\Users\****\AppData\Local\Google\Chrome\User Data')

to

chrome_options.add_argument(r'user-data-dir=C:\Users\****\AppData\Local\Google\Chrome for Testing\User Data')

Answered By - thiagowfx

Read More

[FIXED] Asyncronous programming in machine learning

Issue

I would like to share my plan and idea for applying asynchronous programming with machine learning , so here is my idea : let us consider following code :

import pandas as pd
from sklearn.ensemble import  VotingClassifier
from sklearn.svm import SVC
from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
import time
from  sklearn.model_selection import  train_test_split
from sklearn.preprocessing import LabelEncoder
mylabel =LabelEncoder()
data =pd.read_csv("https://raw.githubusercontent.com/datasciencedojo/datasets/master/titanic.csv",usecols=lambda x: x not in ["PassengerId","Ticket"])
# print(data.columns)

st = time.time()


#Process the data
data['Name'] =data['Name'].map(lambda x:x.split(',')[1].split('.')[0])
data['Age'] =data['Age'].fillna(data['Age'].mean())
data['Sex'] =mylabel.fit_transform(data['Sex'])
data['Cabin'] =data['Cabin'].fillna(data['Cabin'].value_counts().index[0])
data['Cabin'] =data['Cabin'].map(lambda x:x[0])
data['Embarked'] =data['Embarked'].fillna(data['Embarked'].value_counts().index[0])
# print(data.isnull().any())
# print(data.dtypes)
#convert categorical columns
cat_columns =data.select_dtypes(include=['object']).columns
print(cat_columns)
data[cat_columns] = data[cat_columns].apply(LabelEncoder().fit_transform)



#seperate  and divide into train and test

y =data['Survived'].values
X =data.drop('Survived',axis=1).values
X_train,X_test,y_train,y_test =train_test_split(X,y,test_size=0.2,random_state=1)
eclf1 = VotingClassifier(estimators=[('svc', SVC(probability=True)), ('LR', LogisticRegression(max_iter=5000)),
                                     ('cart', DecisionTreeClassifier())], voting='soft')
eclf1.fit(X_train,y_train)
print(eclf1.score(X_test,y_test))

# get the end time
et = time.time()
# get the execution time
elapsed_time = et - st
print('Execution time:', elapsed_time, 'seconds')
# print(data.head())

This is standard machine learning code for applying VotingClassifier algorithm for titanic data (which is by the way very small in size), my algorithm took approximately :

Execution time: 0.10833930969238281 seconds

That is pretty small right? So the code runs very fast, but one main reason is the dataset is very small in size, what about if we have dataset with rows of millions or even billions or even size outside of human imagination? During this week, I have started learning of asyncio library and here is corresponding official python web link :asyncronous

now if we look that process the data line :

#Process the data
data['Name'] =data['Name'].map(lambda x:x.split(',')[1].split('.')[0])
data['Age'] =data['Age'].fillna(data['Age'].mean())
data['Sex'] =mylabel.fit_transform(data['Sex'])
data['Cabin'] =data['Cabin'].fillna(data['Cabin'].value_counts().index[0])
data['Cabin'] =data['Cabin'].map(lambda x:x[0])
data['Embarked'] =data['Embarked'].fillna(data['Embarked'].value_counts().index[0])

Definitely we can divide it into portions and run code in parallel right? we should not wait for any column and in parallel process any other columns ,my question is how can I do it?

I have another question :

eclf1 = VotingClassifier(estimators=[('svc', SVC(probability=True)), ('LR', LogisticRegression(max_iter=5000)),
                                     ('cart', DecisionTreeClassifier())], voting='soft')
eclf1.fit(X_train,y_train)

Can I apply asyncio to votingclassifier? or should I train those algorithms separately and then apply mode function to the predictions? Please if you can , give just a small examples in order to digest those details.

Solution

It goes something like this, but I haven't tested the code (sorry); If you really want it async, you will need to:

1. have a beefy computer (which I do not)
2. merge the params of multiple classifiers each trained async.

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.datasets import load_iris
from sklearn.ensemble import VotingClassifier
from sklearn.svm import SVC
from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import accuracy_score
from concurrent.futures import ThreadPoolExecutor

def split_dataframe(df, chunk_size=10000):
    chunks = []
    num_chunks = len(df) // chunk_size + 1
    for i in range(num_chunks):
        chunks.append(df[i * chunk_size:(i + 1) * chunk_size])
    return chunks

def fit_classifier_async(classifier, train_data, train_labels):
    classifier.fit(train_data, train_labels)
    return classifier

# Load the Iris dataset
iris = load_iris()
X = pd.DataFrame(iris.data, columns=iris.feature_names)
y = pd.Series(iris.target, name='target')

# Split the data into training and testing sets
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Generate an example DataFrame (you can replace this with your actual data)
# For simplicity, I'll use the first 100 samples of the Iris dataset
example_df = pd.concat([X_train.iloc[:50], X_train.iloc[50:100], X_train.iloc[100:150]], ignore_index=True)
example_labels = pd.concat([y_train.iloc[:50], y_train.iloc[50:100], y_train.iloc[100:150]], ignore_index=True)

# Create individual classifiers
svc_classifier = SVC(probability=True)
lr_classifier = LogisticRegression(max_iter=5000)
cart_classifier = DecisionTreeClassifier()

# Create the VotingClassifier
eclf1 = VotingClassifier(estimators=[
    ('svc', svc_classifier),
    ('LR', lr_classifier),
    ('cart', cart_classifier)
], voting='soft')

# Split the dataset into chunks
chunks = split_dataframe(example_df)

# Asynchronously fit the classifier on each chunk
with ThreadPoolExecutor() as executor:     # set the number of threads here
    fitted_classifiers = list(executor.map(
        fit_classifier_async,
        [eclf1] * len(chunks),  # Use the same classifier for each chunk
        chunks,
        [example_labels] * len(chunks)  # Use the same labels for each chunk
    ))

# Combine the fitted classifiers into a single ensemble model
for fitted_classifier in fitted_classifiers:
    eclf1.estimators_ += fitted_classifier.estimators_

# Make predictions on the test set
predictions = eclf1.predict(X_test)

# Evaluate accuracy
accuracy = accuracy_score(y_test, predictions)
print(f'Accuracy: {accuracy:.2f}')

Answered By - Sy Ker

Read More

[FIXED] Does `await` in Python yield to the event loop?

Issue

I was wondering what exactly happens when we await a coroutine in async Python code, for example:

await send_message(string)

(1) send_message is added to the event loop, and the calling coroutine gives up control to the event loop, or

(2) We jump directly into send_message

Most explanations I read point to (1), as they describe the calling coroutine as exiting. But my own experiments suggest (2) is the case: I tried to have a coroutine run after the caller but before the callee and could not achieve this.

Solution

No, await (per se) does not yield to the event loop, yield yields to the event loop, hence for the case given: "(2) We jump directly into send_message". In particular, certain yield expressions are the only points, at bottom, where async tasks can actually be switched out (in terms of nailing down the precise spot where Python code execution can be suspended).

To be proven and demonstrated: 1) by theory/documentation, 2) by implementation code, 3) by example.

By theory/documentation

PEP 492: Coroutines with async and await syntax

While the PEP is not tied to any specific Event Loop implementation, it is relevant only to the kind of coroutine that uses yield as a signal to the scheduler, indicating that the coroutine will be waiting until an event (such as IO) is completed. ...

[await] uses the yield from implementation [with an extra step of validating its argument.] ...

Any yield from chain of calls ends with a yield. This is a fundamental mechanism of how Futures are implemented. Since, internally, coroutines are a special kind of generators, every await is suspended by a yield somewhere down the chain of await calls (please refer to PEP 3156 for a detailed explanation). ...

Coroutines are based on generators internally, thus they share the implementation. Similarly to generator objects, coroutines have throw(), send() and close() methods. ...

The vision behind existing generator-based coroutines and this proposal is to make it easy for users to see where the code might be suspended.

In context, "easy for users to see where the code might be suspended" seems to refer to the fact that in synchronous code yield is the place where execution can be "suspended" within a routine allowing other code to run, and that principle now extends perfectly to the async context wherein a yield (if its value is not consumed within the running task but is propagated up to the scheduler) is the "signal to the scheduler" to switch out tasks.

More succinctly: where does a generator yield control? At a yield. Coroutines (including those using async and await syntax) are generators, hence likewise.

And it is not merely an analogy, in implementation (see below) the actual mechanism by which a task gets "into" and "out of" coroutines is not anything new, magical, or unique to the async world, but simply by calling the coro's <generator>.send() method. That was (as I understand the text) part of the "vision" behind PEP 492: async and await would provide no novel mechanism for code suspension but just pour async-sugar on Python's already well-beloved and powerful generators.

And PEP 3156: The "asyncio" module

The loop.slow_callback_duration attribute controls the maximum execution time allowed between two yield points before a slow callback is reported [emphasis in original].

That is, an uninterrupted segment of code (from the async perspective) is demarcated as that between two successive yield points (whose values reached up to the running Task level (via an await/yield from tunnel) without being consumed within it).

And this:

The scheduler has no public interface. You interact with it by using yield from future and yield from task.

Objection: "That says 'yield from', but you're trying to argue that the task can only switch out at a yield itself! yield from and yield are different things, my friend, and yield from itself doesn't suspend code!"

Ans: Not a contradiction. The PEP is saying you interact with the scheduler by using yield from future/task. But as noted above in PEP 492, any chain of yield from (~aka await) ultimately reaches a yield (the "bottom turtle"). In particular (see below), yield from future does in fact yield that same future after some wrapper work, and that yield is the actual "switch out point" where another task takes over. But it is incorrect for your code to directly yield a Future up to the current Task because you would bypass the necessary wrapper.

The objection having been answered, and its practical coding considerations being noted, the point I wish to make from the above quote remains: that a suitable yield in Python async code is ultimately the one thing which, having suspended code execution in the standard way that any other yield would do, now futher engages the scheduler to bring about a possible task switch.

By implementation code

asyncio/futures.py

class Future:
...
    def __await__(self):
        if not self.done():
            self._asyncio_future_blocking = True
            yield self  # This tells Task to wait for completion.
        if not self.done():
            raise RuntimeError("await wasn't used with future")
        return self.result()  # May raise too.

    __iter__ = __await__  # make compatible with 'yield from'.

Paraphrase: The line yield self is what tells the running task to sit out for now and let other tasks run, coming back to this one sometime after self is done.

Almost all of your awaitables in asyncio world are (multiple layers of) wrappers around a Future. The event loop remains utterly blind to all higher level await awaitable expressions until the code execution trickles down to an await future or yield from future and then (as seen here) calls yield self, which yielded self is then "caught" by none other than the Task under which the present coroutine stack is running thereby signaling to the task to take a break.

Possibly the one and only exception to the above "code suspends at yield self within await future" rule, in an asyncio context, is the potential use of a bare yield such as in asyncio.sleep(0). And since the sleep function is a topic of discourse in the comments of this post, let's look at that.

asyncio/tasks.py

@types.coroutine
def __sleep0():
    """Skip one event loop run cycle.
    This is a private helper for 'asyncio.sleep()', used
    when the 'delay' is set to 0.  It uses a bare 'yield'
    expression (which Task.__step knows how to handle)
    instead of creating a Future object.
    """
    yield


async def sleep(delay, result=None, *, loop=None):
    """Coroutine that completes after a given time (in seconds)."""
    if delay <= 0:
        await __sleep0()
        return result

    if loop is None:
        loop = events.get_running_loop()
    else:
        warnings.warn("The loop argument is deprecated since Python 3.8, "
                      "and scheduled for removal in Python 3.10.",
                      DeprecationWarning, stacklevel=2)

    future = loop.create_future()
    h = loop.call_later(delay,
                        futures._set_result_unless_cancelled,
                        future, result)
    try:
        return await future

    finally:
        h.cancel()

Note: We have here the two interesting cases at which control can shift to the scheduler:

(1) The bare yield in __sleep0 (when called via an await).

(2) The yield self immediately within await future.

The crucial line (for our purposes) in asyncio/tasks.py is when Task._step runs its top-level coroutine via result = self._coro.send(None) and recognizes fourish cases:

(1) result = None is generated by the coro (which, again, is a generator): the task "relinquishes control for one event loop iteration".

(2) result = future is generated within the coro, with further magic member field evidence that the future was yielded in a proper manner from out of Future.__iter__ == Future.__await__: the task relinquishes control to the event loop until the future is complete.

(3) A StopIteration is raised by the coro indicating the coroutine completed (i.e. as a generator it exhausted all its yields): the final result of the task (which is itself a Future) is set to the coroutine return value.

(4) Any other Exception occurs: the task's set_exception is set accordingly.

Modulo details, the main point for our concern is that coroutine segments in an asyncio event loop ultimately run via coro.send(). Initial startup and final termination aside, send() proceeds precisely from the last yield value it generated to the next one.

By example

import asyncio
import types

def task_print(s):
    print(f"{asyncio.current_task().get_name()}: {s}")

async def other_task(s):
    task_print(s)

class AwaitableCls:
    def __await__(self):
        task_print("    'Jumped straight into' another `await`; the act of `await awaitable` *itself* doesn't 'pause' anything")
        yield
        task_print("    We're back to our awaitable object because that other task completed")
        asyncio.create_task(other_task("The event loop gets control when `yield` points (from an iterable coroutine) propagate up to the `current_task` through a suitable chain of `await` or `yield from` statements"))

async def coro():
    task_print("  'Jumped straight into' coro; the `await` keyword itself does nothing to 'pause' the current_task")
    await AwaitableCls()
    task_print("  'Jumped straight back into' coro; we have another pending task, but leaving an `__await__` doesn't 'pause' the task any more than entering the `__await__` does")

@types.coroutine
def iterable_coro(context):
    task_print(f"`{context} iterable_coro`: pre-yield")
    yield None # None or a Future object are the only legitimate yields to the task in asyncio
    task_print(f"`{context} iterable_coro`: post-yield")

async def original_task():
    asyncio.create_task(other_task("Aha, but a (suitably unconsumed) *`yield`* DOES 'pause' the current_task allowing the event scheduler to `_wakeup` another task"))

    task_print("Original task")
    await coro()
    task_print("'Jumped straight out of' coro. Leaving a coro, as with leaving/entering any awaitable, doesn't give control to the event loop")
    res = await iterable_coro("await")
    assert res is None
    asyncio.create_task(other_task("This doesn't run until the very end because the generated None following the creation of this task is consumed by the `for` loop"))
    for y in iterable_coro("for y in"):
        task_print(f"But 'ordinary' `yield` points (those which are consumed by the `current_task` itself) behave as ordinary without relinquishing control at the async/task-level; `y={y}`")
    task_print("Done with original task")

asyncio.get_event_loop().run_until_complete(original_task())

run in python3.8 produces

Task-1: Original task

Task-1: 'Jumped straight into' coro; the await keyword itself does nothing to 'pause' the current_task

Task-1: 'Jumped straight into' another await; the act of await awaitable itself doesn't 'pause' anything

Task-2: Aha, but a (suitably unconsumed) yield DOES 'pause' the current_task allowing the event scheduler to _wakeup another task

Task-1: We're back to our awaitable object because that other task completed

Task-1: 'Jumped straight back into' coro; we have another pending task, but leaving an __await__ doesn't 'pause' the task any more than entering the __await__ does

Task-1: 'Jumped straight out of' coro. Leaving a coro, as with leaving/entering any awaitable, doesn't give control to the event loop

Task-1: await iterable_coro: pre-yield

Task-3: The event loop gets control when yield points (from an iterable coroutine) propagate up to the current_task through a suitable chain of await or yield from statements

Task-1: await iterable_coro: post-yield

Task-1: for y in iterable_coro: pre-yield

Task-1: But 'ordinary' yield points (those which are consumed by the current_task itself) behave as ordinary without relinquishing control at the async/task-level; y=None

Task-1: for y in iterable_coro: post-yield

Task-1: Done with original task

Task-4: This doesn't run until the very end because the generated None following the creation of this task is consumed by the for loop

Indeed, exercises such as the following can help one's mind to decouple the functionality of async/await from notion of "event loops" and such. The former is conducive to nice implementations and usages of the latter, but you can use async and await just as specially syntaxed generator stuff without any "loop" (whether asyncio or otherwise) whatsoever:

import types # no asyncio, nor any other loop framework

async def f1():
    print(1)
    print(await f2(),'= await f2()')
    return 8

@types.coroutine
def f2():
    print(2)
    print((yield 3),'= yield 3')
    return 7

class F3:
   def __await__(self):
        print(4)
        print((yield 5),'= yield 5')
        print(10)
        return 11

task1 = f1()
task2 = F3().__await__()
""" You could say calls to send() represent our
   "manual task management" in this script.
"""
print(task1.send(None), '= task1.send(None)')
print(task2.send(None), '= task2.send(None)')
try:
    print(task1.send(6), 'try task1.send(6)')
except StopIteration as e:
    print(e.value, '= except task1.send(6)')
try:
    print(task2.send(9), 'try task2.send(9)')
except StopIteration as e:
    print(e.value, '= except task2.send(9)')

produces

1

2

3 = task1.send(None)

4

5 = task2.send(None)

6 = yield 3

7 = await f2()

8 = except task1.send(6)

9 = yield 5

10

11 = except task2.send(9)

Answered By - Zach Harris

Read More

[FIXED] Is it possible to run python variable declarations from a string?

Issue

This is the behaviour I'm looking for...

"a = 2" # execute this line
print a
> 2

I know about the exec statement but I need it to work in python 2 and 3 and python 3 does not allow exec to create local variables. Is there another way around this?

EDIT: Like I said - I know that I can't use exec, the accepted answer of the supposed duplicate is saying to use exec.

Solution

I dont necessarilly think this is a good idea ...

import ast
def parse_assignment(s):
    lhs,rhs = s.split("=",1)
    globals()[lhs] = ast.literal_eval(rhs)

parse_assignment("hello=5")
print(hello)
parse_assignment("hello2='words'")
print(hello2)
parse_assignment("hello3=[1,'hello']")
print(hello3)

https://repl.it/repls/BiodegradableLiveProgrammer

Answered By - Joran Beasley

Read More

[FIXED] Installing Pip on 2.4

Issue

How do I install pip on python 2.4? I've tried downloading the get-pip.py file, but it doesn't run on version 2.4 of python. Is there anyway to edit the file to run it? Or is there another download?

Solution

Python 2.4 was supported through v1.1.

Try this: https://pypi.python.org/pypi/pip/1.1

Answered By - teoreda

Read More

[FIXED] SQLite, python, unicode, and non-utf data

Issue

I started by trying to store strings in sqlite using python, and got the message:

sqlite3.ProgrammingError: You must not use 8-bit bytestrings unless you use a text_factory that can interpret 8-bit bytestrings (like text_factory = str). It is highly recommended that you instead just switch your application to Unicode strings.

Ok, I switched to Unicode strings. Then I started getting the message:

sqlite3.OperationalError: Could not decode to UTF-8 column 'tag_artist' with text 'Sigur Rós'

when trying to retrieve data from the db. More research and I started encoding it in utf8, but then 'Sigur Rós' starts looking like 'Sigur Rós'

note: My console was set to display in 'latin_1' as @John Machin pointed out.

What gives? After reading this, describing exactly the same situation I'm in, it seems as if the advice is to ignore the other advice and use 8-bit bytestrings after all.

I didn't know much about unicode and utf before I started this process. I've learned quite a bit in the last couple hours, but I'm still ignorant of whether there is a way to correctly convert 'ó' from latin-1 to utf-8 and not mangle it. If there isn't, why would sqlite 'highly recommend' I switch my application to unicode strings?

---

I'm going to update this question with a summary and some example code of everything I've learned in the last 24 hours so that someone in my shoes can have an easy(er) guide. If the information I post is wrong or misleading in any way please tell me and I'll update, or one of you senior guys can update.

---

Summary of answers

Let me first state the goal as I understand it. The goal in processing various encodings, if you are trying to convert between them, is to understand what your source encoding is, then convert it to unicode using that source encoding, then convert it to your desired encoding. Unicode is a base and encodings are mappings of subsets of that base. utf_8 has room for every character in unicode, but because they aren't in the same place as, for instance, latin_1, a string encoded in utf_8 and sent to a latin_1 console will not look the way you expect. In python the process of getting to unicode and into another encoding looks like:

str.decode('source_encoding').encode('desired_encoding')

or if the str is already in unicode

str.encode('desired_encoding')

For sqlite I didn't actually want to encode it again, I wanted to decode it and leave it in unicode format. Here are four things you might need to be aware of as you try to work with unicode and encodings in python.

1. The encoding of the string you want to work with, and the encoding you want to get it to.
2. The system encoding.
3. The console encoding.
4. The encoding of the source file

Elaboration:

(1) When you read a string from a source, it must have some encoding, like latin_1 or utf_8. In my case, I'm getting strings from filenames, so unfortunately, I could be getting any kind of encoding. Windows XP uses UCS-2 (a Unicode system) as its native string type, which seems like cheating to me. Fortunately for me, the characters in most filenames are not going to be made up of more than one source encoding type, and I think all of mine were either completely latin_1, completely utf_8, or just plain ascii (which is a subset of both of those). So I just read them and decoded them as if they were still in latin_1 or utf_8. It's possible, though, that you could have latin_1 and utf_8 and whatever other characters mixed together in a filename on Windows. Sometimes those characters can show up as boxes, other times they just look mangled, and other times they look correct (accented characters and whatnot). Moving on.

(2) Python has a default system encoding that gets set when python starts and can't be changed during runtime. See here for details. Dirty summary ... well here's the file I added:

\# sitecustomize.py  
\# this file can be anywhere in your Python path,  
\# but it usually goes in ${pythondir}/lib/site-packages/  
import sys  
sys.setdefaultencoding('utf_8')  

This system encoding is the one that gets used when you use the unicode("str") function without any other encoding parameters. To say that another way, python tries to decode "str" to unicode based on the default system encoding.

(3) If you're using IDLE or the command-line python, I think that your console will display according to the default system encoding. I am using pydev with eclipse for some reason, so I had to go into my project settings, edit the launch configuration properties of my test script, go to the Common tab, and change the console from latin-1 to utf-8 so that I could visually confirm what I was doing was working.

(4) If you want to have some test strings, eg

test_str = "ó"

in your source code, then you will have to tell python what kind of encoding you are using in that file. (FYI: when I mistyped an encoding I had to ctrl-Z because my file became unreadable.) This is easily accomplished by putting a line like so at the top of your source code file:

# -*- coding: utf_8 -*-

If you don't have this information, python attempts to parse your code as ascii by default, and so:

SyntaxError: Non-ASCII character '\xf3' in file _redacted_ on line 81, but no encoding declared; see http://www.python.org/peps/pep-0263.html for details

Once your program is working correctly, or, if you aren't using python's console or any other console to look at output, then you will probably really only care about #1 on the list. System default and console encoding are not that important unless you need to look at output and/or you are using the builtin unicode() function (without any encoding parameters) instead of the string.decode() function. I wrote a demo function I will paste into the bottom of this gigantic mess that I hope correctly demonstrates the items in my list. Here is some of the output when I run the character 'ó' through the demo function, showing how various methods react to the character as input. My system encoding and console output are both set to utf_8 for this run:

'�' = original char <type 'str'> repr(char)='\xf3'
'?' = unicode(char) ERROR: 'utf8' codec can't decode byte 0xf3 in position 0: unexpected end of data
'ó' = char.decode('latin_1') <type 'unicode'> repr(char.decode('latin_1'))=u'\xf3'
'?' = char.decode('utf_8')  ERROR: 'utf8' codec can't decode byte 0xf3 in position 0: unexpected end of data

Now I will change the system and console encoding to latin_1, and I get this output for the same input:

'ó' = original char <type 'str'> repr(char)='\xf3'
'ó' = unicode(char) <type 'unicode'> repr(unicode(char))=u'\xf3'
'ó' = char.decode('latin_1') <type 'unicode'> repr(char.decode('latin_1'))=u'\xf3'
'?' = char.decode('utf_8')  ERROR: 'utf8' codec can't decode byte 0xf3 in position 0: unexpected end of data

Notice that the 'original' character displays correctly and the builtin unicode() function works now.

Now I change my console output back to utf_8.

'�' = original char <type 'str'> repr(char)='\xf3'
'�' = unicode(char) <type 'unicode'> repr(unicode(char))=u'\xf3'
'�' = char.decode('latin_1') <type 'unicode'> repr(char.decode('latin_1'))=u'\xf3'
'?' = char.decode('utf_8')  ERROR: 'utf8' codec can't decode byte 0xf3 in position 0: unexpected end of data

Here everything still works the same as last time but the console can't display the output correctly. Etc. The function below also displays more information that this and hopefully would help someone figure out where the gap in their understanding is. I know all this information is in other places and more thoroughly dealt with there, but I hope that this would be a good kickoff point for someone trying to get coding with python and/or sqlite. Ideas are great but sometimes source code can save you a day or two of trying to figure out what functions do what.

Disclaimers: I'm no encoding expert, I put this together to help my own understanding. I kept building on it when I should have probably started passing functions as arguments to avoid so much redundant code, so if I can I'll make it more concise. Also, utf_8 and latin_1 are by no means the only encoding schemes, they are just the two I was playing around with because I think they handle everything I need. Add your own encoding schemes to the demo function and test your own input.

One more thing: there are apparently crazy application developers making life difficult in Windows.

#!/usr/bin/env python
# -*- coding: utf_8 -*-

import os
import sys

def encodingDemo(str):
    validStrings = ()
    try:        
        print "str =",str,"{0} repr(str) = {1}".format(type(str), repr(str))
        validStrings += ((str,""),)
    except UnicodeEncodeError as ude:
        print "Couldn't print the str itself because the console is set to an encoding that doesn't understand some character in the string.  See error:\n\t",
        print ude
    try:
        x = unicode(str)
        print "unicode(str) = ",x
        validStrings+= ((x, " decoded into unicode by the default system encoding"),)
    except UnicodeDecodeError as ude:
        print "ERROR.  unicode(str) couldn't decode the string because the system encoding is set to an encoding that doesn't understand some character in the string."
        print "\tThe system encoding is set to {0}.  See error:\n\t".format(sys.getdefaultencoding()),  
        print ude
    except UnicodeEncodeError as uee:
        print "ERROR.  Couldn't print the unicode(str) because the console is set to an encoding that doesn't understand some character in the string.  See error:\n\t",
        print uee
    try:
        x = str.decode('latin_1')
        print "str.decode('latin_1') =",x
        validStrings+= ((x, " decoded with latin_1 into unicode"),)
        try:        
            print "str.decode('latin_1').encode('utf_8') =",str.decode('latin_1').encode('utf_8')
            validStrings+= ((x, " decoded with latin_1 into unicode and encoded into utf_8"),)
        except UnicodeDecodeError as ude:
            print "The string was decoded into unicode using the latin_1 encoding, but couldn't be encoded into utf_8.  See error:\n\t",
            print ude
    except UnicodeDecodeError as ude:
        print "Something didn't work, probably because the string wasn't latin_1 encoded.  See error:\n\t",
        print ude
    except UnicodeEncodeError as uee:
        print "ERROR.  Couldn't print the str.decode('latin_1') because the console is set to an encoding that doesn't understand some character in the string.  See error:\n\t",
        print uee
    try:
        x = str.decode('utf_8')
        print "str.decode('utf_8') =",x
        validStrings+= ((x, " decoded with utf_8 into unicode"),)
        try:        
            print "str.decode('utf_8').encode('latin_1') =",str.decode('utf_8').encode('latin_1')
        except UnicodeDecodeError as ude:
            print "str.decode('utf_8').encode('latin_1') didn't work.  The string was decoded into unicode using the utf_8 encoding, but couldn't be encoded into latin_1.  See error:\n\t",
            validStrings+= ((x, " decoded with utf_8 into unicode and encoded into latin_1"),)
            print ude
    except UnicodeDecodeError as ude:
        print "str.decode('utf_8') didn't work, probably because the string wasn't utf_8 encoded.  See error:\n\t",
        print ude
    except UnicodeEncodeError as uee:
        print "ERROR.  Couldn't print the str.decode('utf_8') because the console is set to an encoding that doesn't understand some character in the string.  See error:\n\t",uee

    print
    print "Printing information about each character in the original string."
    for char in str:
        try:
            print "\t'" + char + "' = original char {0} repr(char)={1}".format(type(char), repr(char))
        except UnicodeDecodeError as ude:
            print "\t'?' = original char  {0} repr(char)={1} ERROR PRINTING: {2}".format(type(char), repr(char), ude)
        except UnicodeEncodeError as uee:
            print "\t'?' = original char  {0} repr(char)={1} ERROR PRINTING: {2}".format(type(char), repr(char), uee)
            print uee    

        try:
            x = unicode(char)        
            print "\t'" + x + "' = unicode(char) {1} repr(unicode(char))={2}".format(x, type(x), repr(x))
        except UnicodeDecodeError as ude:
            print "\t'?' = unicode(char) ERROR: {0}".format(ude)
        except UnicodeEncodeError as uee:
            print "\t'?' = unicode(char)  {0} repr(char)={1} ERROR PRINTING: {2}".format(type(x), repr(x), uee)

        try:
            x = char.decode('latin_1')
            print "\t'" + x + "' = char.decode('latin_1') {1} repr(char.decode('latin_1'))={2}".format(x, type(x), repr(x))
        except UnicodeDecodeError as ude:
            print "\t'?' = char.decode('latin_1')  ERROR: {0}".format(ude)
        except UnicodeEncodeError as uee:
            print "\t'?' = char.decode('latin_1')  {0} repr(char)={1} ERROR PRINTING: {2}".format(type(x), repr(x), uee)

        try:
            x = char.decode('utf_8')
            print "\t'" + x + "' = char.decode('utf_8') {1} repr(char.decode('utf_8'))={2}".format(x, type(x), repr(x))
        except UnicodeDecodeError as ude:
            print "\t'?' = char.decode('utf_8')  ERROR: {0}".format(ude)
        except UnicodeEncodeError as uee:
            print "\t'?' = char.decode('utf_8')  {0} repr(char)={1} ERROR PRINTING: {2}".format(type(x), repr(x), uee)

        print

x = 'ó'
encodingDemo(x)

Much thanks for the answers below and especially to @John Machin for answering so thoroughly.

Solution

I'm still ignorant of whether there is a way to correctly convert 'ó' from latin-1 to utf-8 and not mangle it

repr() and unicodedata.name() are your friends when it comes to debugging such problems:

>>> oacute_latin1 = "\xF3"
>>> oacute_unicode = oacute_latin1.decode('latin1')
>>> oacute_utf8 = oacute_unicode.encode('utf8')
>>> print repr(oacute_latin1)
'\xf3'
>>> print repr(oacute_unicode)
u'\xf3'
>>> import unicodedata
>>> unicodedata.name(oacute_unicode)
'LATIN SMALL LETTER O WITH ACUTE'
>>> print repr(oacute_utf8)
'\xc3\xb3'
>>>

If you send oacute_utf8 to a terminal that is set up for latin1, you will get A-tilde followed by superscript-3.

I switched to Unicode strings.

What are you calling Unicode strings? UTF-16?

What gives? After reading this, describing exactly the same situation I'm in, it seems as if the advice is to ignore the other advice and use 8-bit bytestrings after all.

I can't imagine how it seems so to you. The story that was being conveyed was that unicode objects in Python and UTF-8 encoding in the database were the way to go. However Martin answered the original question, giving a method ("text factory") for the OP to be able to use latin1 -- this did NOT constitute a recommendation!

Update in response to these further questions raised in a comment:

I didn't understand that the unicode characters still contained an implicit encoding. Am I saying that right?

No. An encoding is a mapping between Unicode and something else, and vice versa. A Unicode character doesn't have an encoding, implicit or otherwise.

It looks to me like unicode("\xF3") and "\xF3".decode('latin1') are the same when evaluated with repr().

Say what? It doesn't look like it to me:

>>> unicode("\xF3")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
UnicodeDecodeError: 'ascii' codec can't decode byte 0xf3 in position 0: ordinal
not in range(128)
>>> "\xF3".decode('latin1')
u'\xf3'
>>>

Perhaps you meant: u'\xf3' == '\xF3'.decode('latin1') ... this is certainly true.

It is also true that unicode(str_object, encoding) does the same as str_object.decode(encoding) ... including blowing up when an inappropriate encoding is supplied.

Is that a happy circumstance

That the first 256 characters in Unicode are the same, code for code, as the 256 characters in latin1 is a good idea. Because all 256 possible latin1 characters are mapped to Unicode, it means that ANY 8-bit byte, ANY Python str object can be decoded into unicode without an exception being raised. This is as it should be.

However there exist certain persons who confuse two quite separate concepts: "my script runs to completion without any exceptions being raised" and "my script is error-free". To them, latin1 is "a snare and a delusion".

In other words, if you have a file that's actually encoded in cp1252 or gbk or koi8-u or whatever and you decode it using latin1, the resulting Unicode will be utter rubbish and Python (or any other language) will not flag an error -- it has no way of knowing that you have commited a silliness.

or is unicode("str") going to always return the correct decoding?

Just like that, with the default encoding being ascii, it will return the correct unicode if the file is actually encoded in ASCII. Otherwise, it'll blow up.

Similarly, if you specify the correct encoding, or one that's a superset of the correct encoding, you'll get the correct result. Otherwise you'll get gibberish or an exception.

In short: the answer is no.

If not, when I receive a python str that has any possible character set in it, how do I know how to decode it?

If the str object is a valid XML document, it will be specified up front. Default is UTF-8. If it's a properly constructed web page, it should be specified up front (look for "charset"). Unfortunately many writers of web pages lie through their teeth (ISO-8859-1 aka latin1, should be Windows-1252 aka cp1252; don't waste resources trying to decode gb2312, use gbk instead). You can get clues from the nationality/language of the website.

UTF-8 is always worth trying. If the data is ascii, it'll work fine, because ascii is a subset of utf8. A string of text that has been written using non-ascii characters and has been encoded in an encoding other than utf8 will almost certainly fail with an exception if you try to decode it as utf8.

All of the above heuristics and more and a lot of statistics are encapsulated in chardet, a module for guessing the encoding of arbitrary files. It usually works well. However you can't make software idiot-proof. For example, if you concatenate data files written some with encoding A and some with encoding B, and feed the result to chardet, the answer is likely to be encoding C with a reduced level of confidence e.g. 0.8. Always check the confidence part of the answer.

If all else fails:

(1) Try asking here, with a small sample from the front of your data ... print repr(your_data[:400]) ... and whatever collateral info about its provenance that you have.

(2) Recent Russian research into techniques for recovering forgotten passwords appears to be quite applicable to deducing unknown encodings.

Update 2 BTW, isn't it about time you opened up another question ?-)

One more thing: there are apparently characters that Windows uses as Unicode for certain characters that aren't the correct Unicode for that character, so you may have to map those characters to the correct ones if you want to use them in other programs that are expecting those characters in the right spot.

It's not Windows that's doing it; it's a bunch of crazy application developers. You might have more understandably not paraphrased but quoted the opening paragraph of the effbot article that you referred to:

Some applications add CP1252 (Windows, Western Europe) characters to documents marked up as ISO 8859-1 (Latin 1) or other encodings. These characters are not valid ISO-8859-1 characters, and may cause all sorts of problems in processing and display applications.

Background:

The range U+0000 to U+001F inclusive is designated in Unicode as "C0 Control Characters". These exist also in ASCII and latin1, with the same meanings. They include such familar things as carriage return, line feed, bell, backspace, tab, and others that are used rarely.

The range U+0080 to U+009F inclusive is designated in Unicode as "C1 Control Characters". These exist also in latin1, and include 32 characters that nobody outside unicode.org can imagine any possible use for.

Consequently, if you run a character frequency count on your unicode or latin1 data, and you find any characters in that range, your data is corrupt. There is no universal solution; it depends on how it became corrupted. The characters may have the same meaning as the cp1252 characters at the same positions, and thus the effbot's solution will work. In another case that I've been looking at recently, the dodgy characters appear to have been caused by concatenating text files encoded in UTF-8 and another encoding which needed to be deduced based on letter frequencies in the (human) language the files were written in.

Answered By - John Machin

Read More