eshangj/stackoverflow_q_and_a_sample huggingface dataset:
Question: Is there a mechanism to automatically register flows/new flows if a local agent is running, without having to manually run e.g. flow.register(...) on each one? In airflow, I believe they have a process that regularly scans for any files with dag in the name in the specified airflow home folder, then searches them for DAG objects. And if it finds them it loads them so they are accessible through the UI without having to manually 'register' them. Does something similar exist for prefect. So for example if I just created the following file test_flow.py, without necessarily running it or adding flow.run_agent() is there a way for it to just be magically registered and accessible through the UI :) - just by it simply existing in the proper place? # prefect_home_folder/test_flow.py import prefect from prefect import task, Flow @task def hello_task(): logger = prefect.context.get("logger") logger.info("Hello, Cloud!") flow = Flow("hello-flow", tasks=[hello_task]) flow.register(project_name='main')  I could write a script that has similar behavior to the airflow process to scan a folder and register flows at regular intervals, but I wonder if it's a bit hacky or if there is a better solution and I'm justing thinking too much in terms of airflow?

Answer: Great question (and awesome username!) - in short, I suggest you are thinking too much in terms of Airflow. There are a few reasons this is not currently available in Prefect:  explicit is better than implicit Prefect flows are not constrained to live in one place and are not constrained to have the same runtime environments; this makes both the automatic discovery of a flow + re-serializing it complicated from a single agent process (which is not required to share the same runtime environment as the flows it submits) agents are better thought of as being parametrized by deployment infrastructure, not flow storage  Ideally for production workflows you'd use a CI/CD process so that anytime you make a code change an automatic job is triggered that re-registers the flow. A few comments that may be helpful:  you don't actually need to re-register the flow for every possible code change; for example, if you changed the message that your hello_task logs in your example, you could simply re-save the flow to its original location (what this looks like depends on the type of storage you use). Ultimately you only need to re-register if any of the metadata about your flow changes (retry settings, task names, dependency relationships, etc.) you can use flow.register("My Project", idempotency_key=flow.serialized_hash()) to automatically capture this; this pattern will only register a new version if the flow's backend representation changes in some way
Question: I'm getting the following warning reading a geojson with geopanda's read_file(): ...geodataframe.py:422: RuntimeWarning: Sequential read of iterator was interrupted. Resetting iterator. This can negatively impact the performance. for feature in features_lst:  Here's the code sample I used: crime_gdf = gpd.read_file('datasets/crimes.geojson', bbox=bbox)  crimes.geojson is a file containing a large number of points, each with a 'Crime type' bbox defines the boundaries The code runs as expected, but I don't understand that warning. EDIT I converted the geojson to feather, and I get the same warning.

Answer: See my comment in Fiona's issue tracker: https://github.com/Toblerity/Fiona/issues/986 GDAL (the library Fiona uses to access the geodata) maintains an iterator over the features that are currently read. There a some operations that, for some drivers, can influence this iterator. Thus, after such operations we have to ensure that the iterator is set to the correct position that a continuous read of the data is ensured. Such operations include counting all features in a dataset, respectively calculating its extent. There are different types of drivers in GDAL. Some drivers support random access, while some do not. For the drivers that do not support random access, the resetting of the iterator involves reading all features again up to the iterator position. As this is a possible costly operation, this RuntimeWarning is emitted, so that users are aware of this behavior.
Question: I got this while running the selenium webdriver script in python I also set the path in System Environment and also tried downloading the webdriver that matches with my chrome version. And also letest version also. But I still get this error: [8552:6856:1120/155118.770:ERROR:device_event_log_impl.cc(211)] [15:51:18.771] USB: usb_device_handle_win.cc:1020 Failed to read descriptor from node connection: A device attached to the system is not functioning. (0x1F) [8552:6856:1120/155118.774:ERROR:device_event_log_impl.cc(211)] [15:51:18.774] USB: usb_device_handle_win.cc:1020 Failed to read descriptor from node connection: A device attached to the system is not functioning. (0x1F) [8552:6856:1120/155118.821:ERROR:device_event_log_impl.cc(211)] [15:51:18.821] USB: usb_device_handle_win.cc:1020 Failed to read descriptor from node connection: A device attached to the system is not functioning. (0x1F)  I used this in my code : driver = webdriver.Chrome(resource_path("C:\\webdriver\\chromedriver.exe")) # to open the chromebrowser driver.get("https://web.whatsapp.com") 

