Add support for PD online status

python/pyosdp_pd.c

@@ -8,6 +8,22 @@
 
 #define TAG "pyosdp_pd"
 
+#define pyosdp_pd_is_online_doc                                                \
+	"Get PD status, (online/offline)\n"                                    \
+	"\n"                                                                   \
+	"@return PD online status (Bool)"
+static PyObject *pyosdp_pd_is_online(pyosdp_pd_t *self, PyObject *args)
+{
+	uint64_t mask;
+
+	osdp_get_status_mask(self->ctx, (uint8_t *)&mask);
+
+	if (mask & 1)
+		Py_RETURN_TRUE;
+	else
+		Py_RETURN_FALSE;
+}
+
 #define pyosdp_pd_is_sc_active_doc                                             \
 	"Get Secure Channel status, (active/inactive)\n"                       \
 	"\n"                                                                   \
@@ -351,6 +367,8 @@ static PyMethodDef pyosdp_pd_tp_methods[] = {
 	  METH_VARARGS, pyosdp_pd_notify_event_doc },
 	{ "is_sc_active", (PyCFunction)pyosdp_pd_is_sc_active,
 	  METH_NOARGS, pyosdp_pd_is_sc_active_doc },
+	{ "is_online", (PyCFunction)pyosdp_pd_is_online,
+	  METH_NOARGS, pyosdp_pd_is_online_doc },
 	{ "flush_events", (PyCFunction)pyosdp_pd_flush_events,
 	  METH_VARARGS, pyosdp_pd_flush_events_doc },
 	{ NULL, NULL, 0, NULL } /* Sentinel */

python/setup.py.in

@@ -33,6 +33,7 @@ sources = utils_sources + pyosdp_sources
 
 libosdp_include_dirs = ('@PROJECT_SOURCE_DIR@/utils/include'.split(';') +
                         '@PROJECT_SOURCE_DIR@/include'.split(';') +
+                        '@LIB_OSDP_INCLUDE_DIRS@'.split(';') +
                         '@LIB_OSDP_PRIVATE_INCLUDE_DIRS@'.split(';'))
 
 libosdp_includes = [ '-I' + path for path in libosdp_include_dirs ]

src/osdp_common.c

@@ -255,7 +255,12 @@ void osdp_get_status_mask(osdp_t *ctx, uint8_t *bitmask)
 	input_check(ctx);
 	int i, pos;
 	uint8_t *mask = bitmask;
-	struct osdp_pd *pd;
+	struct osdp_pd *pd = osdp_to_pd(ctx, 0);
+
+	if (ISSET_FLAG(pd, PD_FLAG_PD_MODE)) {
+		*mask = osdp_millis_since(pd->tstamp) < OSDP_RESP_TOUT_MS;
+		return;
+	}
 
 	*mask = 0;
 	for (i = 0; i < NUM_PD(ctx); i++) {
@@ -265,8 +270,7 @@ void osdp_get_status_mask(osdp_t *ctx, uint8_t *bitmask)
 			*mask = 0;
 		}
 		pd = osdp_to_pd(ctx, i);
-		if (ISSET_FLAG(pd, PD_FLAG_PD_MODE) ||
-		    pd->state == OSDP_CP_STATE_ONLINE) {
+		if (pd->state == OSDP_CP_STATE_ONLINE) {
 			*mask |= 1 << pos;
 		}
 	}

tests/pytest/README.md

@@ -0,0 +1,8 @@
+# Run Tests
+
+`make check` on cmake builds will invoke pytest correctly. During development,
+it might be useful to run an individual test (instead of everything). To do so,
+
+```
+PYTHONPATH=../../build/python/  python3 -m pytest -vv -s test_events.py::test_event_input
+```

tests/pytest/test_status.py

