Skip to content

Commit

Permalink
Fixed a typing issue and included some file clean up
Browse files Browse the repository at this point in the history
  • Loading branch information
krcb197 committed Dec 3, 2024
1 parent 73a6fad commit 3b02ac7
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 19 deletions.
34 changes: 22 additions & 12 deletions src/peakrdl_python/lib/async_register_and_field.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
asynchronous registers and fields
"""
from enum import Enum
from typing import List, Union, Iterator, Tuple, Optional, Dict, TypeVar, cast
from typing import List, Union, Iterator, Tuple, Optional, Dict, TypeVar, cast, TypeGuard
from typing import AsyncGenerator
from abc import ABC, abstractmethod
from contextlib import asynccontextmanager
Expand Down Expand Up @@ -97,7 +97,7 @@ def _callbacks(self) -> Union[AsyncCallbackSet, AsyncCallbackSetLegacy]:
@property
@abstractmethod
def fields(self) -> \
Iterator[Union['FieldAsyncReadOnly','FieldAsyncWriteOnly', 'FieldAsyncReadWrite']]:
Iterator[Union['FieldAsyncReadOnly', 'FieldAsyncWriteOnly', 'FieldAsyncReadWrite']]:
"""
generator that produces has all the fields within the register
"""
Expand Down Expand Up @@ -193,8 +193,13 @@ def readable_fields(self) -> Iterator[Union['FieldAsyncReadOnly', 'FieldAsyncRea
"""
generator that produces has all the readable fields within the register
"""
return filter(lambda x: isinstance(x, (FieldAsyncReadOnly, FieldAsyncReadWrite)),
self.fields)
def is_readable(field: Union['FieldAsyncReadOnly',
'FieldAsyncWriteOnly',
'FieldAsyncReadWrite']) -> \
TypeGuard[Union['FieldAsyncReadOnly', 'FieldAsyncReadWrite']]:
return isinstance(field, (FieldAsyncReadOnly, FieldAsyncReadWrite))

return filter(is_readable, self.fields)

async def read_fields(self) -> Dict['str', Union[bool, Enum, int]]:
"""
Expand Down Expand Up @@ -298,8 +303,13 @@ def writable_fields(self) -> Iterator[Union['FieldAsyncWriteOnly', 'FieldAsyncRe
"""
generator that produces has all the writable fields within the register
"""
return filter(lambda x: isinstance(x, (FieldAsyncWriteOnly, FieldAsyncReadWrite)),
self.fields)
def is_writable(field: Union['FieldAsyncReadOnly',
'FieldAsyncWriteOnly',
'FieldAsyncReadWrite']) -> \
TypeGuard[Union['FieldAsyncWriteOnly', 'FieldAsyncReadWrite']]:
return isinstance(field, (FieldAsyncWriteOnly, FieldAsyncReadWrite))

return filter(is_writable, self.fields)

@abstractmethod
async def write_fields(self, **kwargs) -> None: # type: ignore[no-untyped-def]
Expand Down Expand Up @@ -348,7 +358,7 @@ def __init__(self, *,

@asynccontextmanager
async def single_read_modify_write(self, verify: bool = False, skip_write: bool = False) -> \
AsyncGenerator[Self, None]:
AsyncGenerator[Self, None]:
"""
Context manager to allow multiple field reads/write to be done with a single set of
field operations
Expand Down Expand Up @@ -1048,7 +1058,7 @@ class for an asynchronous read only register field
object
"""
__slots__ : List[str] = []
__slots__: List[str] = []

def __init__(self, *,
parent_register: ReadableAsyncRegister,
Expand All @@ -1069,7 +1079,7 @@ def __init__(self, *,
inst_name=inst_name)
# pylint: enable=duplicate-code

async def read(self) -> int: # pylint: disable=invalid-overridden-method
async def read(self) -> int: # pylint: disable=invalid-overridden-method
"""
Asynchronously reads the register that this field is located in and retries the field
value applying the required masking and shifting
Expand Down Expand Up @@ -1101,7 +1111,7 @@ class for an asynchronous write only register field
object
"""
__slots__ : List[str] = []
__slots__: List[str] = []

def __init__(self, *,
parent_register: WritableAsyncRegister,
Expand All @@ -1122,7 +1132,7 @@ def __init__(self, *,
inst_name=inst_name)
# pylint: enable=duplicate-code

async def write(self, value: int) -> None: # pylint: disable=invalid-overridden-method
async def write(self, value: int) -> None: # pylint: disable=invalid-overridden-method
"""
The behaviour of this method depends on whether the field is located in
a readable register or not:
Expand Down Expand Up @@ -1184,7 +1194,7 @@ class for an asyncronous read/write register field
object
"""
__slots__ : List[str] = []
__slots__: List[str] = []

def __init__(self, *,
parent_register: RegAsyncReadWrite,
Expand Down
21 changes: 14 additions & 7 deletions src/peakrdl_python/lib/register_and_field.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
registers and fields
"""
from enum import Enum
from typing import List, Union, Iterator, Tuple, cast, Optional, Dict, TypeVar
from typing import List, Union, Iterator, Tuple, cast, Optional, Dict, TypeVar, TypeGuard
from typing import Generator
from abc import ABC, abstractmethod
from contextlib import contextmanager
Expand Down Expand Up @@ -178,7 +178,7 @@ def __block_read_legacy(self) -> Array:
if self.__register_address_array is None:
raise RuntimeError('This address array has not be initialised')

for entry,address in enumerate(self.__register_address_array):
for entry, address in enumerate(self.__register_address_array):

# python 3.7 doesn't have the callback defined as protocol so mypy doesn't
# recognise the arguments in the call back functions
Expand Down Expand Up @@ -469,7 +469,6 @@ class RegReadOnly(Reg, ABC):
class for a read only register
Args:
callbacks: set of callback to be used for accessing the hardware or simulator
address: address of the register
width: width of the register in bits
accesswidth: minimum access width of the register in bits
Expand Down Expand Up @@ -546,7 +545,11 @@ def readable_fields(self) -> Iterator[Union['FieldReadOnly', 'FieldReadWrite']]:
"""
generator that produces has all the readable fields within the register
"""
return filter(lambda x: isinstance(x, (FieldReadOnly, FieldReadWrite)), self.fields)
def is_readable(field: Union['FieldReadOnly', 'FieldWriteOnly', 'FieldReadWrite']) -> \
TypeGuard[Union['FieldReadOnly', 'FieldReadWrite']]:
return isinstance(field, (FieldReadOnly, FieldReadWrite))

return filter(is_readable, self.fields)

def read_fields(self) -> Dict['str', Union[bool, Enum, int]]:
"""
Expand Down Expand Up @@ -642,7 +645,11 @@ def writable_fields(self) -> Iterator[Union['FieldWriteOnly', 'FieldReadWrite']]
"""
generator that produces has all the readable fields within the register
"""
return filter(lambda x: isinstance(x, (FieldWriteOnly, FieldReadWrite)), self.fields)
def is_writable(field: Union['FieldReadOnly', 'FieldWriteOnly', 'FieldReadWrite']) -> \
TypeGuard[Union['FieldWriteOnly', 'FieldReadWrite']]:
return isinstance(field, (FieldWriteOnly, FieldReadWrite))

return filter(is_writable, self.fields)

@abstractmethod
def write_fields(self, **kwargs) -> None: # type: ignore[no-untyped-def]
Expand Down Expand Up @@ -1060,7 +1067,7 @@ class for a write only register field
object
"""
__slots__ : List[str] = []
__slots__: List[str] = []

def __init__(self, *,
parent_register: WritableRegister,
Expand Down Expand Up @@ -1139,7 +1146,7 @@ class for a read/write register field
object
"""
__slots__ : List[str] = []
__slots__: List[str] = []

def __init__(self, *,
parent_register: RegReadWrite,
Expand Down

0 comments on commit 3b02ac7

Please sign in to comment.