Answer: After a week of finding an answer to my error, I ended up with a solution that you just need to install pywin32 library and it will not gives you an error open cmd and type pip install pywin32 and you are good to go.....!
Question: I have deployed a fastapi endpoint, from fastapi import FastAPI, UploadFile from typing import List app = FastAPI() @app.post('/work/test') async def testing(files: List(UploadFile)): for i in files: ....... # do a lot of operations on each file # after than I am just writing that processed data into mysql database # cur.execute(...) # cur.commit() ....... # just returning "OK" to confirm data is written into mysql return {"response" : "OK"}  I can request output from the API endpoint and its working fine for me perfectly. Now, the biggest challenge for me to know how much time it is taking for each iteration. Because in the UI part (those who are accessing my API endpoint) I want to help them show a progress bar (TIME TAKEN) for each iteration/file being processed. Is there any possible way for me to achieve it? If so, please help me out on how can I proceed further? Thank you.

Answer: Below is solution which uses uniq identifiers and globally available dictionary which holds information about the jobs: NOTE: Code below is safe to use until you use dynamic keys values ( In sample uuid in use) and keep application within single process.  To start the app create a file main.py Run uvicorn main:app --reload Create job entry by accessing http://127.0.0.1:8000/ Repeat step 3 to create multiple jobs Go to http://127.0.0.1/status page to see page statuses. Go to http://127.0.0.1/status/{identifier} to see progression of the job by the job id.  Code of app: from fastapi import FastAPI, UploadFile import uuid from typing import List import asyncio context = {'jobs': {}} app = FastAPI() async def do_work(job_key, files=None): iter_over = files if files else range(100) for file, file_number in enumerate(iter_over): jobs = context['jobs'] job_info = jobs[job_key] job_info['iteration'] = file_number job_info['status'] = 'inprogress' await asyncio.sleep(1) pending_jobs[job_key]['status'] = 'done' @app.post('/work/test') async def testing(files: List[UploadFile]): identifier = str(uuid.uuid4()) context[jobs][identifier] = {} asyncio.run_coroutine_threadsafe(do_work(identifier, files), loop=asyncio.get_running_loop()) return {"identifier": identifier} @app.get('/') async def get_testing(): identifier = str(uuid.uuid4()) context['jobs'][identifier] = {} asyncio.run_coroutine_threadsafe(do_work(identifier), loop=asyncio.get_running_loop()) return {"identifier": identifier} @app.get('/status') def status(): return { 'all': list(context['jobs'].values()), } @app.get('/status/{identifier}') async def status(identifier): return { "status": context['jobs'].get(identifier, 'job with that identifier is undefined'), }
Question: I have trained two separate models  ModelA: Checks if the input text is related to my work (Binary Classifier [related/not-related]) ModelB: Classifier of related texts (Classifier [good/normal/bad]). Only the related texts are relayed to this model from ModelA  I want  ModelC: Ensemble classifier that outputs [good/normal/bad/not-related] I'll be training in batches. And there can be mix of not-related and good/normal/bad in one batch. I need them separated.  Some pseudo code of what i need # Output of modelA will be a vector I presume `(1, None)` where `None` is batch def ModelC.predict(input): outputA = ModelA(input) if outputA == 'not-related': return outputA return ModelB(outputA)  I don't know how to include the if logic inside the models inference. How can I achieve this?

Answer: Just define your own model. I'm surprised your other models are outputting strings instead of numbers, but without more info this is about all I can give you, so I will assume the output of model A is a string. import tensorflow as tf class ModelC(tf.keras.Model): def __init__(self, A, B): super(ModelC, self).__init__() self.A = A self.B = B def call(self, inputs, training=False): x = self.A(inputs, training) if x == 'not-related': return x return self.B(inputs, training)
Question: I'm using Linux Eclipse (pydev) as IDE to develop python scripts that are launched by an application written in C++. I can debug the python script without problems in the IDE, but the environment is not real (the C++ program sends and receives messages through the stdin/stdout and it's a complex communication channel that I can't fully reproduce writing the messages by hand). Until now I was using log messages to debug (poor man's debug) but it's getting too complex. When I do something similar in PHP I can just leave xdebug listening and add breakpoints in Netbeans. Very neat and easy. Is it possible to do something like that in Python 3.X (with Eclipse or other IDE)? NOTE: I know there is a Pydev / Attach to Process functionality, but it doesn't work. Always fails to attach. NOTE2: There is also a built-in "breakpoint()" in Python 3.7 but it links to a debugger and if also fails, the IDE never gets the control.

