openwebrx-clone/csdr/chain/__init__.py

160 lines
4.6 KiB
Python
Raw Normal View History

2021-08-23 12:25:28 +00:00
from pycsdr.modules import Buffer
from typing import Union, Callable
2021-07-16 14:12:16 +00:00
class Chain:
2021-08-23 12:25:28 +00:00
def __init__(self, workers):
2021-07-24 16:50:30 +00:00
self.reader = None
2021-08-16 14:41:18 +00:00
self.writer = None
self.clientReader = None
2021-08-23 12:25:28 +00:00
self.workers = workers
2021-08-16 14:41:18 +00:00
for i in range(1, len(self.workers)):
self._connect(self.workers[i - 1], self.workers[i])
2021-08-27 14:11:03 +00:00
def empty(self):
return not self.workers
def _connect(self, w1, w2):
2021-07-24 16:50:30 +00:00
writer = Buffer(w1.getOutputFormat())
w1.setWriter(writer)
2021-08-16 14:41:18 +00:00
w2.setReader(writer.getReader())
2020-12-16 17:52:00 +00:00
2021-08-16 14:41:18 +00:00
def setReader(self, reader):
if self.reader is reader:
return
2021-08-16 14:41:18 +00:00
self.reader = reader
2021-07-19 17:04:14 +00:00
if self.workers:
2021-08-16 14:41:18 +00:00
self.workers[0].setReader(reader)
2021-08-16 14:41:18 +00:00
def setWriter(self, writer):
if self.writer is writer:
2021-07-24 16:50:30 +00:00
return
2021-08-16 14:41:18 +00:00
self.writer = writer
2021-07-24 16:50:30 +00:00
if self.workers:
2021-08-16 14:41:18 +00:00
self.workers[-1].setWriter(writer)
2021-08-23 12:25:28 +00:00
def indexOf(self, search: Union[Callable, object]) -> int:
def searchFn(x):
if callable(search):
return search(x)
else:
return x is search
2021-08-23 12:25:28 +00:00
try:
return next(i for i, v in enumerate(self.workers) if searchFn(v))
except StopIteration:
return -1
2021-07-16 14:12:16 +00:00
def replace(self, index, newWorker):
if index >= len(self.workers):
raise IndexError("Index {} does not exist".format(index))
self.workers[index].stop()
self.workers[index] = newWorker
error = None
if index == 0:
2021-08-16 14:41:18 +00:00
if self.reader is not None:
newWorker.setReader(self.reader)
else:
try:
previousWorker = self.workers[index - 1]
self._connect(previousWorker, newWorker)
except ValueError as e:
# store error for later raising, but still attempt the second connection
error = e
2021-08-16 14:41:18 +00:00
if index == len(self.workers) - 1:
if self.writer is not None:
newWorker.setWriter(self.writer)
else:
try:
nextWorker = self.workers[index + 1]
self._connect(newWorker, nextWorker)
except ValueError as e:
error = e
if error is not None:
2021-08-27 14:11:03 +00:00
raise error
2021-08-23 12:25:28 +00:00
def append(self, newWorker):
previousWorker = None
if self.workers:
previousWorker = self.workers[-1]
self.workers.append(newWorker)
if previousWorker:
self._connect(previousWorker, newWorker)
elif self.reader is not None:
newWorker.setReader(self.reader)
if self.writer is not None:
newWorker.setWriter(self.writer)
2021-08-27 14:11:03 +00:00
def insert(self, newWorker):
nextWorker = None
if self.workers:
nextWorker = self.workers[0]
self.workers.insert(0, newWorker)
if nextWorker:
self._connect(newWorker, nextWorker)
elif self.writer is not None:
newWorker.setWriter(self.writer)
if self.reader is not None:
newWorker.setReader(self.reader)
2021-08-23 12:25:28 +00:00
def remove(self, index):
removedWorker = self.workers[index]
self.workers.remove(removedWorker)
removedWorker.stop()
if index == 0:
if self.reader is not None:
self.workers[0].setReader(self.reader)
elif index == len(self.workers):
if self.writer is not None:
self.workers[-1].setWriter(self.writer)
else:
previousWorker = self.workers[index - 1]
nextWorker = self.workers[index]
self._connect(previousWorker, nextWorker)
def stop(self):
for w in self.workers:
w.stop()
if self.clientReader is not None:
# TODO should be covered by finalize
self.clientReader.stop()
self.clientReader = None
def getOutputFormat(self):
if self.workers:
return self.workers[-1].getOutputFormat()
else:
raise BufferError("getOutputFormat on empty chain")
def pump(self, write):
2021-08-16 14:41:18 +00:00
if self.writer is None:
2021-07-24 16:50:30 +00:00
self.setWriter(Buffer(self.getOutputFormat()))
2021-08-16 14:41:18 +00:00
self.clientReader = self.writer.getReader()
def copy():
run = True
while run:
data = None
try:
2021-08-16 14:41:18 +00:00
data = self.clientReader.read()
except ValueError:
pass
2021-07-25 15:36:32 +00:00
if data is None:
run = False
else:
write(data)
return copy