Refactor instruction decoding

pioemu/decoding/instruction_decoder.py

@@ -0,0 +1,125 @@
+# Copyright 2025 Nathan Young
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from typing import Optional
+from pioemu.instruction import (
+    InInstruction,
+    Instruction,
+    JmpInstruction,
+    OutInstruction,
+    PushInstruction,
+)
+
+
+class InstructionDecoder:
+    """
+    Decodes state-machine opcodes into higher-level representations.
+    """
+
+    def __init__(self, side_set_count: int = 0):
+        """
+        Parameters
+        ----------
+        side_set_count (int): Number of bits of side-set information encoded into opcodes.
+        """
+        self.side_set_count = side_set_count
+        self.bits_for_delay = 5 - self.side_set_count
+        self.delay_cycles_mask = (1 << self.bits_for_delay) - 1
+
+    def decode(self, opcode: int) -> Optional[Instruction]:
+        """
+        Decodes an opcode into an object representing the instruction and its parameters.
+
+        Parameters:
+        opcode (int): The opcode to decode.
+
+        Returns:
+        Instruction: Representation of the given opcode or None when invalid/not supported
+        """
+
+        match (opcode >> 13) & 7:
+            case 0:
+                return self._decode_jmp(opcode)
+            case 2:
+                return self._decode_in(opcode)
+            case 3:
+                return self._decode_out(opcode)
+            case 4:
+                return self._decode_push(opcode)
+            case _:
+                return None
+
+    def _decode_in(self, opcode: int) -> Optional[InInstruction]:
+        bit_count = opcode & 0x1F
+        if bit_count == 0:
+            bit_count = 32
+
+        source = (opcode >> 5) & 7
+
+        # Check if source has been reserved for future use
+        if source == 4 or source == 5:
+            return None
+
+        delay_cycles, side_set_value = self._extract_delay_cycles_and_side_set(opcode)
+
+        return InInstruction(
+            opcode=opcode,
+            source=source,
+            bit_count=bit_count,
+            delay_cycles=delay_cycles,
+            side_set_value=side_set_value,
+        )
+
+    def _decode_jmp(self, opcode: int) -> JmpInstruction:
+        delay_cycles, side_set_value = self._extract_delay_cycles_and_side_set(opcode)
+
+        return JmpInstruction(
+            opcode=opcode,
+            target_address=opcode & 0x1F,
+            condition=(opcode >> 5) & 7,
+            delay_cycles=delay_cycles,
+            side_set_value=side_set_value,
+        )
+
+    def _decode_out(self, opcode: int) -> OutInstruction:
+        bit_count = opcode & 0x1F
+        if bit_count == 0:
+            bit_count = 32
+
+        delay_cycles, side_set_value = self._extract_delay_cycles_and_side_set(opcode)
+
+        return OutInstruction(
+            opcode=opcode,
+            destination=(opcode >> 5) & 7,
+            bit_count=bit_count,
+            delay_cycles=delay_cycles,
+            side_set_value=side_set_value,
+        )
+
+    def _decode_push(self, opcode: int) -> Optional[PushInstruction]:
+        delay_cycles, side_set_value = self._extract_delay_cycles_and_side_set(opcode)
+
+        return PushInstruction(
+            opcode=opcode,
+            if_full=bool(opcode & 0x0040),
+            block=bool(opcode & 0x0020),
+            delay_cycles=delay_cycles,
+            side_set_value=side_set_value,
+        )
+
+    def _extract_delay_cycles_and_side_set(self, opcode: int):
+        delay_cycles_and_side_set = (opcode >> 8) & 0x1F
+        delay_cycles = delay_cycles_and_side_set & self.delay_cycles_mask
+        side_set_value = delay_cycles_and_side_set >> self.bits_for_delay
+
+        return (delay_cycles, side_set_value)

pioemu/emulation.py

@@ -14,9 +14,17 @@
 import inspect
 import logging
 from dataclasses import replace
-from typing import Callable, Generator, List, Tuple
-
-from .instruction import Emulation, ProgramCounterAdvance
+from typing import Callable, Generator, List, Optional, Tuple
+
+from .decoding.instruction_decoder import InstructionDecoder as NewInstructionDecoder
+from .instruction import (
+    Emulation,
+    InInstruction,
+    Instruction,
+    JmpInstruction,
+    OutInstruction,
+    ProgramCounterAdvance,
+)
 from .instruction_decoder import InstructionDecoder
 from .shift_register import ShiftRegister
 from .state import State
@@ -107,7 +115,9 @@ def emulate(
         ShiftRegister.shift_right if shift_osr_right else ShiftRegister.shift_left
     )
 
-    instruction_decoder = InstructionDecoder(
+    new_instruction_decoder = NewInstructionDecoder(side_set_count)
+
+    old_instruction_decoder = InstructionDecoder(
         shift_isr_method, shift_osr_method, jmp_pin
     )
 
@@ -133,7 +143,13 @@ def emulate(
             opcode, side_set_count
         )
 
-        emulation = instruction_decoder.decode(opcode)
+        instruction = new_instruction_decoder.decode(opcode)
+
+        emulation = (
+            old_instruction_decoder.create_emulation(instruction)
+            if instruction
+            else old_instruction_decoder.decode(opcode)
+        )
 
         if emulation is None:
             return
@@ -144,7 +160,7 @@ def emulate(
             # into a full FIFO. Please refer to the Autopush Details section (3.5.4.1) within the
             # RP2040 Datasheet for more details.
             if (
-                _is_in_instruction(opcode)
+                isinstance(instruction, InInstruction)
                 and auto_push
                 and current_state.input_shift_register.counter >= push_threshold
                 and len(current_state.receive_fifo) >= 4
@@ -155,7 +171,7 @@ def emulate(
             # the same clock cycle. Please refer to the Autopull Details section (3.4.5.2) within
             # the RP2040 Datasheet for more details.
             elif (
-                _is_out_instruction(opcode)
+                isinstance(instruction, OutInstruction)
                 and auto_pull
                 and current_state.output_shift_register.counter >= pull_threshold
             ):
@@ -170,7 +186,13 @@ def emulate(
                 stalled = True
 
         current_state = _apply_side_effects(
-            opcode, current_state, auto_push, push_threshold, auto_pull, pull_threshold
+            instruction,
+            opcode,
+            current_state,
+            auto_push,
+            push_threshold,
+            auto_pull,
+            pull_threshold,
         )
 
         # TODO: Check that the following still applies when an instruction is stalled
@@ -185,22 +207,14 @@ def emulate(
             )
 
             current_state = _apply_delay_value(
-                opcode, condition_met, delay_value, current_state
+                instruction, condition_met, delay_value, current_state
             )
 
         current_state = replace(current_state, clock=current_state.clock + 1)
 
         yield (previous_state, current_state)
 
 
-def _is_in_instruction(opcode: int) -> bool:
-    return ((opcode >> 13) & 7) == 2
-
-
-def _is_out_instruction(opcode: int) -> bool:
-    return ((opcode >> 13) & 7) == 3
-
-
 def _normalize_input_source(logger: logging.Logger, input_source: Callable):
     parameter_type = _get_input_source_parameter_type(input_source)
 
@@ -252,17 +266,19 @@ def _advance_program_counter(
 
 
 def _apply_delay_value(
-    opcode: int, condition_met: bool, delay_value: int, state: State
+    instruction: Optional[Instruction],
+    condition_met: bool,
+    delay_value: int,
+    state: State,
 ) -> State:
-    jump_instruction = (opcode >> 13) == 0
-
-    if jump_instruction or condition_met:
+    if isinstance(instruction, JmpInstruction) or condition_met:
         return replace(state, clock=state.clock + delay_value)
 
     return state
 
 
 def _apply_side_effects(
+    instruction: Optional[Instruction],
     opcode: int,
     state: State,
     auto_push: bool,
@@ -271,7 +287,7 @@ def _apply_side_effects(
     pull_threshold: int,
 ) -> State:
     if (
-        _is_in_instruction(opcode)
+        isinstance(instruction, InInstruction)
         and auto_push
         and state.input_shift_register.counter >= push_threshold
         and len(state.receive_fifo) < 4
@@ -286,7 +302,7 @@ def _apply_side_effects(
             input_shift_register=new_input_shift_register,
         )
     elif (
-        _is_out_instruction(opcode)
+        isinstance(instruction, OutInstruction)
         and auto_pull
         and state.output_shift_register.counter >= pull_threshold
         and state.transmit_fifo

pioemu/instruction.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 from dataclasses import dataclass
 from enum import auto, Enum
-from typing import Callable
+from typing import Callable, Optional
 from .state import State
 
 
@@ -24,8 +24,40 @@ class ProgramCounterAdvance(Enum):
     NEVER = auto()
 
 
+@dataclass(frozen=True, kw_only=True)
+class Instruction:
+    opcode: int
+    delay_cycles: int
+    side_set_value: int
+
+
+@dataclass(frozen=True, kw_only=True)
+class InInstruction(Instruction):
+    source: int  # TODO: Use an enumeration instead of an integer?
+    bit_count: int
+
+
+@dataclass(frozen=True, kw_only=True)
+class JmpInstruction(Instruction):
+    target_address: int
+    condition: int  # TODO: Use an enumeration instead of an integer?
+
+
+@dataclass(frozen=True, kw_only=True)
+class OutInstruction(Instruction):
+    destination: int  # TODO: Use an enumeration instead of an integer?
+    bit_count: int
+
+
+@dataclass(frozen=True, kw_only=True)
+class PushInstruction(Instruction):
+    if_full: bool
+    block: bool
+
+
 @dataclass(frozen=True)
 class Emulation:
     condition: Callable[[State], bool]
     emulate: Callable[[State], State | None]
     program_counter_advance: ProgramCounterAdvance
+    instruction: Optional[Instruction] = None