Answer: After some research, this is the best option I have found. Without any other solution provided, I post it just in case anyone has the same problem. Python has an integrated debugger: pdb. It works as a module and it doesn't allow to use it if you don't have the window control (i.e. you launch the script). To solve this there are some coders that have created modules that add a layer on pdb. I have tried some and the most easy and still visual interesting is rpudb (but have a look also to this). To install it: pip3 install https://github.com/msbrogli/rpudb/archive/master.zip  (if you install it using the pip3 install rpudb command it will install an old version only valid for python 2) Then, you use it just adding an import and a function call: import rpudb ..... rpudb.set_trace('127.0.0.1', 4444) .....  Launch the program and it will stop in the set_trace call. To debug it (and continue) open a terminal and launch a telnet like this: telnet 127.0.0.1 4444  You will have a visual debugger in front of you with the advantage that you can not only debug local programs, but also remote (just change the IP).
Question: After updating to macOS Big Sur I get the following error when I run jupyter notebook: dyld: Library not loaded: /System/Library/Frameworks/CoreFoundation.framework/Versions/A/CoreFoundation Referenced from: /Library/Frameworks/Python.framework/Versions/3.6/Resources/Python.app/Contents/MacOS/Python Reason: image not found [1] 2971 abort jupyter notebook  Any idea as to how to solve this? I tried re-installing jupyter-notebook using brew (I also updated brew) but I keep getting the same error.

Answer: I had the same problem after i upgraded to macOS Big Sur. I updated my python version (3.6.4) to (3.9.0) and after that i just uninstalled the notebook and reinstalled it. Now it works.
Question: I am trying to get all the PC cores to work simultaneously while filling a PostgreSQL database, I have edited the code to make a reproducible error of what I am getting Traceback (most recent call last): File "test2.py", line 50, in <module> download_all_sites(sites) File "test2.py", line 36, in download_all_sites pool.map(download_site, sites) File "/usr/lib/python3.8/multiprocessing/pool.py", line 364, in map return self._map_async(func, iterable, mapstar, chunksize).get() File "/usr/lib/python3.8/multiprocessing/pool.py", line 771, in get raise self._value psycopg2.OperationalError: SSL error: decryption failed or bad record mac  The full code which makes the error import requests import multiprocessing import time import os import psycopg2 session = None conn = psycopg2.connect(user="user", password="pass123", host="127.0.0.1", port="5432", database="my_db") cursor = conn.cursor() def set_global_session(): global session if not session: session = requests.Session() def download_site(domain): url = "http://" + domain with session.get(url) as response: temp = response.text.lower() found = [i for i in keywords if i in temp] query = """INSERT INTO test (domain, keyword) VALUES (%s, %s)""" cursor.execute(query, (domain, found)) def download_all_sites(sites): with multiprocessing.Pool(processes=os.cpu_count(), initializer=set_global_session) as pool: pool.map(download_site, sites) if __name__ == "__main__": sites = ['google.com'] * 10 keywords = ['google', 'success'] start_time = time.time() download_all_sites(sites) duration = time.time() - start_time conn.commit() print(f"Finished {len(sites)} in {duration} seconds") 

