#631999863

Committed 08 Apr 2025 07:22AM UTC coverage: 67.546% (-0.005%) from 67.551%

Build # #631999863

Build Type

Pull #302

travis-ci

Pull Request Pull Request #302: Fix `tox -e mypy` errors with `mypy 1.14.1`

Run Details

22 of 66 new or added lines in 15 files covered. (33.33%)

10 existing lines in 1 file now uncovered.

4712 of 6976 relevant lines covered (67.55%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.03

/nixnet/convert.py

import itertools
import typing  # NOQA: F401
import warnings

from nixnet import _frames
from nixnet import _funcs
from nixnet import _props
from nixnet import _utils
from nixnet import constants
from nixnet import errors
from nixnet import types

from nixnet._session import j1939 as session_j1939
from nixnet._session import signals as session_signals


__all__ = [
    "SignalConversionSinglePointSession"]


class SignalConversionSinglePointSession(object):
    """Convert NI-XNET signal data to frame data or vice versa.

    Conversion works similar to Single-Point mode. You specify a set of signals
    that can span multiple frames. Signal to frame conversion reads a set of
    values for the signals specified and writes them to the respective
    frame(s). Frame to signal conversion parses a set of frames and returns the
    latest signal value read from a corresponding frame.
    """

    def __init__(
            self,
            database_name,  # type: typing.Text
            cluster_name,  # type: typing.Text
            signals,  # type: typing.Union[typing.Text, typing.List[typing.Text]]
    ):
        # type: (...) -> None
        """Create an XNET session at run time using named references to database objects.

        Args:
            database_name(str): XNET database name to use for
                interface configuration. The database name must use the <alias>
                or <filepath> syntax (refer to Databases).
            cluster_name(str): XNET cluster name to use for
                interface configuration. The name must specify a cluster from
                the database given in the database_name parameter. If it is left
                blank, the cluster is extracted from the ``signals`` parameter.
            signals(list of str): Strings describing signals for the session. The
                list syntax is as follows:

                ``signals`` contains one or more XNET Signal names. Each name must
                be one of the following options, whichever uniquely
                identifies a signal within the database given:

                    - ``<Signal>``
                    - ``<Frame>.<Signal>``
                    - ``<Cluster>.<Frame>.<Signal>``
                    - ``<PDU>.<Signal>``
                    - ``<Cluster>.<PDU>.<Signal>``

                ``signals`` may also contain one or more trigger signals. For
                information about trigger signals, refer to Signal Output
                Single-Point Mode or Signal Input Single-Point Mode.
        """
        flattened_list = _utils.flatten_items(signals)

        self._handle = None  # To satisfy `__del__` in case nx_create_session throws
        self._handle = _funcs.nx_create_session(
            database_name,
            cluster_name,
            flattened_list,
            "",
            constants.CreateSessionMode.SIGNAL_CONVERSION_SINGLE_POINT)
        self._j1939 = session_j1939.J1939(self._handle)
        self._signals = session_signals.Signals(self._handle)

    def __del__(self):
        if self._handle is not None:
            warnings.warn(
                'Session was not explicitly closed before it was destructed. '
                'Resources on the device may still be reserved.',
                errors.XnetResourceWarning)

    def __enter__(self):
        return self

    def __exit__(self, exception_type, exception_value, traceback):
        self.close()

    def __eq__(self, other):
        if isinstance(other, self.__class__):
            return self._handle == typing.cast(SignalConversionSinglePointSession, other)._handle
        else:
            return NotImplemented

    def __ne__(self, other):
        result = self.__eq__(other)
        if result is NotImplemented:
            return result
        else:
            return not result

    def __hash__(self):
        return hash(self._handle)

    def __repr__(self):
        # type: () -> typing.Text
        return '{}(handle={})'.format(type(self).__name__, self._handle)

    def close(self):
        # type: () -> None
        """Close (clear) the XNET session."""
        if self._handle is None:
            warnings.warn(
                'Attempting to close NI-XNET session but session was already '
                'closed', errors.XnetResourceWarning)
            return

        _funcs.nx_clear(self._handle)

        self._handle = None

    @property
    def signals(self):
        # type: () -> session_signals.Signals
        """:any:`nixnet._session.signals.Signals`: Operate on session's signals"""
        return self._signals

    @property
    def j1939(self):
        # type: () -> session_j1939.J1939
        """:any:`nixnet._session.j1939.J1939`: Returns the J1939 configuration object for the session."""
        return self._j1939

    @property
    def application_protocol(self):
        # type: () -> constants.AppProtocol
        """:any:`nixnet._enums.AppProtocol`: This property returns the application protocol that the session uses.

        The database used with the session determines the application protocol.
        """
        return constants.AppProtocol(
            _props.get_session_application_protocol(self._handle))  # type: ignore

    @property
    def cluster_name(self):
        # type: () -> typing.Text
        """str: This property returns the cluster (network) name used with the session."""
        return _props.get_session_cluster_name(self._handle)  # type: ignore

    @property
    def database_name(self):
        # type: () -> typing.Text
        """str: This property returns the database name used with the session."""
        return _props.get_session_database_name(self._handle)  # type: ignore

    @property
    def mode(self):
        # type: () -> constants.CreateSessionMode
        """:any:`nixnet._enums.CreateSessionMode`: This property returns the mode associated with the session.

        For more information, refer to :any:`nixnet._enums.CreateSessionMode`.
        """
        return constants.CreateSessionMode(
            _props.get_session_mode(self._handle))  # type: ignore

    @property
    def protocol(self):
        # type: () -> constants.Protocol
        """:any:`nixnet._enums.Protocol`: This property returns the protocol that the interface in the session uses."""
        return constants.Protocol(
            _props.get_session_protocol(self._handle))  # type: ignore

    def _convert_bytes_to_signals(self, bytes):
        # type: (bytes) -> typing.Iterable[typing.Tuple[int, float]]
        num_signals = len(self.signals)
        timestamps, values = _funcs.nx_convert_frames_to_signals_single_point(
            self._handle, bytes, num_signals)  # type: ignore
        for timestamp, value in zip(timestamps, values):
            yield timestamp.value, value.value

    def convert_frames_to_signals(self, frames):
        # type: (typing.Iterable[types.Frame]) -> typing.Iterable[typing.Tuple[int, float]]
        """Convert Frames to signals.

        The frames passed into the ``frames`` array are read one by one, and
        the signal values found are written to internal buffers for each
        signal. Frames are identified by their identifier (FlexRay: slot)
        field. After all frames in ``frames`` array are processed, the internal
        signal buffers' status is returned with the corresponding timestamps
        from the frames where a signal value was found. The signal internal
        buffers' status is being preserved over multiple calls to this
        function.

        This way, for example, data returned from multiple calls of nxFrameRead
        for a Frame Input Stream Mode session (or any other Frame Input
        session) can be passed to this function directly.

        .. note:: Frames unknown to the session are silently ignored.
        """
        units = itertools.chain.from_iterable(
            _frames.serialize_frame(frame.to_raw())
            for frame in frames)
        bytes = b"".join(units)
        return self._convert_bytes_to_signals(bytes)

    def _convert_signals_to_bytes(self, signals, num_bytes):
        # type: (typing.Iterable[float], int) -> bytes
        buffer, number_of_bytes_returned = _funcs.nx_convert_signals_to_frames_single_point(
            self._handle,  # type: ignore
            list(signals),
            num_bytes)
        return buffer[0:number_of_bytes_returned]

    def convert_signals_to_frames(self, signals, frame_type=types.XnetFrame):
        # type: (typing.Iterable[float], typing.Type[types.FrameFactory]) -> typing.Iterable[types.Frame]
        """Convert signals to frames.

        The signal values written to the ``signals`` array are written to a raw
        frame buffer array. For each frame included in the session, one frame
        is generated in the array that contains the signal values. Signals not
        present in the session are written as their respective default values;
        empty space in the frames that signals do not occupy is written with
        the frame's default payload.

        The frame header values are filled with appropriate values so that this
        function's output can be directly written to a Frame Output session.

        Args:
            signals(list of float): Values corresponding to signals configured
                in this session.
            frame_type(:any:`nixnet.types.FrameFactory`): A factory for the
                desired frame formats.

        Yields:
            :any:`nixnet.types.Frame`
        """
        from_raw = typing.cast(typing.Callable[[types.RawFrame], types.Frame], frame_type.from_raw)
        # Unlike some session reads, this should be safe from asking to read too much.
        num_frames_to_read = 5
        while True:
            try:
                num_bytes_to_read = num_frames_to_read * _frames.nxFrameFixed_t.size
                buffer = self._convert_signals_to_bytes(signals, num_bytes_to_read)
                break
            except errors.XnetError as e:
                if e.error_type == constants.Err.BUFFER_TOO_SMALL:
                    num_frames_to_read *= 2
                else:
                    raise
        for frame in _frames.iterate_frames(buffer):
            yield from_raw(frame)

1	import itertools	1✔
2	import typing # NOQA: F401	1✔
3	import warnings	1✔
4
5	from nixnet import _frames	1✔
6	from nixnet import _funcs	1✔
7	from nixnet import _props	1✔
8	from nixnet import _utils
9	from nixnet import constants	1✔
10	from nixnet import errors	1✔
11	from nixnet import types	1✔
12		1✔
13	from nixnet._session import j1939 as session_j1939	1✔
14	from nixnet._session import signals as session_signals	1✔
15		1✔
16
17	__all__ = [	1✔
18	"SignalConversionSinglePointSession"]	1✔
19
20
21	class SignalConversionSinglePointSession(object):	1✔
22	"""Convert NI-XNET signal data to frame data or vice versa.
23
24	Conversion works similar to Single-Point mode. You specify a set of signals
25	that can span multiple frames. Signal to frame conversion reads a set of	1✔
26	values for the signals specified and writes them to the respective
27	frame(s). Frame to signal conversion parses a set of frames and returns the
28	latest signal value read from a corresponding frame.
29	"""
30
31	def __init__(
32	self,
33	database_name, # type: typing.Text
34	cluster_name, # type: typing.Text
35	signals, # type: typing.Union[typing.Text, typing.List[typing.Text]]	1✔
36	):
37	# type: (...) -> None
38	"""Create an XNET session at run time using named references to database objects.
39
40	Args:
41	database_name(str): XNET database name to use for
42	interface configuration. The database name must use the <alias>
43	or <filepath> syntax (refer to Databases).
44	cluster_name(str): XNET cluster name to use for
45	interface configuration. The name must specify a cluster from
46	the database given in the database_name parameter. If it is left
47	blank, the cluster is extracted from the ``signals`` parameter.
48	signals(list of str): Strings describing signals for the session. The
49	list syntax is as follows:
50
51	``signals`` contains one or more XNET Signal names. Each name must
52	be one of the following options, whichever uniquely
53	identifies a signal within the database given:
54
55	- ``<Signal>``
56	- ``<Frame>.<Signal>``
57	- ``<Cluster>.<Frame>.<Signal>``
58	- ``<PDU>.<Signal>``
59	- ``<Cluster>.<PDU>.<Signal>``
60
61	``signals`` may also contain one or more trigger signals. For
62	information about trigger signals, refer to Signal Output
63	Single-Point Mode or Signal Input Single-Point Mode.
64	"""
65	flattened_list = _utils.flatten_items(signals)
66
67	self._handle = None # To satisfy `__del__` in case nx_create_session throws
68	self._handle = _funcs.nx_create_session(
69	database_name,	1✔
70	cluster_name,
71	flattened_list,	1✔
72	"",	1✔
73	constants.CreateSessionMode.SIGNAL_CONVERSION_SINGLE_POINT)
74	self._j1939 = session_j1939.J1939(self._handle)
75	self._signals = session_signals.Signals(self._handle)
76
77	def __del__(self):
78	if self._handle is not None:	1✔
79	warnings.warn(	1✔
80	'Session was not explicitly closed before it was destructed. '
81	'Resources on the device may still be reserved.',	1✔
82	errors.XnetResourceWarning)	1✔
83		×
84	def __enter__(self):
85	return self
86
87	def __exit__(self, exception_type, exception_value, traceback):
88	self.close()	1✔
89		1✔
90	def __eq__(self, other):
91	if isinstance(other, self.__class__):	1✔
92	return self._handle == typing.cast(SignalConversionSinglePointSession, other)._handle	1✔
93	else:
94	return NotImplemented	1✔
95		×
96	def __ne__(self, other):	×
97	result = self.__eq__(other)
98	if result is NotImplemented:	×
99	return result
100	else:	1✔
101	return not result	×
102		×
103	def __hash__(self):	×
104	return hash(self._handle)
105		×
106	def __repr__(self):
107	# type: () -> typing.Text	1✔
108	return '{}(handle={})'.format(type(self).__name__, self._handle)	×
109
110	def close(self):	1✔
111	# type: () -> None
112	"""Close (clear) the XNET session."""	×
113	if self._handle is None:
114	warnings.warn(	1✔
115	'Attempting to close NI-XNET session but session was already '
116	'closed', errors.XnetResourceWarning)
117	return	1✔
118		×
119	_funcs.nx_clear(self._handle)
120
121	self._handle = None	×
122
123	@property	1✔
124	def signals(self):
125	# type: () -> session_signals.Signals	1✔
126	""":any:`nixnet._session.signals.Signals`: Operate on session's signals"""
127	return self._signals	1✔
128
129	@property
130	def j1939(self):
131	# type: () -> session_j1939.J1939	1✔
132	""":any:`nixnet._session.j1939.J1939`: Returns the J1939 configuration object for the session."""
133	return self._j1939	1✔
134
135	@property
136	def application_protocol(self):
137	# type: () -> constants.AppProtocol	×
138	""":any:`nixnet._enums.AppProtocol`: This property returns the application protocol that the session uses.
139		1✔
140	The database used with the session determines the application protocol.
141	"""
142	return constants.AppProtocol(
143	_props.get_session_application_protocol(self._handle)) # type: ignore
144
145	@property
NEW 146	def cluster_name(self):	×
147	# type: () -> typing.Text
148	"""str: This property returns the cluster (network) name used with the session."""
149	return _props.get_session_cluster_name(self._handle) # type: ignore	1✔
150
151	@property
152	def database_name(self):
NEW 153	# type: () -> typing.Text	×
154	"""str: This property returns the database name used with the session."""
155	return _props.get_session_database_name(self._handle) # type: ignore	1✔
156
157	@property
158	def mode(self):
NEW 159	# type: () -> constants.CreateSessionMode	×
160	""":any:`nixnet._enums.CreateSessionMode`: This property returns the mode associated with the session.
161		1✔
162	For more information, refer to :any:`nixnet._enums.CreateSessionMode`.
163	"""
164	return constants.CreateSessionMode(
165	_props.get_session_mode(self._handle)) # type: ignore
166
167	@property
NEW 168	def protocol(self):	×
169	# type: () -> constants.Protocol
170	""":any:`nixnet._enums.Protocol`: This property returns the protocol that the interface in the session uses."""
171	return constants.Protocol(	1✔
172	_props.get_session_protocol(self._handle)) # type: ignore
173
174	def _convert_bytes_to_signals(self, bytes):
NEW 175	# type: (bytes) -> typing.Iterable[typing.Tuple[int, float]]	×
176	num_signals = len(self.signals)
177	timestamps, values = _funcs.nx_convert_frames_to_signals_single_point(
178	self._handle, bytes, num_signals) # type: ignore	1✔
179	for timestamp, value in zip(timestamps, values):
180	yield timestamp.value, value.value	1✔
181		1✔
182	def convert_frames_to_signals(self, frames):
183	# type: (typing.Iterable[types.Frame]) -> typing.Iterable[typing.Tuple[int, float]]	1✔
184	"""Convert Frames to signals.	×
185
186	The frames passed into the ``frames`` array are read one by one, and	1✔
187	the signal values found are written to internal buffers for each
188	signal. Frames are identified by their identifier (FlexRay: slot)
189	field. After all frames in ``frames`` array are processed, the internal
190	signal buffers' status is returned with the corresponding timestamps
191	from the frames where a signal value was found. The signal internal
192	buffers' status is being preserved over multiple calls to this
193	function.
194
195	This way, for example, data returned from multiple calls of nxFrameRead
196	for a Frame Input Stream Mode session (or any other Frame Input
197	session) can be passed to this function directly.
198
199	.. note:: Frames unknown to the session are silently ignored.
200	"""
201	units = itertools.chain.from_iterable(
202	_frames.serialize_frame(frame.to_raw())
203	for frame in frames)
204	bytes = b"".join(units)
205	return self._convert_bytes_to_signals(bytes)	1✔
206
207	def _convert_signals_to_bytes(self, signals, num_bytes):
208	# type: (typing.Iterable[float], int) -> bytes	1✔
209	buffer, number_of_bytes_returned = _funcs.nx_convert_signals_to_frames_single_point(	1✔
210	self._handle, # type: ignore
211	list(signals),	1✔
212	num_bytes)
213	return buffer[0:number_of_bytes_returned]	1✔
214
215	def convert_signals_to_frames(self, signals, frame_type=types.XnetFrame):
216	# type: (typing.Iterable[float], typing.Type[types.FrameFactory]) -> typing.Iterable[types.Frame]
217	"""Convert signals to frames.	1✔
218
219	The signal values written to the ``signals`` array are written to a raw	1✔
220	frame buffer array. For each frame included in the session, one frame
221	is generated in the array that contains the signal values. Signals not
222	present in the session are written as their respective default values;
223	empty space in the frames that signals do not occupy is written with
224	the frame's default payload.
225
226	The frame header values are filled with appropriate values so that this
227	function's output can be directly written to a Frame Output session.
228
229	Args:
230	signals(list of float): Values corresponding to signals configured
231	in this session.
232	frame_type(:any:`nixnet.types.FrameFactory`): A factory for the
233	desired frame formats.
234
235	Yields:
236	:any:`nixnet.types.Frame`
237	"""
238	from_raw = typing.cast(typing.Callable[[types.RawFrame], types.Frame], frame_type.from_raw)
239	# Unlike some session reads, this should be safe from asking to read too much.
240	num_frames_to_read = 5
241	while True:
242	try:	1✔
243	num_bytes_to_read = num_frames_to_read * _frames.nxFrameFixed_t.size
244	buffer = self._convert_signals_to_bytes(signals, num_bytes_to_read)	1✔
245	break	1✔
246	except errors.XnetError as e:	1✔
247	if e.error_type == constants.Err.BUFFER_TOO_SMALL:	1✔
248	num_frames_to_read *= 2	1✔
249	else:	1✔
250	raise	×
251	for frame in _frames.iterate_frames(buffer):	×
252	yield from_raw(frame)	×

ni / nixnet-python / #631999863

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous