eshangj/stackoverflow_q_and_a_sample huggingface dataset:
Question: We use YAML configuration for services scaling. Usually it goes like this: service: scalingPolicy: capacity: min: 1 max: 1  So it's easy to open with basic PyYAML and parse as an dict to get config['service']['scalingPolicy']['capacity']['min'] result as 1. Problem is that some configs are built with dots delimiter e.g: service.scalingPolicy.capacity: min: 1 max: 1  Basic consumer of this configs is Java's Spring and somehow it's treated equally as the example above. But due to need to also parse these configs with Python - I get whole dot separated line as a config['service.scalingPolicy.capacity'] key. The question is - how would I make python parse any kind of keys combinations (both separated by dots and separated by tabulation and :). I didn't find related parameters for Python YAML libs (I've checked standard PyYAML and ruamel.yaml) and handling any possible combination manually seems like a crazy idea. The only possible idea I have is to write my own parser but maybe there is something I'm missing so I won't have to reinvent the bicycle.

Answer: This is not trivial, it is much more easy to split a lookup with a key with dots into recursing into a nested data structure. Here you have a nested data structure and different [key] lookups mean different things at different levels. If you use ruamel.yaml in the default round-trip mode, you can add a class-variable to the type that represents a mapping, that defines on what the keys were split and an instance variable that keeps track of the prefix already matched: import sys import ruamel.yaml from ruamel.yaml.compat import ordereddict from ruamel.yaml.comments import merge_attrib yaml_str = """\ service.scalingPolicy.capacity: min: 1 max: 1 """ def mapgetitem(self, key): sep = getattr(ruamel.yaml.comments.CommentedMap, 'sep') if sep is not None: if not hasattr(self, 'splitprefix'): self.splitprefix = '' if self.splitprefix: self.splitprefix += sep + key else: self.splitprefix = key if self.splitprefix not in self: for k in self.keys(): if k.startswith(self.splitprefix): break else: raise KeyError(self.splitprefix) return self key = self.splitprefix delattr(self, 'splitprefix') # to make the next lookup work from start try: return ordereddict.__getitem__(self, key) except KeyError: for merged in getattr(self, merge_attrib, []): if key in merged[1]: return merged[1][key] raise old_mapgetitem = ruamel.yaml.comments.CommentedMap.__getitem__ # save the original __getitem__ ruamel.yaml.comments.CommentedMap.__getitem__ = mapgetitem ruamel.yaml.comments.CommentedMap.sep = '.' yaml = ruamel.yaml.YAML() # yaml.indent(mapping=4, sequence=4, offset=2) # yaml.preserve_quotes = True config = yaml.load(yaml_str) print('min:', config['service']['scalingPolicy']['capacity']['min']) print('max:', config['service']['scalingPolicy']['capacity']['max']) print('---------') config['service']['scalingPolicy']['capacity']['max'] = 42 # and dump with the original routine, as it uses __getitem__ ruamel.yaml.comments.CommentedMap.__getitem__ = old_mapgetitem yaml.dump(config, sys.stdout)  which gives: min: 1 max: 1 --------- service.scalingPolicy.capacity: min: 1 max: 42
Question: I can get a correct token when calling the URL /token/ but I wish to create a token manually for the user when /login/ is called. urls.py:  from django.urls import path from . import views from .views import MyTokenObtainPairView from rest_framework_simplejwt.views import ( TokenRefreshView, TokenVerifyView ) urlpatterns = [ path('', views.api_root), path('register/', views.register), path('login/', views.login), path('token/', MyTokenObtainPairView.as_view(), name='token_obtain_pair'), path('token/refresh/', TokenRefreshView.as_view(), name='token_refresh'), path('token/verify/', TokenVerifyView.as_view(), name='token_verify'), ]  views.py: @api_view(['POST']) def login(request): email = request.data.get('email') password = request.data.get('password') user = get_object_or_404(User, email=email) if make_password(password) == user.password: if not user.is_active: return Response({"error": "User is not active"}, status=400) tokens = MyTokenObtainPairView.get_token(user) parse_token = { 'refresh': str(tokens), 'access': str(tokens.access_token), } return Response(status=200, data=parse_token) else: return Response(status=401) class MyTokenObtainPairSerializer(TokenObtainPairSerializer): def validate(self, attrs): data = super().validate(attrs) refresh = self.get_token(self.user) data['refresh'] = str(refresh) data['access'] = str(refresh.access_token) # Add extra responses here data['username'] = self.user.username data['groups'] = self.user.groups.values_list('name', flat=True) data['test'] = '1234' return data class MyTokenObtainPairView(TokenObtainPairView): serializer_class = MyTokenObtainPairSerializer  How do I modify this line to get my token for a specific user? tokens = MyTokenObtainPairView.get_token(user) I have read the doc about manually create token by importing this: from rest_framework_simplejwt.tokens import RefreshToken but it is not adding the payload into the token...

Answer: Actually after Googling for an hour I finally got a solution from another post... Proper way to do this:  refresh = RefreshToken.for_user(user) refresh['user_name'] = user.username refresh['first_name'] = user.first_name refresh['last_name'] = user.last_name refresh['full_name'] = user.get_full_name() return { 'refresh': str(refresh), 'access': str(refresh.access_token), }  In case someone would need this...
Question: I'm using mutex for blocking part of code in the first function. Can I unlock mutex in the second function? For example: import threading mutex = threading.Lock() def function1(): mutex.acquire() #do something def function2(): #do something mutex.release() #do something 

Answer: You certainly can do what you're asking, locking the mutex in one function and unlocking it in another one. But you probably shouldn't. It's bad design. If the code that uses those functions calls them in the wrong order, the mutex may be locked and never unlocked, or be unlocked when it isn't locked (or even worse, when it's locked by a different thread). If you can only ever call the functions in exactly one order, why are they even separate functions? A better idea may be to move the lock-handling code out of the functions and make the caller responsible for locking and unlocking. Then you can use a with statement that ensures the lock and unlock are exactly paired up, even in the face of exceptions or other unexpected behavior. with mutex: function1() function2()  Or if not all parts of the two functions are "hot" and need the lock held to ensure they run correctly, you might consider factoring out the parts that need the lock into a third function that runs in between the other two: function1_cold_parts() with mutex: hot_parts() function2_cold_parts()
Question: I am trying to display multiple images (as labels) to the window but only the last images is displayed from tkinter import * from PIL import ImageTk, Image root = Tk() f = open("data/itemIDsList.txt") ids = [] for line in f: line = line.rstrip("\n") ids.append(line) f.close() for i in range(10): img = ImageTk.PhotoImage(Image.open(f"website/images/{ids[i]}.png")) Label(root, image=img, width=60, height=80).grid() root.mainloop() 

Answer: Each time you reassign img in the loop, the data of the previous image gets destroyed and can no longer be displayed. To fix this, add the images to a list to store them permanently: from tkinter import * from PIL import ImageTk, Image root = Tk() f = open("data/itemIDsList.txt") ids = [] for line in f: line = line.rstrip("\n") ids.append(line) f.close() imgs = [] for i in range(10): imgs.append(ImageTk.PhotoImage(Image.open(f"website/images/{ids[i]}.png"))) Label(root, image=imgs[-1], width=60, height=80).grid() root.mainloop()
Question: Suppose I have the following two DFs: DF A: First column is a date, and then there are columns that start with a year (2021, 2022...) Date 2021.Water 2021.Gas 2022.Electricity may-04 500 470 473 may-05 520 490 493 may-06 540 510 513  DF B: First column is a date, and then there are columns that start with a year (2021, 2022...) Date 2021.Amount 2022.Amount may-04 100 95 may-05 110 105 may-06 120 115  The expected result is a DF with the columns from DF A, but that have the rows divided by the values for the matching year in DF B. Such as: Date 2021.Water 2021.Gas 2022.Electricity may-04 5.0 4.7 5.0 may-05 4.7 4.5 4.7 may-06 4.5 4.3 4.5  I am really struggling with this problem. Let me know if any clarifications are needed and will be glad to help.

Answer: Try this: dfai = dfa.set_index('Date') dfai.columns = dfai.columns.str.split('.', expand=True) dfbi = dfb.set_index('Date').rename(columns = lambda x: x.split('.')[0]) df_out = dfai.div(dfbi, level=0).round(1) df_out.columns = df_out.columns.map('.'.join) df_out.reset_index()  Output:  Date 2021.Water 2021.Gas 2022.Electricity 0 may-04 5.0 4.7 5.0 1 may-05 4.7 4.5 4.7 2 may-06 4.5 4.2 4.5  Details First, move 'Date' into the index of both dataframes, then use string split to get years into a level in each dataframe. Use, pd.DataFrame.div with level=0 to align operations on the top level index of each dataframe. Flatten multiindex column header back to a single level and reset_index.
Question: When I was reading the code online, I have encountered the following cases of using abstract classes: from abc import abstractmethod,ABCMeta class Generator(object,metaclass=ABCMeta): @abstractmethod def generate(self): raise NotImplementedError("method not implemented") generator=Generator() generator.generate()  The following error is returned, as expected: TypeError: Can't instantiate abstract class Generator with abstract methods generate But if I write it like this (the only difference is in the second line) from abc import abstractmethod,ABCMeta class Generator(object): @abstractmethod def generate(self): raise NotImplementedError("method not implemented") generator=Generator() generator.generate()  Although there are changes in the error message, NotImplementedError: method not implemented When I implemented the generate method, both of the above ways of Generator were executed correctly, class GeneticAlgorithm(Generator): def generate(self): print("ABC") ga=GeneticAlgorithm() ga.generate()  >>> ABC  So why do we need the statement metaclass=ABCMeta? I know something from GeeksforGeeks that  ABCMeta metaclass provides a method called register method that can be invoked by its instance. By using this register method, any abstract base class can become an ancestor of any arbitrary concrete class.  But this still doesn't make me understand the necessity of declaring metaclass=ABCMeta, it feels like @abstractmethod modifying the method is enough.

Answer: You "need" the metaclass=ABCMeta to enforce the rules at instantiation time. generator=Generator() # Errors immediately when using ABCMeta generator.generate() # Only errors if and when you call generate otherwise  Imagine if the class had several abstract methods, only some of which were implemented in a child. It might work for quite a while, and only error when you got around to calling an unimplemented method. Failing eagerly before you rely on the ABC is generally a good thing, in the same way it's usually better for a function to raise an exception rather than just returning None to indicate failure; you want to know as soon as things are wrong, not get a weird error later without knowing the ultimate cause of the error. Side-note: There's a much more succinct way to be an ABC than explicitly using the metaclass=ABCMeta syntax: from abc import abstractmethod, ABC class Generator(ABC):  Python almost always makes empty base classes that use the metaclass to simplify use (especially during the 2 to 3 transition period, where there was no compatible metaclass syntax that worked in both, and direct inheritance was the only thing that worked).
Question: I often find myself having to start a debugging session in PyCharm only in order to inspect a variable and look up its class with something.__class__ so that I can insert the type hint into the code in order to make it more readable. Is there a way to do it automatically in PyCharm via a context action, in VSCode or maybe some other tool?

Answer: Have you tried Adding type hints in the PyCharm?  The VSCode-Python has not supported this feature, and I have submitted a feature request on GitHub.
Question: How can I create a venn diagram in python from 4 sets? Seems like the limit in matplotlib is only 3? from matplotlib_venn import venn3 v = venn3( [ set(ether_list), set(bitcoin_list), set(doge_list), ], ) 

Answer: Venn diagrams with circles can work only with <4 sets, because the geometrical properties of intersections (some won't be possible to show). Some python libraries that allow you to show venn diagrams with more exotic shapes are:  pyvenn venn
Question: I want to compile my DQN Agent but I get error: AttributeError: 'Adam' object has no attribute '_name', DQN = buildAgent(model, actions) DQN.compile(Adam(lr=1e-3), metrics=['mae'])  I tried adding fake _name but it doesn't work, I'm following a tutorial and it works on tutor's machine, it's probably some new update change but how to fix this Here is my full code: from keras.layers import Dense, Flatten import gym from keras.optimizer_v1 import Adam from rl.agents.dqn import DQNAgent from rl.policy import BoltzmannQPolicy from rl.memory import SequentialMemory env = gym.make('CartPole-v0') states = env.observation_space.shape[0] actions = env.action_space.n episodes = 10 def buildModel(statez, actiones): model = Sequential() model.add(Flatten(input_shape=(1, statez))) model.add(Dense(24, activation='relu')) model.add(Dense(24, activation='relu')) model.add(Dense(actiones, activation='linear')) return model model = buildModel(states, actions) def buildAgent(modell, actionz): policy = BoltzmannQPolicy() memory = SequentialMemory(limit=50000, window_length=1) dqn = DQNAgent(model=modell, memory=memory, policy=policy, nb_actions=actionz, nb_steps_warmup=10, target_model_update=1e-2) return dqn DQN = buildAgent(model, actions) DQN.compile(Adam(lr=1e-3), metrics=['mae']) DQN.fit(env, nb_steps=50000, visualize=False, verbose=1) 

Answer: Your error came from importing Adam with from keras.optimizer_v1 import Adam, You can solve your problem with tf.keras.optimizers.Adam from TensorFlow >= v2 like below: (The lr argument is deprecated, it's better to use learning_rate instead.) # !pip install keras-rl2 import tensorflow as tf from keras.layers import Dense, Flatten import gym from rl.agents.dqn import DQNAgent from rl.policy import BoltzmannQPolicy from rl.memory import SequentialMemory env = gym.make('CartPole-v0') states = env.observation_space.shape[0] actions = env.action_space.n episodes = 10 def buildModel(statez, actiones): model = tf.keras.Sequential() model.add(Flatten(input_shape=(1, statez))) model.add(Dense(24, activation='relu')) model.add(Dense(24, activation='relu')) model.add(Dense(actiones, activation='linear')) return model def buildAgent(modell, actionz): policy = BoltzmannQPolicy() memory = SequentialMemory(limit=50000, window_length=1) dqn = DQNAgent(model=modell, memory=memory, policy=policy, nb_actions=actionz, nb_steps_warmup=10, target_model_update=1e-2) return dqn model = buildModel(states, actions) DQN = buildAgent(model, actions) DQN.compile(tf.keras.optimizers.Adam(learning_rate=1e-3), metrics=['mae']) DQN.fit(env, nb_steps=50000, visualize=False, verbose=1)
Question: So i am making a big social media app but i have a problem which framework to choose flask or express.js i like flask so much but it cant handle too much requests. Express.js can handle about 25k request per second (google). So is there anyway to make flask handle 25k request per second using gunicorn currently i am using this command $ gunicorn -w 4 -b 0.0.0.0:5000 your_project:app but it can only handle 4 request at a time. And uh one more question can flask handle 1Million user at a time. Should i choose express.js because it can handle 25k request

Answer: You can use multithreads or gevent to increase gunicorn's concurrency. Option1 multithreads eg: gunicorn -w 4 --threads 100 -b 0.0.0.0:5000 your_project:app  --threads 100 means 100 threads per process. -w 4 means 4 processes, so -w 4 --threads 100 means 400 requests at a time Option2 gevent worker eg: pip install gevent gunicorn -w 4 -k gevent --worker-connections 1000 -b 0.0.0.0:5000 your_project:app  -k gevent --worker-connections 1000 means 1000 coroutines per gevent worker process. -w 4 means 4 processes, so -w 4 -k gevent --worker-connections 1000 means 4000 requests at a time. For more information, you can refer to my blog post: https://easydevguide.com/posts/gunicorn_concurrency
Question: I created a scheduled task and my cypress script is being run once an hour. But after that I want to execute a python script and pass the result data there. Run the script and get the "results" as failed or success. $ cypress run --spec "cypress/integration/myproject/myscript.js"  And pass the "results" data to a python script. $ python test.py results  How can I do this?

Answer: There is a subprocess module which is able to run external commands, here is the example: import subprocess def get_test_output(): filepath = './cypress/integration/myproject/myscript.js' res = subprocess.run( ['echo', filepath], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) # In your case it will be: # res = subprocess.run( # ['cypress', 'run', '--spec', filepath], # stdout=subprocess.PIPE, # stderr=subprocess.STDOUT, # ) return res.stdout.decode() if __name__ == '__main__': test_res = get_test_output() print(test_res) # => ./cypress/integration/myproject/myscript.js  You could run cypress in the begining of test.py and pass results further to needed functions
Question: I'm looking for an efficient Python function that randomly allocates an integer across k bins. That is, some function allocate(n, k) will produce a k-sized array of integers summing to n. For example, allocate(4, 3) could produce [4, 0, 0], [0, 2, 2], [1, 2, 1], etc. It should be randomly distributed per item, assigning each of the n items randomly to each of the k bins.

Answer: Adapting Michael Szczesny's comment based on numpy's new paradigm: def allocate(n, k): return np.random.default_rng().multinomial(n, [1 / k] * k)  This notebook verifies that it returns the same distribution as my brute-force approach.
Question: I am trying to download YouTube playlist from url "https://www.youtube.com/watch?v=uyVYfSNb_Pc&list=PLBxwSeQlMDNiNt72UmSvKBLsxPgGY_Jy-", but getting the error 'get_throttling_function_name: could not find match for multiple'. Code block is: ` from pytube import Playlist play_list = Playlist('https://www.youtube.com/watch?v=uyVYfSNb_Pc&list=PLBxwSeQlMDNiNt72UmSvKBLsxPgGY_Jy-') print(f'Downloading: {play_list.title}') for video in play_list.videos: print(video.title) st = video.streams.get_highest_resolution() st.download(r'path') `  i am using the latest version of pytube.

Answer: Becuase youtube changed something on its end, and now you have to change pytube's ciper.py's function_patterns to the following r'a\.[a-zA-Z]\s*&&\s*\([a-z]\s*=\s*a\.get\("n"\)\)\s*&&\s*' r'\([a-z]\s*=\s*([a-zA-Z0-9$]{2,3})(\[\d+\])?\([a-z]\)'  And you also have to change line 288 to this: nfunc=re.escape(function_match.group(1))),  You'll have to use this workaround until pytube officially releases a fix.
Question: I want the ability to use custom functions in pandas groupby agg(). I Know there is the option of using apply but doing several aggregations is what I want. Below is my test code that I tried to get working for the weighted average. Python Code import pandas as pd import numpy as np def weighted_avg(df, values, weights): '''To calculate a weighted average in Pandas. Demo see https://www.statology.org/pandas-weighted-average/ Example: df.groupby('Group Names').apply(w_avg, 'Results', 'AFY')''' v = df[values] w = df[weights] return (v * w).sum() / w.sum() # below creates a dataframe. dfr = pd.DataFrame(np.random.randint(1,50,size=(4,4)), columns=list('ABCD')) dfr['group'] = [1, 1, 0, 1] print(dfr) dfr = dfr.groupby('group').agg({'A':'mean', 'B':'sum', 'C': lambda x: weighted_avg(dfr, 'D', 'C')}).reset_index() print(dfr)  Results - Output  A B C D group 0 5 2 17 38 1 1 35 30 22 32 1 2 15 18 16 11 0 3 46 6 20 34 1 group A B C 0 0 15.000000 18 29.413333 1 1 28.666667 38 29.413333  The problem: The weighted average is returning the value for the whole table and not the 'group' column. How can I get the weighted average by group working? I did try placing the groupby inside the function like shown here but no success. Thank you for taking a look.

Answer: You can use x you have in lambda (specifically, use it's .index to get values you want). For example: import pandas as pd import numpy as np def weighted_avg(group_df, whole_df, values, weights): v = whole_df.loc[group_df.index, values] w = whole_df.loc[group_df.index, weights] return (v * w).sum() / w.sum() dfr = pd.DataFrame(np.random.randint(1, 50, size=(4, 4)), columns=list("ABCD")) dfr["group"] = [1, 1, 0, 1] print(dfr) dfr = ( dfr.groupby("group") .agg( {"A": "mean", "B": "sum", "C": lambda x: weighted_avg(x, dfr, "D", "C")} ) .reset_index() ) print(dfr)  Prints:  A B C D group 0 32 2 34 29 1 1 33 32 15 49 1 2 4 43 41 10 0 3 39 33 7 31 1 group A B C 0 0 4.000000 43 10.000000 1 1 34.666667 67 34.607143   EDIT: As @enke stated in comments, you can call your weighted_avg function with already filtered dataframe: weighted_avg(dfr.loc[x.index], 'D', 'C')
Question: I am planning to split the questions from this PDF document. The challenge is that the questions are not orderly spaced. For example the first question occupies an entire page, second also the same while the third and fourth together make up one page. If I have to manually slice it, it will be ages. So, I thought to split it up into images and work on them. Is there a possibility to take image as this  and split into individual components like this?  

Answer: We may solve it using (mostly) morphological operations:  Read the input image as grayscale. Apply thresholding with inversion. Automatic thresholding using cv2.THRESH_OTSU is working well. Apply opening morphological operation for removing small artifacts (using the kernel np.ones(1, 3)) Dilate horizontally with very long horizontal kernel - make horizontal lines out of the text lines. Apply closing vertically - create two large clusters. The size of the vertical kernel should be tuned according to the typical gap. Finding connected components with statistics. Iterate the connected components and crop the relevant area in the vertical direction.   Complete code sample: import cv2 import numpy as np img = cv2.imread('scanned_image.png', cv2.IMREAD_GRAYSCALE) # Read image as grayscale thesh = cv2.threshold(img, 0, 255, cv2.THRESH_OTSU + cv2.THRESH_BINARY_INV)[1] # Apply automatic thresholding with inversion. thesh = cv2.morphologyEx(thesh, cv2.MORPH_OPEN, np.ones((1, 3), np.uint8)) # Apply opening morphological operation for removing small artifacts. thesh = cv2.dilate(thesh, np.ones((1, img.shape[1]), np.uint8)) # Dilate horizontally - make horizontally lines out of the text. thesh = cv2.morphologyEx(thesh, cv2.MORPH_CLOSE, np.ones((50, 1), np.uint8)) # Apply closing vertically - create two large clusters nlabel, labels, stats, centroids = cv2.connectedComponentsWithStats(thesh, 4) # Finding connected components with statistics parts_list = [] # Iterate connected components: for i in range(1, nlabel): top = int(stats[i, cv2.CC_STAT_TOP]) # Get most top y coordinate of the connected component height = int(stats[i, cv2.CC_STAT_HEIGHT]) # Get the height of the connected component roi = img[top-5:top+height+5, :] # Crop the relevant part of the image (add 5 extra rows from top and bottom). parts_list.append(roi.copy()) # Add the cropped area to a list cv2.imwrite(f'part{i}.png', roi) # Save the image part for testing cv2.imshow(f'part{i}', roi) # Show part for testing # Show image and thesh testing cv2.imshow('img', img) cv2.imshow('thesh', thesh) cv2.waitKey() cv2.destroyAllWindows()   Results: Stage 1:  Stage 2:  Stage 3:  Stage 4:  Top area:  Bottom area: