Introduce InInstruction and update existing logic

pioemu/decoding/instruction_decoder.py

@@ -11,7 +11,7 @@
 # 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 pioemu.instruction import JmpInstruction
+from pioemu.instruction import InInstruction, JmpInstruction
 
 
 class InstructionDecoder:
@@ -27,9 +27,9 @@ def __init__(self, side_set_count: int = 0):
         """
         self.side_set_count = side_set_count
         self.bits_for_delay = 5 - self.side_set_count
-        self.delay_mask = (1 << self.bits_for_delay) - 1
+        self.delay_cycles_mask = (1 << self.bits_for_delay) - 1
 
-    def decode(self, opcode: int) -> JmpInstruction | None:
+    def decode(self, opcode: int) -> InInstruction | JmpInstruction | None:
         """
         Decodes an opcode into an object representing the instruction and its parameters.
 
@@ -40,18 +40,36 @@ def decode(self, opcode: int) -> JmpInstruction | None:
         Instruction: Representation of the given opcode or None when invalid/not supported
         """
 
-        if (opcode >> 13) & 7 == 0:
-            return self._decode_jmp(opcode)
+        match (opcode >> 13) & 7:
+            case 0:
+                return self._decode_jmp(opcode)
+            case 2:
+                return self._decode_in(opcode)
+            case _:
+                return None
 
-        return None
+    def _decode_in(self, opcode: int) -> InInstruction | None:
+        bit_count = opcode & 0x1F
+        if bit_count == 0:
+            bit_count = 32
+
+        source = (opcode >> 5) & 7
+        delay_cycles, side_set_value = self._extract_delay_cycles_and_side_set(opcode)
+
+        return InInstruction(opcode, source, bit_count, delay_cycles, side_set_value)
 
     def _decode_jmp(self, opcode: int) -> JmpInstruction:
         target_address = opcode & 0x1F
         condition = (opcode >> 5) & 7
-
-        delay_cycles = (opcode >> 8) & self.delay_mask
-        side_set_value = opcode >> 8 + self.bits_for_delay
+        delay_cycles, side_set_value = self._extract_delay_cycles_and_side_set(opcode)
 
         return JmpInstruction(
             opcode, target_address, condition, delay_cycles, 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

@@ -17,7 +17,7 @@
 from typing import Callable, Generator, List, Tuple
 
 from .decoding.instruction_decoder import InstructionDecoder as NewInstructionDecoder
-from .instruction import Emulation, JmpInstruction, ProgramCounterAdvance
+from .instruction import Emulation, InInstruction, JmpInstruction, ProgramCounterAdvance
 from .instruction_decoder import InstructionDecoder
 from .shift_register import ShiftRegister
 from .state import State
@@ -153,7 +153,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
@@ -179,7 +179,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
@@ -202,10 +208,6 @@ def emulate(
         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
 
@@ -273,6 +275,7 @@ def _apply_delay_value(
 
 
 def _apply_side_effects(
+    instruction: InInstruction,
     opcode: int,
     state: State,
     auto_push: bool,
@@ -281,7 +284,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

pioemu/instruction.py

@@ -24,6 +24,15 @@ class ProgramCounterAdvance(Enum):
     NEVER = auto()
 
 
+@dataclass(frozen=True)
+class InInstruction:
+    opcode: int
+    source: int  # TODO: Use an enumeration instead of an integer?
+    bit_count: int
+    delay_cycles: int
+    side_set_value: int
+
+
 @dataclass(frozen=True)
 class JmpInstruction:
     opcode: int

pioemu/instruction_decoder.py

@@ -29,6 +29,7 @@
 )
 from .instruction import (
     Emulation,
+    InInstruction,
     JmpInstruction,
     ProgramCounterAdvance,
 )
@@ -171,7 +172,9 @@ def __init__(
             None,
         ]
 
-    def create_emulation(self, instruction: JmpInstruction) -> Emulation | None:
+    def create_emulation(
+        self, instruction: JmpInstruction | InInstruction
+    ) -> Emulation | None:
         """
         Returns an emulation for the given instruction.
 
@@ -182,10 +185,13 @@ def create_emulation(self, instruction: JmpInstruction) -> Emulation | None:
         Emulation: Emulation for the given instruction or None when invalid/not supported
         """
 
-        if isinstance(instruction, JmpInstruction):
-            return self._decode_jmp(instruction)
-
-        return None
+        match instruction:
+            case JmpInstruction():
+                return self._decode_jmp(instruction)
+            case InInstruction():
+                return self._decode_in(instruction)
+            case _:
+                return None
 
     def decode(self, opcode: int) -> Emulation | None:
         """
@@ -216,7 +222,7 @@ def _decode_jmp(self, instruction: JmpInstruction) -> Emulation | None:
             instruction,
         )
 
-    def _decode_mov(self, opcode: int) -> Emulation:
+    def _decode_mov(self, opcode: int) -> Emulation | None:
         read_from_source = self.mov_sources[opcode & 7]
 
         destination = (opcode >> 5) & 7