Answer: Create a new postgres connection for each multiprocess. Libpq connections shouldn’t be used with forked processes (what multiprocessing is doing), it is mentioned in the second warning box at the postgres docs. import requests import multiprocessing import time import os import psycopg2 session = None def set_global_session(): global session if not session: session = requests.Session() def download_site(domain): url = "http://" + domain with session.get(url) as response: #temp = response.text.lower() #found = [i for i in keywords if i in temp] #query = """INSERT INTO test (domain, keyword) VALUES (%s, %s)""" conn = psycopg2.connect( "dbname=mf port=5959 host=localhost user=mf_usr" ) cursor = conn.cursor() query = """INSERT INTO mytable (name) VALUES (%s)""" cursor.execute(query, (domain, )) conn.commit() conn.close() def download_all_sites(sites): with multiprocessing.Pool( processes=os.cpu_count(), initializer=set_global_session ) as pool: pool.map(download_site, sites) if __name__ == "__main__": sites = ['google.com'] * 10 keywords = ['google', 'success'] start_time = time.time() download_all_sites(sites) duration = time.time() - start_time print(f"Finished {len(sites)} in {duration} seconds") # make sure it worked! conn = psycopg2.connect("dbname=mf port=5959 host=localhost user=mf_usr") cursor = conn.cursor() cursor.execute('select count(name) from mytable') print(cursor.fetchall()) # verify 10 downloads == 10 records in database  Out: Finished 10 in 0.9922008514404297 seconds [(10,)]
Question: I was wondering if it could be possible to control the power of usb ports in Python, using vendor ids and product ids. It should be controlling powers instead of just enabling and disabling the ports. It would be appreciated if you could provide some examples.

Answer: Look into the subprocess module in the standard library: What commands you need will depend on the OS. Windows For windows you will want to look into devcon This has been answered in previous posts import subprocess # Fetches the list of all usb devices: result = subprocess.run(['devcon', 'hwids', '=usb'], capture_output=True, text=True) # ... add code to parse the result and get the hwid of the device you want ... subprocess.run(['devcon', 'disable', parsed_hwid]) # to disable subprocess.run(['devcon', 'enable', parsed_hwid]) # to enable  Linux See posts on shell comands import subprocess # determine desired usb device # to disable subprocess.run(['echo', '0', '>' '/sys/bus/usb/devices/usbX/power/autosuspend_delay_ms']) subprocess.run(['echo', 'auto', '>' '/sys/bus/usb/devices/usbX/power/control']) # to enable subprocess.run(['echo', 'on', '>' '/sys/bus/usb/devices/usbX/power/control'])
Question: The problem that I want to solve related the project setup:  Good names of directories so that their purpose is clear. Keeping all project files (including virtualenv) in one place, so I can easily copy, move, archive, remove the whole project, or estimate disk space usage. Creating multiple copies of some selected file sets such as entire application, repository, or virtualenv, while keeping a single copy of other files that I don't want to clone. Deploying the right set of files to the server simply by resyncing selected one dir. handling both frontend and backend nicely. 

Answer: Harsha already mentioned my project generator but I think it can be helpful for future readers to explain the ideas behind of it. If you are going to serve your frontend something like yarn or npm. You should not worry about the structure between them. With something like axios or the Javascript's fetch you can easily talk with your backend from anywhere. When it comes to structuring the backend, if you want to render templates with Jinja, you can have something that is close to MVC Pattern. your_project ├── __init__.py ├── main.py ├── core │ ├── models │ │ ├── database.py │ │ └── __init__.py │ ├── schemas │ │ ├── __init__.py │ │ └── schema.py │ └── settings.py ├── tests │ ├── __init__.py │ └── v1 │ ├── __init__.py │ └── test_v1.py └── v1 ├── api.py ├── endpoints │ ├── endpoint.py │ └── __init__.py └── __init__.py  By using __init__ everywhere, we can access the variables from the all over the app, just like Django. Let's the folders into parts:  Core  models  database.py   schemas  users.py something.py   settings.py views (Add this if you are going to render templates)  v1_views.py v2_views.py     tests v1 v2  Models It is for your database models, by doing this you can import the same database session or object from v1 and v2. Schemas Schemas are your Pydantic models, we call it schemas because it is actually used for creating OpenAPI schemas since FastAPI is based on OpenAPI specification we use schemas everywhere, from Swagger generation to endpoint's expected request body. settings.py It is for Pydantic's Settings Management which is extremely useful, you can use the same variables without redeclaring it, to see how it could be useful for you check out our documentation for Settings and Environment Variables Views This is optional if you are going to render your frontend with Jinja, you can have something close to MVC pattern  Core  views  v1_views.py v2_views.py      It would look something like this if you want to add views. Tests It is good to have your tests inside your backend folder. APIs Create them independently by APIRouter, instead of gathering all your APIs inside one file. Notes You can use absolute import for all your importing since we are using __init__ everywhere, see Python's packaging docs. So assume you are trying to import v1's endpoint.py from v2, you can simply do from my_project.v1.endpoints.endpoint import something
Question: I would like to list all of my local profiles using boto3, as I think boto3 is not picking up my credentials correctly. I have tried the following: import boto3 boto3.Session.available_profiles  Which doesn't give me a list, but a property object.

