Skip to content

Commit 0588379

Browse files
ryanhzryanhz
committed
Added test case for control panel, bugfixes
1 parent 1dc00b3 commit 0588379

7 files changed

Lines changed: 129 additions & 13 deletions

File tree

‎osdp/_bus.py‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
from threading import Lock
66
from uuid import UUID, uuid4
77

8+
from ._types import *
89
from ._connection import OsdpConnection
910
from ._device import Device
1011
from ._message import Message
@@ -31,7 +32,7 @@ def __init__(self, connection: OsdpConnection, on_reply_received):
3132

3233
@property
3334
def idle_line_delay(self) -> timedelta:
34-
return timedelta(milliseconds=(1000.0/self._connection.baud_rate * 16.0))
35+
return timedelta(milliseconds=(1000.0/self._connection.baud_rate * 16.0) * 100)
3536

3637
def close(self):
3738
self._is_shutting_down = True
@@ -41,6 +42,8 @@ def send_command(self, command: Command):
4142
found_device = self._configured_devices.get(command.address)
4243
if found_device is not None:
4344
found_device.send_command(command)
45+
else:
46+
log.warning("Device not found with address %s", command.address)
4447

4548
def add_device(self, address: int, use_crc: bool, use_secure_channel: bool) -> Device:
4649
found_device = self._configured_devices.get(address)
@@ -118,10 +121,11 @@ def process_reply(self, reply: Reply, device: Device):
118121
if reply.type != ReplyType.Busy:
119122
device.valid_reply_has_been_received()
120123

121-
extract_reply_data = reply.extract_reply_data
122-
error_code = ErrorCode(extract_reply_data[0])
123-
if reply.type == ReplyType.Nak and (error_code==ErrorCode.DoesNotSupportSecurityBlock or error_code==ErrorCode.DoesNotSupportSecurityBlock):
124-
device.reset_security()
124+
if reply.type == ReplyType.Nak:
125+
extract_reply_data = reply.extract_reply_data
126+
error_code = ErrorCode(extract_reply_data[0])
127+
if error_code==ErrorCode.DoesNotSupportSecurityBlock or error_code==ErrorCode.DoesNotSupportSecurityBlock:
128+
device.reset_security()
125129

126130
if reply.type==ReplyType.CrypticData:
127131
device.initialize_secure_channel(reply)

‎osdp/_command.py‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
from ._types import *
44
from ._message import Message
5-
from ._device import Device
65
import datetime
76

87
class Command(Message):
@@ -23,7 +22,7 @@ def security_control_block(self) -> bytes:
2322
def custom_command_update(self, command_buffer: bytearray):
2423
pass
2524

