arno/trezor-firmware
1
0
Fork 0
mirror of https://github.com/trezor/trezor-firmware.git synced 2024-11-26 01:18:28 +00:00
Code Issues Releases Wiki Activity

style(core/src): use new syntax for typing

Browse Source
This commit is contained in:
Pavol Rusnak 2020-10-16 19:39:32 +02:00 committed by matejcik
parent cd829a99de
commit a1dda836ec
16 changed files with 55 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
core/src/protobuf.py
View File

@ -212,7 +212,7 @@ def load_message(
raise ValueError # experimental messages not enabled
# we need to avoid calling __init__, which enforces required arguments
msg = object.__new__(msg_type) # type: LoadedMessageType
msg: LoadedMessageType = object.__new__(msg_type)
# pre-seed the object with defaults
for fname, _, fdefault in fields.values():
if fdefault is FLAG_REPEATED:
@ -224,7 +224,7 @@ def load_message(
if False:
SingularValue = Union[int, bool, bytearray, str, MessageType]
Value = Union[SingularValue, List[SingularValue]]
fvalue = 0 # type: Value
fvalue: Value = 0
while True:
try:

10
core/src/storage/cache.py
View File

@ -2,7 +2,7 @@ from trezor import wire
from trezor.crypto import random
if False:
from typing import Optional, Dict, List
from typing import Optional, Dict, List, Any
_MAX_SESSIONS_COUNT = 10
_SESSIONLESS_FLAG = 128
@ -18,10 +18,10 @@ APP_COMMON_SEED_WITHOUT_PASSPHRASE = 1 | _SESSIONLESS_FLAG
APP_COMMON_SAFETY_CHECKS_TEMPORARY = 2 | _SESSIONLESS_FLAG
_active_session_id = None # type: Optional[bytes]
_caches = {} # type: Dict[bytes, Dict[int, Any]]
_session_ids = [] # type: List[bytes]
_sessionless_cache = {} # type: Dict[int, Any]
_active_session_id: Optional[bytes] = None
_caches: Dict[bytes, Dict[int, Any]] = {}
_session_ids: List[bytes] = []
_sessionless_cache: Dict[int, Any] = {}
if False:
from typing import Any, Callable, TypeVar

4
core/src/storage/device.py
View File

@ -40,8 +40,8 @@ _EXPERIMENTAL_FEATURES = const(0x15) # bool (0x01 or empty)
_DEFAULT_BACKUP_TYPE = BackupType.Bip39
SAFETY_CHECK_LEVEL_STRICT = const(0) # type: Literal[0]
SAFETY_CHECK_LEVEL_PROMPT = const(1) # type: Literal[1]
SAFETY_CHECK_LEVEL_STRICT : Literal[0] = const(0)
SAFETY_CHECK_LEVEL_PROMPT : Literal[1] = const(1)
_DEFAULT_SAFETY_CHECK_LEVEL = SAFETY_CHECK_LEVEL_STRICT
if False:
StorageSafetyCheckLevel = Literal[0, 1]

4
core/src/trezor/crypto/slip39.py
View File

@ -229,7 +229,7 @@ def split_ems(
# Split the Encrypted Master Secret on the group level.
group_shares = _split_secret(group_threshold, len(groups), encrypted_master_secret)
mnemonics = [] # type: List[List[str]]
mnemonics: List[List[str]] = []
for (member_threshold, member_count), (group_index, group_secret) in zip(
groups, group_shares
):
@ -599,7 +599,7 @@ def _decode_mnemonics(
group_counts = set()
# { group_index : [threshold, set_of_member_shares] }
groups = {} # type: MnemonicGroups
groups: MnemonicGroups = {}
for mnemonic in mnemonics:
share = decode_mnemonic(mnemonic)
identifiers.add(share.identifier)

28
core/src/trezor/loop.py
View File

@ -30,23 +30,23 @@ if False:
Finalizer = Callable[[Task, Any], None]
# function to call after every task step
after_step_hook = None # type: Optional[Callable[[], None]]
after_step_hook: Optional[Callable[[], None]] = None
# tasks scheduled for execution in the future
_queue = utimeq.utimeq(64)
# tasks paused on I/O
_paused = {} # type: Dict[int, Set[Task]]
_paused: Dict[int, Set[Task]] = {}
# functions to execute after a task is finished
_finalizers = {} # type: Dict[int, Finalizer]
_finalizers: Dict[int, Finalizer] = {}
# reference to the task that is currently executing
this_task = None # type: Optional[Task]
this_task: Optional[Task] = None
if __debug__:
# synthetic event queue
synthetic_events = [] # type: List[Tuple[int, Any]]
synthetic_events: List[Tuple[int, Any]] = []
class TaskClosed(Exception):
@ -300,8 +300,8 @@ class race(Syscall):
def __init__(self, *children: Awaitable, exit_others: bool = True) -> None:
self.children = children
self.exit_others = exit_others
self.finished = [] # type: List[Awaitable] # children that finished
self.scheduled = [] # type: List[Task] # scheduled wrapper tasks
self.finished: List[Awaitable] = [] # children that finished
self.scheduled: List[Task] = [] # scheduled wrapper tasks
def handle(self, task: Task) -> None:
"""
@ -387,7 +387,7 @@ class chan:
def __init__(self, ch: "chan", value: Any) -> None:
self.ch = ch
self.value = value
self.task = None # type: Optional[Task]
self.task: Optional[Task] = None
def handle(self, task: Task) -> None:
self.task = task
@ -396,15 +396,15 @@ class chan:
class Take(Syscall):
def __init__(self, ch: "chan") -> None:
self.ch = ch
self.task = None # type: Optional[Task]
self.task: Optional[Task] = None
def handle(self, task: Task) -> None:
self.task = task
self.ch._schedule_take(task)
def __init__(self) -> None:
self.putters = [] # type: List[Tuple[Optional[Task], Any]]
self.takers = [] # type: List[Task]
self.putters: List[Tuple[Optional[Task], Any]] = []
self.takers: List[Task] = []
def put(self, value: Any) -> Awaitable[None]: # type: ignore
put = chan.Put(self, value)
@ -484,10 +484,10 @@ class spawn(Syscall):
def __init__(self, task: Task) -> None:
self.task = task
self.callback = None # type: Optional[Task]
self.finalizer_callback = None # type: Optional[Callable[["spawn"], None]]
self.callback: Optional[Task] = None
self.finalizer_callback: Optional[Callable[["spawn"], None]] = None
self.finished = False
self.return_value = None # type: Any
self.return_value: Any = None
# schedule task immediately
if __debug__:

6
core/src/trezor/pin.py
View File

@ -8,9 +8,9 @@ def pin_to_int(pin: str) -> int:
return int("1" + pin)
_previous_progress = None # type: Optional[int]
_previous_seconds = None # type: Optional[int]
keepalive_callback = None # type: Any
_previous_progress: Optional[int] = None
_previous_seconds: Optional[int] = None
keepalive_callback: Any = None
def show_pin_timeout(seconds: int, progress: int, message: str) -> bool:

2
core/src/trezor/ui/__init__.py
View File

@ -308,7 +308,7 @@ class Layout(Component):
"""
BACKLIGHT_LEVEL = style.BACKLIGHT_NORMAL
RENDER_SLEEP = loop.sleep(_RENDER_DELAY_MS) # type: loop.Syscall
RENDER_SLEEP: loop.Syscall = loop.sleep(_RENDER_DELAY_MS)
async def __iter__(self) -> ResultValue:
"""

2
core/src/trezor/ui/checklist.py
View File

@ -18,7 +18,7 @@ class Checklist(ui.Component):
super().__init__()
self.title = title
self.icon = icon
self.items = [] # type: List[ChecklistItem]
self.items: List[ChecklistItem] = []
self.active = 0
def add(self, item: ChecklistItem) -> None:

6
core/src/trezor/ui/confirm.py
View File

@ -48,9 +48,7 @@ class Confirm(ui.Layout):
area = ui.grid(13, cells_x=2)
else:
area = ui.grid(9, n_x=2)
self.confirm = Button(
area, confirm, confirm_style
) # type: Optional[Button]
self.confirm: Optional[Button] = Button(area, confirm, confirm_style)
self.confirm.on_click = self.on_confirm # type: ignore
else:
self.confirm = None
@ -62,7 +60,7 @@ class Confirm(ui.Layout):
area = ui.grid(12, cells_x=1)
else:
area = ui.grid(8, n_x=2)
self.cancel = Button(area, cancel, cancel_style) # type: Optional[Button]
self.cancel: Optional[Button] = Button(area, cancel, cancel_style)
self.cancel.on_click = self.on_cancel # type: ignore
else:
self.cancel = None

8
core/src/trezor/ui/loader.py
View File

@ -12,8 +12,8 @@ class LoaderDefault:
class normal:
bg_color = ui.BG
fg_color = ui.GREEN
icon = None # type: Optional[str]
icon_fg_color = None # type: Optional[int]
icon: Optional[str] = None
icon_fg_color: Optional[int] = None
class active(normal):
bg_color = ui.BG
@ -43,8 +43,8 @@ class Loader(ui.Component):
self.normal_style = style.normal
self.active_style = style.active
self.target_ms = _TARGET_MS
self.start_ms = None # type: Optional[int]
self.stop_ms = None # type: Optional[int]
self.start_ms: Optional[int] = None
self.stop_ms: Optional[int] = None
def start(self) -> None:
self.start_ms = utime.ticks_ms()

6
core/src/trezor/ui/passphrase.py
View File

@ -145,7 +145,7 @@ class PassphraseKeyboard(ui.Layout):
self.done.on_click = self.on_confirm # type: ignore
self.keys = key_buttons(KEYBOARD_KEYS[self.page], self)
self.pending_button = None # type: Optional[KeyButton]
self.pending_button: Optional[KeyButton] = None
self.pending_index = 0
def dispatch(self, event: int, x: int, y: int) -> None:
@ -246,11 +246,11 @@ class PassphraseKeyboard(ui.Layout):
raise ui.Result(self.input.text)
def create_tasks(self) -> Tuple[loop.Task, ...]:
tasks = (
tasks: Tuple[loop.Task, ...] = (
self.handle_input(),
self.handle_rendering(),
self.handle_paging(),
) # type: Tuple[loop.Task, ...]
)
if __debug__:
from apps.debug import input_signal

4
core/src/trezor/ui/scroll.py
View File

@ -98,11 +98,11 @@ class Paginated(ui.Layout):
self.on_change()
def create_tasks(self) -> Tuple[loop.Task, ...]:
tasks = (
tasks: Tuple[loop.Task, ...] = (
self.handle_input(),
self.handle_rendering(),
self.handle_paging(),
) # type: Tuple[loop.Task, ...]
)
if __debug__:
# XXX This isn't strictly correct, as it allows *any* Paginated layout to be

2
core/src/trezor/ui/text.py
View File

@ -135,7 +135,7 @@ class Text(ui.Component):
self.icon_color = icon_color
self.max_lines = max_lines
self.new_lines = new_lines
self.content = [] # type: List[TextContent]
self.content: List[TextContent] = []
def normal(self, *content: TextContent) -> None:
self.content.append(ui.NORMAL)

14
core/src/trezor/wire/__init__.py
View File

@ -67,14 +67,14 @@ if False:
# Maps a wire type directly to a handler.
workflow_handlers = {} # type: Dict[int, Handler]
workflow_handlers: Dict[int, Handler] = {}
# Maps a wire type to a tuple of package and module. This allows handlers
# to be dynamically imported when such message arrives.
workflow_packages = {} # type: Dict[int, Tuple[str, str]]
workflow_packages: Dict[int, Tuple[str, str]] = {}
# If set to False protobuf messages marked with "unstable" option are rejected.
experimental_enabled = False # type: bool
experimental_enabled: bool = False
def add(wire_type: int, pkgname: str, modname: str) -> None:
@ -163,7 +163,7 @@ class Context:
self.buffer_reader = utils.BufferReader(self.buffer)
self.buffer_writer = utils.BufferWriter(self.buffer)
self._field_cache = {} # type: protobuf.FieldCache
self._field_cache: protobuf.FieldCache = {}
async def call(
self,
@ -307,8 +307,8 @@ async def handle_session(
iface: WireInterface, session_id: int, use_workflow: bool = True
) -> None:
ctx = Context(iface, session_id)
next_msg = None # type: Optional[codec_v1.Message]
res_msg = None # type: Optional[protobuf.MessageType]
next_msg: Optional[codec_v1.Message] = None
res_msg: Optional[protobuf.MessageType] = None
req_type = None
req_msg = None
while True:
@ -361,7 +361,7 @@ async def handle_session(
# We found a valid handler for this message type.
# Workflow task, declared for the finally block
wf_task = None # type: Optional[HandlerTask]
wf_task: Optional[HandlerTask] = None
# Here we make sure we always respond with a Failure response
# in case of any errors.

2
core/src/trezor/wire/codec_v1.py
View File

@ -44,7 +44,7 @@ async def read_message(iface: WireInterface, buffer: utils.BufferType) -> Messag
if msize > len(buffer):
# allocate a new buffer to fit the message
try:
mdata = bytearray(msize) # type: utils.BufferType
mdata: utils.BufferType = bytearray(msize)
except MemoryError:
mdata = bytearray(_REP_LEN)
read_and_throw_away = True

10
core/src/trezor/workflow.py
View File

@ -16,14 +16,14 @@ if __debug__:
# Set of workflow tasks. Multiple workflows can be running at the same time.
tasks = set() # type: Set[loop.spawn]
tasks: Set[loop.spawn] = set()
# Default workflow task, if a default workflow is running. Default workflow
# is not contained in the `tasks` set above.
default_task = None # type: Optional[loop.spawn]
default_task: Optional[loop.spawn] = None
# Constructor for the default workflow. Returns a workflow task.
default_constructor = None # type: Optional[Callable[[], loop.Task]]
default_constructor: Optional[Callable[[], loop.Task]] = None
def _on_start(workflow: loop.spawn) -> None:
@ -159,8 +159,8 @@ class IdleTimer:
"""
def __init__(self) -> None:
self.timeouts = {} # type: Dict[IdleCallback, int]
self.tasks = {} # type: Dict[IdleCallback, loop.Task]
self.timeouts: Dict[IdleCallback, int] = {}
self.tasks: Dict[IdleCallback, loop.Task] = {}
async def _timeout_task(self, callback: IdleCallback) -> None:
# This function is async, so the result of self._timeout_task() is an awaitable,