6502/py65
1
0
Fork 0
mirror of https://github.com/mnaberez/py65.git synced 2025-01-22 13:31:39 +00:00
Code Issues Projects Releases Wiki Activity

Changed ObservableMemory interface. (offe)

Browse Source
This commit is contained in:
Mike Naberezny 2009-04-04 19:40:22 -07:00
parent 354117fef1
commit 1ed2582247
1 changed files with 32 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/py65/memory.py
View File

@ -1,44 +1,45 @@
class ObservableMemory: from collections import defaultdict
READ = 0
WRITE = 1
RW = 2
class ObservableMemory:
def __init__(self, subject=None): def __init__(self, subject=None):
if subject is None: if subject is None:
subject = [] subject = 0x10000 * [0x00]
for addr in range(0x0000, 0xFFFF+1):
subject.insert(addr, 0x00)
self._subject = subject self._subject = subject
self._observers = []
self._listeners = defaultdict(list)
self._providers = defaultdict(list)
def __setitem__(self, address, value): def __setitem__(self, address, value):
for oper, addr_range, callback in self._observers: callbacks = self._listeners[address]
if address in addr_range: for callback in callbacks:
if (oper == self.RW) or (oper == self.WRITE): result = callback(address, value)
result = callback(self.WRITE, address, value) if result is not None:
if result is not None: value = result
value = result
self._subject[address] = value
def __getitem__(self, address): def __getitem__(self, address):
for oper, addr_range, callback in self._observers: callbacks = self._providers[address]
if address in addr_range: final_result = None
if (oper == self.RW) or (oper == self.READ):
result = callback(self.READ, address, None) for callback in callbacks:
if result is not None: result = callback(address)
return result if result is not None:
return self._subject[address] final_result = result
if final_result:
return final_result
else:
return self._subject[address]
def __getattr__(self, address): def __getattr__(self, address):
return getattr(self._subject, address) return getattr(self._subject, address)
def subscribe(self, operation, addr_range, callback): def register_provider(self, address_range, callback):
if operation not in (self.READ, self.WRITE, self.RW): for address in address_range:
raise ValueError("Unsupported operation") self._providers[address].append(callback)
self._observers.append([operation, addr_range, callback])
def dma_read(self, key): def register_listener(self, address_range, callback):
return self._subject[key] for address in address_range:
self._listeners[address].append(callback)
def dma_write(self, key, value):
self._subject[key] = value def write(self, start_address, bytes):
self._subject[start_address:start_address+len(bytes)] = bytes