26-
def build_command(self, device: Device) -> bytes:
25+
def build_command(self, device) -> bytes:
2726
command_buffer = bytearray([
2827
self.SOM,
2928
self.address,

‎osdp/_control_panel.py‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,14 @@
66

77
from ._types import *
88
from ._connection import OsdpConnection
9-
from ._command import Command
10-
from ._reply import Reply
9+
from ._command import *
10+
from ._reply import *
11+
from ._bus import Bus
12+
1113

1214
log = logging.getLogger('osdp')
15+
console_handler = logging.StreamHandler()
16+
log.addHandler(console_handler)
1317

1418
class ControlPanel:
1519

@@ -79,7 +83,7 @@ def reply_fetcher(reply: Reply):
7983
raise TimeoutError()
8084

8185
def shutdown(self):
82-
for bus in self._buses:
86+
for bus in list(self._buses.values()):
8387
bus.close()
8488

8589
def add_device(self, connection_id: UUID, address: int, use_crc: bool, use_secure_channel: bool):

‎osdp/_device.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
import datetime
33

44
from ._types import *
5+
from ._command import *
56
from ._secure_channel import SecureChannel
67

78

‎osdp/_message.py‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,4 @@
11
from abc import ABC, abstractmethod
2-
from ._device import Device
32

43
class Message(ABC):
54

@@ -67,7 +66,7 @@ def add_checksum(self, packet: bytearray):
6766
checksum = self.calculate_checksum(bytes(packet[:-1]))
6867
packet[-1] = checksum & 0xFF
6968

70-
def encrypted_data(self, device: Device) -> bytes:
69+
def encrypted_data(self, device) -> bytes:
7170
data = self.data()
7271
if len(data)>0:
7372
return device.encrypt_data(data)

‎osdp/_reply.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,7 @@ def secure_cryptogram_has_been_accepted(self) -> bool:
111111
return self.secure_block_data[0]!=0
112112

113113
def match_issuing_command(self, command: Command) -> bool:
114-
return command==self._issuingCommand
114+
return command==self._issuing_command
115115

116116
def is_valid_mac(self, mac: bytes) -> bool:
117117
return mac[:self.MAC_SIZE]==self.mac

‎tests/test_control_panel.py‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
#!/usr/bin/env python
2+
# -*- coding: utf-8 -*-
3+
4+
"""Tests for OSDP Bus"""
5+
6+
import logging
7+
import os
8+
import sys
9+
import time
10+
import unittest
11+
12+
sys.path.insert(0, os.path.abspath('..'))
13+
from osdp import *
14+
15+
log = logging.getLogger('osdp')
16+
17+
class ControlPanelTestCase(unittest.TestCase):
18+
19+
"""Test Bus for OSDP Python Module."""
20+
21+
def setUp(self):
22+
"""Setup."""
23+
self.last_reply = None
24+
25+
def tearDown(self):
26+
"""Teardown."""
27+
28+
def test_cp_checksum_unsecure(self):
29+
conn = SerialPortOsdpConnection(port='/dev/tty.wchusbserial1420', baud_rate=9600)
30+
cp = ControlPanel()
31+
bus_id = cp.start_connection(conn)
32+
self.assertIsNotNone(bus_id)
33+
34+
cp.add_device(connection_id=bus_id, address=0x7F, use_crc=False, use_secure_channel=False)
35+
36+
id_report = cp.id_report(connection_id=bus_id, address=0x7F)
37+
print("\r\n")
38+
print(id_report)
39+
40+
device_capabilities = cp.device_capabilities(connection_id=bus_id, address=0x7F)
41+
print("\r\n")
42+
print(device_capabilities)
43+
44+
local_status = cp.local_status(connection_id=bus_id, address=0x7F)
45+
print("\r\n")
46+
print(local_status)
47+
48+
input_status = cp.input_status(connection_id=bus_id, address=0x7F)
49+
print("\r\n")
50+
print(input_status)
51+
52+
output_status = cp.output_status(connection_id=bus_id, address=0x7F)
53+
print("\r\n")
54+
print(output_status)
55+
56+
reader_status = cp.reader_status(connection_id=bus_id, address=0x7F)
57+
print("\r\n")
58+
print(reader_status)
59+
60+
output_status = cp.output_status(connection_id=bus_id, address=0x7F)
61+
print("\r\n")
62+
print(output_status)
63+
64+
granted_led = [ReaderLedControl(
65+
reader_number = 0x0,
66+
led_number = 0x0,
67+
temporary_mode = TemporaryReaderControlCode.SetTemporaryAndStartTimer,
68+
temporary_on_time = 0x02,
69+
temporary_off_time = 0x01,
70+
temporary_on_color = LedColor.Green,
71+
temporary_off_color = LedColor.Black,
72+
temporary_timer = 0x000A,
73+
permanent_mode = PermanentReaderControlCode.Nop,
74+
permanent_on_time = 0x00,
75+
permanent_off_time = 0x00,
76+
permanent_on_color = LedColor.Black,
77+
permanent_off_color = LedColor.Black
78+
)]
79+
result = cp.reader_led_control(connection_id=bus_id, address=0x7F, reader_led_controls=ReaderLedControls(granted_led))
80+
print("\r\n")
81+
print(result)
82+
83+
time.sleep(1.0)
84+
85+
denied_led = [ReaderLedControl(
86+
reader_number = 0x0,
87+
led_number = 0x0,
88+
temporary_mode = TemporaryReaderControlCode.SetTemporaryAndStartTimer,
89+
temporary_on_time = 0x02,
90+
temporary_off_time = 0x01,
91+
temporary_on_color = LedColor.Red,
92+
temporary_off_color = LedColor.Black,
93+
temporary_timer = 0x000A,
94+
permanent_mode = PermanentReaderControlCode.Nop,
95+
permanent_on_time = 0x00,
96+
permanent_off_time = 0x00,
97+
permanent_on_color = LedColor.Black,
98+
permanent_off_color = LedColor.Black
99+
)]
100+
101+
result = cp.reader_led_control(connection_id=bus_id, address=0x7F, reader_led_controls=ReaderLedControls(denied_led))
102+
print("\r\n")
103+
print(result)
104+
105+
cp.shutdown()
106+
107+
108+
if __name__ == '__main__':
109+
unittest.main()

0 commit comments

Comments
 (0)