Merge pull request #2 from fglvs/add_mfg_command

Add manufacturer-specific command

osdp/__init__.py

@@ -18,7 +18,7 @@
 from ._connection import OsdpConnection, SerialPortOsdpConnection, TcpClientOsdpConnection, TcpServerOsdpConnection
 from ._device import Device
 from ._message import Message
-from ._command import Command, PollCommand, IdReportCommand, DeviceCapabilitiesCommand, LocalStatusReportCommand, InputStatusReportCommand, OutputStatusReportCommand, ReaderStatusReportCommand, OutputControlCommand, ReaderLedControlCommand, ReaderBuzzerControlCommand, ReaderTextOutputCommand, SetDateTimeCommand, SecurityInitializationRequestCommand, ServerCryptogramCommand
+from ._command import Command, PollCommand, IdReportCommand, DeviceCapabilitiesCommand, LocalStatusReportCommand, InputStatusReportCommand, OutputStatusReportCommand, ReaderStatusReportCommand, OutputControlCommand, ReaderLedControlCommand, ReaderBuzzerControlCommand, ReaderTextOutputCommand, SetDateTimeCommand, SecurityInitializationRequestCommand, ServerCryptogramCommand, ManufacturerSpecificCommand
 from ._reply import Reply, AckReply, UnknownReply
 from ._secure_channel import SecureChannel
 from ._bus import Bus

osdp/_command.py

@@ -293,6 +293,25 @@ def data(self) -> bytes:
 	def custom_command_update(self, command_buffer: bytearray):
 		pass
 
+class ManufacturerSpecificCommand(Command):
+
+	def __init__(self, address: int, manufacturer_data: bytes):
+		self.address = address
+		self.manufacturer_data = manufacturer_data
+
+	@property
+	def command_code(self) -> int:
+		return 0x80
+
+	def security_control_block(self) -> bytes:
+		return bytes([ 0x02, 0x17 ])
+
+	def data(self) -> bytes:
+		return self.manufacturer_data
+
+	def custom_command_update(self, command_buffer: bytearray):
+		pass
+
 class SecurityInitializationRequestCommand(Command):
 
 	def __init__(self, address: int, server_random_number: bytes):

tests/test_command.py

@@ -236,6 +236,23 @@ def test_set_date_time_command_checksum(self):
 		content = command.build_command(device)
 		self.assertEqual(content.hex().upper(), '537F0E00036DE3070B0C090200A4')
 
+	def test_reader_mfg_command_checksum(self):
+		device = Device(address=0x7F, use_crc=False, use_secure_channel=False)
+
+		device.message_control.increment_sequence()
+		self.assertEqual(device.message_control.sequence, 1)
+
+		manufacturer_data = bytes([0x0B, 0x0E, 0x0E, 0x0F])
+		command = ManufacturerSpecificCommand(address=0x7F, manufacturer_data = manufacturer_data)
+		content = command.build_command(device)
+		self.assertEqual(content.hex().upper(), '537F0B0001800B0E0E0F6C')
+
+		device.message_control.increment_sequence()
+		self.assertEqual(device.message_control.sequence, 2)
+
+		device.message_control.increment_sequence()
+		self.assertEqual(device.message_control.sequence, 3)
+
 	def test_poll_command_crc(self):
 		device = Device(address=0x7F, use_crc=True, use_secure_channel=False)
 
@@ -454,6 +471,22 @@ def test_set_date_time_command_crc(self):
 		device.message_control.increment_sequence()
 		self.assertEqual(device.message_control.sequence, 3)
 
+	def test_reader_mfg_command_crc(self):
+		device = Device(address=0x7F, use_crc=True, use_secure_channel=False)
+
+		device.message_control.increment_sequence()
+		self.assertEqual(device.message_control.sequence, 1)
+
+		device.message_control.increment_sequence()
+		self.assertEqual(device.message_control.sequence, 2)
+
+		device.message_control.increment_sequence()
+		self.assertEqual(device.message_control.sequence, 3)
+
+		manufacturer_data = bytes([0x0B, 0x0E, 0x0E, 0x0F])
+		command = ManufacturerSpecificCommand(address=0x7F, manufacturer_data = manufacturer_data)
+		content = command.build_command(device)
+		self.assertEqual(content.hex().upper(), '537F0C0007800B0E0E0FEDCC')
 
 if __name__ == '__main__':
 	unittest.main()