@@ -243,21 +249,21 @@ def _decode_mov(self, opcode: int) -> Emulation:
             program_counter_advance,
         )
 
-    def _decode_in(self, opcode: int) -> Emulation:
-        read_from_source = self.in_sources[(opcode >> 5) & 7]
-
-        bit_count = opcode & 0x1F
-
-        if bit_count == 0:
-            bit_count = 32
+    def _decode_in(self, instruction: InInstruction) -> Emulation:
+        read_from_source = self.in_sources[instruction.source]
 
         return Emulation(
             always,
-            partial(shift_into_isr, read_from_source, self.shift_isr_method, bit_count),
+            partial(
+                shift_into_isr,
+                read_from_source,
+                self.shift_isr_method,
+                instruction.bit_count,
+            ),
             ProgramCounterAdvance.ALWAYS,
         )
 
-    def _decode_out(self, opcode: int) -> Emulation:
+    def _decode_out(self, opcode: int) -> Emulation | None:
         destination = (opcode >> 5) & 7
         write_to_destination = self.out_destinations[destination]
 
@@ -289,7 +295,7 @@ def emulate_out(state: State) -> State:
 
         return Emulation(always, emulate_out, ProgramCounterAdvance.ALWAYS)
 
-    def _decode_set(self, opcode: int) -> Emulation:
+    def _decode_set(self, opcode: int) -> Emulation | None:
         write_to_destination = self.set_destinations[(opcode >> 5) & 7]
 
         if write_to_destination is None:

tests/decoding/test_instruction_decoder.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 import pytest
 from pioemu.decoding.instruction_decoder import InstructionDecoder
-from pioemu.instruction import JmpInstruction
+from pioemu.instruction import InInstruction, JmpInstruction
 from tests.opcodes import Opcodes
 
 
@@ -23,6 +23,44 @@ def test_none_returned_for_unsupported_opcodes():
     assert instruction is None
 
 
+@pytest.mark.parametrize(
+    "opcode, side_set_count, expected_source, expected_bit_count, expected_delay_cycles, expected_side_set_value",
+    [
+        pytest.param(0x5C01, 0, 0, 1, 28, 0, id="in pins, 1 [28]"),
+        pytest.param(0x5B21, 0, 1, 1, 27, 0, id="in x, 1 [27]"),
+        pytest.param(0x4F41, 3, 2, 1, 3, 3, id="in y, 1 side 0b011 [3]"),
+        pytest.param(0x5C61, 4, 3, 1, 0, 14, id="in null, 1 side 0b1110"),
+        pytest.param(0x50C1, 1, 6, 1, 0, 1, id="in isr, 1 side 0b1"),
+        pytest.param(0x42E1, 5, 7, 1, 0, 2, id="in osr, 1 side 0b00010"),
+        pytest.param(0x5AE0, 4, 7, 32, 0, 13, id="in osr, 32 side 0b1101"),
+        pytest.param(0x4DC0, 2, 6, 32, 5, 1, id="in isr, 32 side 0b01 [5]"),
+        pytest.param(0x5060, 2, 3, 32, 0, 2, id="in null, 32 side 0b10"),
+        pytest.param(0x5340, 3, 2, 32, 3, 4, id="in y, 32 side 0b100 [3]"),
+        pytest.param(0x4820, 2, 1, 32, 0, 1, id="in x, 32 side 0b01"),
+        pytest.param(0x4D00, 5, 0, 32, 0, 13, id="in pins, 32 side 0b01101"),
+    ],
+)
+def test_decoding_of_in_instruction(
+    opcode: int,
+    side_set_count: int,
+    expected_source: int,
+    expected_bit_count: int,
+    expected_delay_cycles: int,
+    expected_side_set_value: int,
+):
+    instruction_decoder = InstructionDecoder(side_set_count)
+
+    decoded_instruction = instruction_decoder.decode(opcode)
+
+    assert decoded_instruction == InInstruction(
+        opcode,
+        expected_source,
+        expected_bit_count,
+        expected_delay_cycles,
+        expected_side_set_value,
+    )
+
+
 @pytest.mark.parametrize(
     "opcode, side_set_count, expected_target_address, expected_condition, expected_delay_cycles, expected_side_set_value",
     [
@@ -52,12 +90,13 @@ def test_decoding_of_jmp_instruction(
     expected_side_set_value: int,
 ):
     instruction_decoder = InstructionDecoder(side_set_count)
-    instruction = instruction_decoder.decode(opcode)
 
-    assert isinstance(instruction, JmpInstruction)
+    decoded_instruction = instruction_decoder.decode(opcode)
 
-    assert instruction.opcode == opcode
-    assert instruction.target_address == expected_target_address
-    assert instruction.condition == expected_condition
-    assert instruction.delay_cycles == expected_delay_cycles
-    assert instruction.side_set_value == expected_side_set_value
+    assert decoded_instruction == JmpInstruction(
+        opcode,
+        expected_target_address,
+        expected_condition,
+        expected_delay_cycles,
+        expected_side_set_value,
+    )