@@ -0,0 +1,68 @@
+#
+#  Copyright (c) 2023 Siddharth Chandrasekaran <[email protected]>
+#
+#  SPDX-License-Identifier: Apache-2.0
+#
+
+import time
+import pytest
+
+from testlib import *
+
+pd_cap = PDCapabilities([
+    (Capability.OutputControl, 1, 1),
+    (Capability.LEDControl, 1, 1),
+    (Capability.AudibleControl, 1, 1),
+    (Capability.TextOutput, 1, 1),
+])
+
+pd_info = [
+    PDInfo(101, scbk=KeyStore.gen_key(), name='chn-0'),
+]
+
+# TODO remove this.
+pd_addr = pd_info[0].address
+pd = PeripheralDevice(pd_info[0], pd_cap, log_level=LogLevel.Debug)
+cp = ControlPanel(pd_info)
+
+@pytest.fixture(scope='module', autouse=True)
+def setup_test():
+    pd.start()
+    cp.start()
+    cp.sc_wait_all()
+    yield
+    teardown_test()
+
+def teardown_test():
+    cp.teardown()
+    pd.teardown()
+
+def test_cp_status():
+    assert cp.online_wait(pd.address)
+    pd.stop()
+    assert cp.online_wait(pd.address) == False
+    pd.start()
+    assert cp.online_wait(pd.address)
+
+def test_cp_sc_status():
+    assert cp.sc_wait(pd.address)
+    pd.stop()
+    assert cp.sc_wait(pd.address) == False
+    pd.start()
+    assert cp.sc_wait(pd.address)
+
+def test_pd_status():
+    cp.stop()
+    time.sleep(1)
+    assert pd.is_online() == False
+    cp.start()
+    assert cp.sc_wait(pd.address)
+    assert pd.is_online()
+
+def test_pd_sc_status():
+    cp.stop()
+    time.sleep(1)
+    assert pd.is_sc_active() == False
+    cp.start()
+    assert cp.sc_wait(pd.address)
+    assert pd.is_sc_active()

tests/pytest/testlib/control_panel.py

@@ -30,10 +30,9 @@ def __init__(self, pd_info_list, log_level: LogLevel=LogLevel.Info,
             self.ctx = osdp.ControlPanel(info_list)
         self.ctx.set_event_callback(self.event_handler)
         self.ctx.set_loglevel(log_level)
-        self.event = threading.Event()
-        self.lock = threading.Lock()
-        args = (self.event, self.lock, self.ctx,)
-        self.thread = threading.Thread(name='cp', target=self.refresh, args=args)
+        self.event = None
+        self.lock = None
+        self.thread = None
 
     @staticmethod
     def refresh(event, lock, ctx):
@@ -126,20 +125,26 @@ def get_file_tx_status(self, address):
         return ret
 
     def start(self):
-        if not self.thread:
-            return False
+        if self.thread:
+            raise RuntimeError("Thread already running!")
+        self.event = threading.Event()
+        self.lock = threading.Lock()
+        args=(self.event, self.lock, self.ctx)
+        self.thread = threading.Thread(name='cp', target=self.refresh, args=args)
         self.thread.start()
 
     def stop(self):
-        while self.thread and self.thread.is_alive():
+        if not self.thread:
+            raise RuntimeError("Thread not running!")
+        while self.thread.is_alive():
             self.event.set()
             self.thread.join(2)
             if not self.thread.is_alive():
-                return True
-        return False
+                self.thread = None
+                break
 
     def online_wait_all(self, timeout=10):
-        count = 10
+        count = 0
         res = False
         while count < timeout * 2:
             time.sleep(0.5)
@@ -150,7 +155,7 @@ def online_wait_all(self, timeout=10):
         return res
 
     def online_wait(self, address, timeout=5):
-        count = 10
+        count = 0
         res = False
         while count < timeout * 2:
             time.sleep(0.5)

tests/pytest/testlib/peripheral_device.py

@@ -20,10 +20,9 @@ def __init__(self, pd_info: PDInfo, pd_cap: PDCapabilities,
         self.ctx = osdp.PeripheralDevice(pd_info.get(), capabilities=pd_cap.get())
         self.ctx.set_loglevel(log_level)
         self.ctx.set_command_callback(self.command_handler)
-        self.event = threading.Event()
-        self.lock = threading.Lock()
-        args = (self.event, self.lock, self.ctx,)
-        self.thread = threading.Thread(name='pd', target=self.refresh, args=args)
+        self.event = None
+        self.lock = None
+        self.thread = None
 
     @staticmethod
     def refresh(event, lock, ctx):
@@ -59,19 +58,27 @@ def register_file_ops(self, fops):
     def is_sc_active(self):
         return self.ctx.is_sc_active()
 
+    def is_online(self):
+        return self.ctx.is_online()
+
     def start(self):
         if self.thread:
-            self.thread.start()
-            return True
-        return False
+            raise RuntimeError("Thread already running!")
+        self.event = threading.Event()
+        self.lock = threading.Lock()
+        args = (self.event, self.lock, self.ctx,)
+        self.thread = threading.Thread(name='pd', target=self.refresh, args=args)
+        self.thread.start()
 
     def stop(self):
-        while self.thread and self.thread.is_alive():
+        if not self.thread:
+            raise RuntimeError("Thread not running!")
+        while self.thread.is_alive():
             self.event.set()
             self.thread.join(2)
             if not self.thread.is_alive():
-                return True
-        return False
+                self.thread = None
+                break
 
     def teardown(self):
         self.stop()