11"""Module defining the base class and static func for interfaces."""
22from abc import ABCMeta , abstractmethod
33from dataclasses import dataclass , field
4- from functools import lru_cache
54
65from uoshardware import Persistence , UOSUnsupportedError
76
@@ -69,6 +68,14 @@ def enumerate_functions() -> list:
6968 if isinstance (getattr (UOSFunctions , member_name ), UOSFunction )
7069 ]
7170
71+ @staticmethod
72+ def get_from_address (address : int ) -> UOSFunction | None :
73+ """Look up function from the address."""
74+ for function in UOSFunctions .enumerate_functions ():
75+ if address in function .address_lut .values ():
76+ return function # function located.
77+ return None # function not found.
78+
7279
7380@dataclass
7481class ComResult :
@@ -78,7 +85,6 @@ class ComResult:
7885 exception : str = ""
7986 ack_packet : list = field (default_factory = list )
8087 rx_packets : list = field (default_factory = list )
81- aux_data : dict = field (default_factory = dict )
8288
8389
8490@dataclass
@@ -91,18 +97,81 @@ class InstructionArguments:
9197 volatility : Persistence = Persistence .NONE
9298
9399
100+ @dataclass (init = False )
101+ class NPCPacket :
102+ """Class contains functions and data for the packet based communication."""
103+
104+ to_address : int
105+ from_address : int
106+ payload : tuple [int , ...]
107+ packet : bytes
108+
109+ def __init__ (self , to_address : int , from_address : int , payload : tuple [int , ...]):
110+ """Construct a new packet object."""
111+ self .to_address = to_address
112+ self .from_address = from_address
113+ self .payload = payload
114+ self .packet = self .compute_packet ()
115+
116+ def compute_packet (self ) -> bytes :
117+ """Generate a standardised NPC binary packet."""
118+ if (
119+ self .to_address < 256
120+ and self .from_address < 256
121+ and len (self .payload ) < 256
122+ ): # check input is possible to parse
123+ packet_data = tuple (
124+ [self .to_address , self .from_address , len (self .payload )]
125+ + list (self .payload )
126+ )
127+ lrc = NPCPacket .get_npc_checksum (packet_data )
128+ return bytes (
129+ [0x3E , packet_data [0 ], packet_data [1 ], len (self .payload )]
130+ + list (self .payload )
131+ + [lrc , 0x3C ]
132+ )
133+ return bytes ([])
134+
135+ @staticmethod
136+ def get_npc_checksum (packet_data : tuple [int , ...]) -> int :
137+ """Generate a NPC LRC checksum.
138+
139+ :param packet_data: List of the uint8 values from an NPC packet.
140+ :return: NPC checksum as an 8-bit integer.
141+ """
142+ lrc = 0
143+ for byte in packet_data :
144+ lrc = (lrc + byte ) & 0xFF
145+ return ((lrc ^ 0xFF ) + 1 ) & 0xFF
146+
147+ def expects_ack (self ) -> bool :
148+ """Check if this packet is expected to be acknowledged."""
149+ if function := UOSFunctions .get_from_address (self .to_address ):
150+ return function .ack
151+ raise UOSUnsupportedError (
152+ "When checking `gets_ack`, "
153+ f"function for address { self .to_address } ,
154+ )
155+
156+ def expects_rx_packets (self ) -> list [int ]:
157+ """Check if this packet expects rx packets from the function def."""
158+ if function := UOSFunctions .get_from_address (self .to_address ):
159+ return function .rx_packets_expected
160+ raise UOSUnsupportedError (
161+ "When checking `expects_rx_packets, "
162+ f"function for address { self .to_address }
163+ )
164+
165+
94166class UOSInterface (metaclass = ABCMeta ):
95167 """Base class for low level UOS interfaces classes to inherit."""
96168
97169 # Dead code suppression used as abstract interfaces are false positives.
98170 @abstractmethod
99- def execute_instruction (
100- self , address : int , payload : tuple [int , ...], ** kwargs # dead: disable
101- ) -> ComResult :
171+ def execute_instruction (self , packet : NPCPacket ) -> ComResult : # dead: disable
102172 """Abstract method for executing instructions on UOSInterfaces.
103173
104- :param address: An 8-bit unsigned integer of the UOS subsystem targeted by the instruction.
105- :param payload: A tuple containing the uint8 parameters of the UOS instruction.
174+ :param packet: A tuple containing the uint8 npc packet for the UOS instruction.
106175 :returns: ComResult object.
107176 :raises: UOSUnsupportedError if the interface hasn't been built correctly.
108177 :raises: UOSCommunicationError if there is a problem completing the action.
@@ -184,40 +253,6 @@ def enumerate_devices() -> list:
184253 f"UOSInterfaces must over-ride { UOSInterface .enumerate_devices .__name__ }
185254 )
186255
187- @staticmethod
188- @lru_cache (maxsize = 100 )
189- def get_npc_packet (to_addr : int , from_addr : int , payload : tuple [int , ...]) -> bytes :
190- """Generate a standardised NPC binary packet.
191-
192- :param to_addr: An 8-bit unsigned integer of the UOS subsystem targeted by the instruction.
193- :param from_addr: An 8-bit unsigned integer of the host system, usually 0.
194- :param payload: A tuple containing the unsigned 8-bit integers of the command.
195- :return: NPC packet as a bytes object. No bytes returned on fault.
196- """
197- if (
198- to_addr < 256 and from_addr < 256 and len (payload ) < 256
199- ): # check input is possible to parse
200- packet_data = tuple ([to_addr , from_addr , len (payload )] + list (payload ))
201- lrc = UOSInterface .get_npc_checksum (packet_data )
202- return bytes (
203- [0x3E , packet_data [0 ], packet_data [1 ], len (payload )]
204- + list (payload )
205- + [lrc , 0x3C ]
206- )
207- return bytes ([])
208-
209- @staticmethod
210- def get_npc_checksum (packet_data : tuple [int , ...]) -> int :
211- """Generate a NPC LRC checksum.
212-
213- :param packet_data: List of the uint8 values from an NPC packet.
214- :return: NPC checksum as a 8-bit integer.
215- """
216- lrc = 0
217- for byte in packet_data :
218- lrc = (lrc + byte ) & 0xFF
219- return ((lrc ^ 0xFF ) + 1 ) & 0xFF
220-
221256
222257@dataclass (frozen = True )
223258class Pin :
0 commit comments