Answer: You might want to use awscli instead of boto3 to list your profiles. aws configure list  This should output something like this:  Name Value Type Location ---- ----- ---- -------- profile <not set> None None access_key ****************ABCD config_file ~/.aws/config secret_key ****************ABCD config_file ~/.aws/config region us-west-2 env AWS_DEFAULT_REGION  As for boto3, try this: for profile in boto3.session.Session().available_profiles: print(profile)
Question: I try to restrict the 'parameter' type to be int or list like the function 'f' below. However, Pycharm does not show a warning at the line f("weewfwef") about wrong parameter type, which means this (parameter : [int, list]) is not correct. In Python, is it possible to restrict the type of a python function parameter to two possible types? def f(parameter : [int, list]): if len(str(parameter)) <= 3: return 3 else: return [1] if __name__ == '__main__': f("weewfwef") 

Answer: The term you're looking for is a union type. from typing import Union def f(parameter: Union[int, list]): ...  Union is not limited to two types. If you ever have a value which is one of several known types, but you can't necessarily know which one, you can use Union[...] to encapsulate that information.
Question: I would like to perform a rolling average but with a window that only has a finite 'vision' in x. I would like something similar to what I have below, but I want a window range that based on the x value rather than position index. While doing this within pandas is preferred numpy/scipy equivalents are also OK import numpy as np import pandas as pd x_val = [1,2,4,8,16,32,64,128,256,512] y_val = [x+np.random.random()*200 for x in x_val] df = pd.DataFrame(data={'x':x_val,'y':y_val}) df.set_index('x', inplace=True) df.plot() df.rolling(1, win_type='gaussian').mean(std=2).plot()  So I would expect the first 5 values to be averaged together because they are within 10 xunits of each other, but the last values to be unchanged.

Answer: According to pandas documentation on rolling  Size of the moving window. This is the number of observations used for calculating the statistic. Each window will be a fixed size.  Therefore, maybe you need to fake a rolling operation with various window sizes like this test_df = pd.DataFrame({'x':np.linspace(1,10,10),'y':np.linspace(1,10,10)}) test_df['win_locs'] = np.linspace(1,10,10).astype('object') for ind in range(10): test_df.at[ind,'win_locs'] = np.random.randint(0,10,np.random.randint(5)).tolist() # rolling operation with various window sizes def worker(idx_list): x_slice = test_df.loc[idx_list,'x'] return np.sum(x_slice) test_df['rolling'] = test_df['win_locs'].apply(worker)  As you can see, test_df is  x y win_locs rolling 0 1.0 1.0 [5, 2] 9.0 1 2.0 2.0 [4, 8, 7, 1] 24.0 2 3.0 3.0 [] 0.0 3 4.0 4.0 [9] 10.0 4 5.0 5.0 [6, 2, 9] 20.0 5 6.0 6.0 [] 0.0 6 7.0 7.0 [5, 7, 9] 24.0 7 8.0 8.0 [] 0.0 8 9.0 9.0 [] 0.0 9 10.0 10.0 [9, 4, 7, 1] 25.0  where the rolling operation is achieved with apply method. However, this approach is significantly slower than the native rolling, for example, test_df = pd.DataFrame({'x':np.linspace(1,10,10),'y':np.linspace(1,10,10)}) test_df['win_locs'] = np.linspace(1,10,10).astype('object') for ind in range(10): test_df.at[ind,'win_locs'] = np.arange(ind-1,ind+1).tolist() if ind >= 1 else []  using the approach above %%timeit # rolling operation with various window sizes def worker(idx_list): x_slice = test_df.loc[idx_list,'x'] return np.sum(x_slice) test_df['rolling_apply'] = test_df['win_locs'].apply(worker)  the result is 41.4 ms ± 4.44 ms per loop (mean ± std. dev. of 7 runs, 100 loops each)  while using native rolling is ~x50 faster %%timeit test_df['rolling_native'] = test_df['x'].rolling(window=2).sum() 863 µs ± 118 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)