I have lots of multiprocessing processes which have to add to and search in a dict. Deletion of values is not needed.
Atm I am using multiprocessing.Manager() and
`dict = manager.dict()`
This works pretty well, but I think that the manager is a huge bottleneck here. Any ideas? It has to run on older Python 3 versions, otherwise I would use this cool thing I found: https://github.com/ronny-rentner/UltraDict
I have two projects, one is working and the other isn't. I want to make the first project include the same functionality from the second project, or the second project to work as the first one does.
I've thought of adding tests to the first project that I can use with the second project so that they end up working the same way. But I don't know which kind of tests to write, should I be using a bunch of videos as test subjects? I think that would make testing very slow, but I don't know what else to do.
https://codeberg.org/LifeSymbiont/video-diet
https://codeberg.org/LifeSymbiont/reduce_video_size/
I'm trying to print a progress bar while executing ffmpeg but I'm having trouble getting the total number of frames and the current frame being processed. The error I get is `AttributeError: 'NoneType' object has no attribute 'groups'`. I've copied the function from [this program][1] and for some reason it works there but not here, even though I haven't changed that part.
[main.py][2]
```
pattern_duration = re.compile(
'duration[ \t\r]?:[ \t\r]?(.+?),[ \t\r]?start', re.IGNORECASE)
pattern_progress = re.compile('time=(.+?)[ \t\r]?bitrate', re.IGNORECASE)
def execute_ffmpeg(self, manager, cmd):
proc = expect.spawn(cmd, encoding='utf-8')
self.update_progress_bar(proc, manager)
self.raise_ffmpeg_error(proc)
def update_progress_bar(self, proc, manager):
total = self.get_total_frames(proc)
cont = 0
pbar = self.initialize_progress_bar(manager)
try:
proc.expect(pattern_duration)
while True:
progress = self.get_current_frame(proc)
percent = progress / total * 100
pbar.update(percent - cont)
cont = percent
except expect.EOF:
pass
finally:
if pbar is not None:
pbar.close()
def raise_ffmpeg_error(self, proc):
proc.expect(expect.EOF)
res = proc.before
res += proc.read()
exitstatus = proc.wait()
if exitstatus:
raise ffmpeg.Error('ffmpeg', '', res)
def initialize_progress_bar(self, manager):
pbar = None
pbar = manager.counter(
total=100,
desc=self.path.rsplit(os.path.sep, 1)[-1],
unit='%',
bar_format=BAR_FMT,
counter_format=COUNTER_FMT,
leave=False
)
return pbar
def get_total_frames(self, proc):
return sum(map(lambda x: float(
x[1])*60**x[0], enumerate(reversed(proc.match.groups()[0].strip().split(':')))))
def get_current_frame(self, proc):
proc.expect(pattern_progress)
return sum(map(lambda x: float(
x[1])*60**x[0], enumerate(reversed(proc.match.groups()[0].strip().split(':')))))
```
[1]: https://github.com/JavierOramas/video-diet/blob/966b5192902e55bd60ae06a5df195ec41bcd5d71/diet_video/__init__.py
[2]: https://codeberg.org/LifeSymbiont/reduce_video_size/src/commit/ef3cd0974ecd1c4d0a17b6394499650c9fc3da2b/main.py
This function is giving me an error but I've copied it from [the video-diet program](https://github.com/JavierOramas/video-diet/blob/81a5df4ad27e8cd6fff1be4974067631343a4354/diet_video/__init__.py#L42) and I don't really understand it so I can't simplify it. Could someone who understands it, explain it step by step?
```py
def convert_video_progress_bar(self, manager, cmd):
name = self.path.rsplit(os.path.sep, 1)[-1]
proc = expect.spawn(cmd, encoding='utf-8')
pbar = None
try:
proc.expect(pattern_duration)
total = sum(map(lambda x: float(
x[1])*60**x[0], enumerate(reversed(proc.match.groups()[0].strip().split(':')))))
cont = 0
pbar = manager.counter(
total=100,
desc=name,
unit='%',
bar_format=BAR_FMT,
counter_format=COUNTER_FMT,
leave=False
)
while True:
proc.expect(pattern_progress)
progress = sum(map(lambda x: float(
x[1])*60**x[0], enumerate(reversed(proc.match.groups()[0].strip().split(':')))))
percent = progress/total*100
pbar.update(percent-cont)
cont = percent
except expect.EOF:
traceback.print_exc()
finally:
if pbar is not None:
pbar.close()
proc.expect(expect.EOF)
res = proc.before
res += proc.read()
exitstatus = proc.wait()
if exitstatus:
raise ffmpeg.Error('ffmpeg', '', res)
```
I want to do the following but I don't know how to get the name and percentage:
`mymodule_test.py`
```
import unittest
from unittest import TestCase
from mymodule import get_wanted_cards, get_percentage
class GetCardsTestCase(TestCase):
def test_get_wanted_cards(self):
s = '''
↑ A Little Chat 92 (1%) 0 (0%) NEW
↑ An Offer You Can't Refuse 88 (87%) 0 (0%) NEW
↑ Angelic Observer 92 (91%) 0 (0%) NEW
'''
expected = ["4 An Offer You Can't Refuse", "4 Angelic Observer"]
actual = get_wanted_cards(s)
self.assertEqual(actual, expected)
def test_get_percentage(self):
s = "92 (1%)"
expected = 1
actual = get_percentage(s)
self.assertEqual(actual, expected)
if __name__ == '__main__':
unittest.main()
```
`mymodule.py`
```
import re
from typing import List
def get_wanted_cards(s: str) -> List[str]:
res = []
for line in s.splitlines():
array = line.split("\t")
if len(array) < 5:
continue
name = array[1]
percent = get_percentage(array[2])
if percent >= 50:
res += "4 " + name
return res
def get_percentage(s: str) -> int:
return int(re.match(r'\(([0-9]*)%\)', s).group(1))
if __name__ == "__main__":
pass
```