diff --git a/blinkpy/api.py b/blinkpy/api.py
index 24f2b9f8..a0e77140 100644
--- a/blinkpy/api.py
+++ b/blinkpy/api.py
@@ -218,6 +218,50 @@ async def request_syncmodule(blink, network):
     return await http_get(blink, url)
 
 
+async def request_programs(blink, network):
+    """
+    Request all programs (schedules) for a network.
+
+    :param blink: Blink instance.
+    :param network: Sync module network id.
+    """
+    url = (
+        f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}"
+        f"/networks/{network}/programs"
+    )
+    return await http_get(blink, url)
+
+
+async def request_program_enable(blink, network, program_id):
+    """
+    Enable a program (schedule) on a network.
+
+    :param blink: Blink instance.
+    :param network: Sync module network id.
+    :param program_id: ID of the program to enable.
+    """
+    url = (
+        f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}"
+        f"/networks/{network}/programs/{program_id}/enable"
+    )
+    return await http_post(blink, url)
+
+
+async def request_program_disable(blink, network, program_id):
+    """
+    Disable a program (schedule) on a network.
+
+    :param blink: Blink instance.
+    :param network: Sync module network id.
+    :param program_id: ID of the program to disable.
+    """
+    url = (
+        f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}"
+        f"/networks/{network}/programs/{program_id}/disable"
+    )
+    return await http_post(blink, url)
+
+
 @Throttle(seconds=MIN_THROTTLE_TIME)
 async def request_system_arm(blink, network, **kwargs):
     """
diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py
index 63b6aef6..afdcd32f 100644
--- a/blinkpy/sync_module.py
+++ b/blinkpy/sync_module.py
@@ -111,6 +111,34 @@ def arm(self):
             self.available = False
             return None
 
+    @property
+    def scheduler_enabled(self):
+        """Return boolean status of schedule."""
+        if self.network_info is None:
+            return False
+        try:
+            network = self.network_info.get("network", {})
+            if network.get("active_program_id") is not None or network.get(
+                "autoarm_time_enable", False
+            ):
+                return True
+            # Fallback: check the programs list cached during get_network_info.
+            # If the key is absent the network info has not yet been fully
+            # populated - log a debug message so callers are aware.
+            if "programs" not in self.network_info:
+                _LOGGER.debug(
+                    "Programs list not yet populated for network %s; "
+                    "scheduler_enabled may be inaccurate.",
+                    self.name,
+                )
+                return False
+            for program in self.network_info["programs"]:
+                if program.get("status") == "enabled":
+                    return True
+            return False
+        except (AttributeError, KeyError, TypeError):
+            return False
+
     @property
     def local_storage(self):
         """Indicate if local storage is activated or not (True/False)."""
@@ -127,6 +155,111 @@ async def async_arm(self, value):
             return await api.request_system_arm(self.blink, self.network_id)
         return await api.request_system_disarm(self.blink, self.network_id)
 
+    async def async_set_scheduler(self, enable):
+        """Enable or disable the Blink native schedule for this network.
+
+        :param enable: True to enable the schedule, False to disable it.
+        """
+        if self.scheduler_enabled == enable:
+            _LOGGER.debug("Schedule already in state %s, skipping.", enable)
+            return True
+
+        # Use the programs list already cached by get_network_info() where
+        # possible to avoid a redundant API call.  Fall back to a fresh fetch
+        # only if the cache is absent (e.g. in tests or before first refresh).
+        cached_programs = (
+            self.network_info.get("programs") if self.network_info else None
+        )
+        if cached_programs is None:
+            _LOGGER.debug("Programs not cached; fetching from API.")
+            cached_programs = await api.request_programs(self.blink, self.network_id)
+
+        if not enable:
+            # Prefer active_program_id from the network summary if present
+            program_id = self.network_info.get("network", {}).get("active_program_id")
+
+            # Otherwise find the currently enabled program from the list
+            if program_id is None and isinstance(cached_programs, list):
+                program_id = next(
+                    (
+                        p.get("id")
+                        for p in cached_programs
+                        if p.get("status") == "enabled"
+                    ),
+                    None,
+                )
+
+            # If scheduler_enabled is True but no program ID is found, we cannot call
+            # the API disable endpoint (which requires a program ID). Since the
+            # requested operation fails, this is logged as a warning.
+            if program_id is None:
+                _LOGGER.warning("Could not find an active program to disable.")
+                return False
+
+            result = await api.request_program_disable(
+                self.blink, self.network_id, program_id
+            )
+            # Optimistically update the cache so scheduler_enabled reflects
+            # the new state without requiring a full refresh.
+            if result:
+                self._update_cached_program_status(program_id, "disabled")
+            return result
+
+        # To enable, use the first available program from the cached list.
+        # Blink networks typically support only a single native schedule (program),
+        # so we default to enabling the first one in the list.
+        try:
+            program_id = cached_programs[0]["id"]
+        except (KeyError, IndexError, TypeError):
+            _LOGGER.warning(
+                "Could not find a program to enable on network %s.", self.name
+            )
+            return False
+
+        result = await api.request_program_enable(
+            self.blink, self.network_id, program_id
+        )
+        # Optimistically update the cache so scheduler_enabled reflects
+        # the new state without requiring a full refresh.
+        if result:
+            self._update_cached_program_status(program_id, "enabled")
+        return result
+
+    def _update_cached_program_status(self, program_id, status):
+        """Update the cached programs list after an enable/disable call.
+
+        The Blink API response for request_program_enable/disable only returns
+        a generic success message rather than the new program state. By
+        optimistically updating our local cache, scheduler_enabled remains
+        accurate immediately following async_set_scheduler, saving an
+        expensive and redundant network refresh.
+
+        :param program_id: ID of the program that was changed.
+        :param status: New status string, e.g. ``"enabled"`` or ``"disabled"``.
+        """
+        try:
+            programs = (
+                self.network_info.get("programs", []) if self.network_info else []
+            )
+            if isinstance(programs, list):
+                program = next((p for p in programs if p.get("id") == program_id), None)
+                if program:
+                    program["status"] = status
+                    _LOGGER.debug(
+                        "Optimistically updated program %s status to %s.",
+                        program_id,
+                        status,
+                    )
+            # active_program_id in the network summary takes priority in
+            # scheduler_enabled, so keep it consistent with the new state.
+            network = self.network_info.get("network", {}) if self.network_info else {}
+            if status == "disabled":
+                network["active_program_id"] = None
+            else:
+                network["active_program_id"] = program_id
+        except (AttributeError, TypeError):
+            pass
+
     async def start(self):
         """Initialize the system."""
         _LOGGER.debug("Initializing the sync module")
@@ -269,6 +402,12 @@ async def get_network_info(self):
         try:
             if self.network_info["network"]["sync_module_error"]:
                 raise KeyError
+
+            # Cache the programs list locally during refresh so scheduler_enabled
+            # can check it synchronously without triggering redundant network calls.
+            programs = await api.request_programs(self.blink, self.network_id)
+            if isinstance(programs, list):
+                self.network_info["programs"] = programs
         except (TypeError, KeyError):
             self.available = False
             return False
@@ -378,8 +517,7 @@ async def check_new_videos(self):
                 # Exit the loop once there are no new videos in the list.
                 if not self.check_new_video_time(iso_timestamp, last_manifest_read):
                     _LOGGER.info(
-                        "No new local storage videos since last manifest "
-                        "read at %s.",
+                        "No new local storage videos since last manifest read at %s.",
                         last_read_local,
                     )
                     break
diff --git a/tests/test_api.py b/tests/test_api.py
index f8f77b89..384e6244 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -205,3 +205,25 @@ async def test_wait_for_command(self, mock_resp):
 
         response = await api.wait_for_command(self.blink, None)
         self.assertFalse(response)
+
+    async def test_request_programs(self, mock_resp):
+        """Test request_programs."""
+        programs = [{"id": 123, "status": "enabled", "name": "My Schedule"}]
+        mock_resp.return_value = programs
+        self.assertEqual(await api.request_programs(self.blink, "network"), programs)
+
+    async def test_request_program_enable(self, mock_resp):
+        """Test request_program_enable."""
+        mock_resp.return_value = {"status": 200}
+        self.assertEqual(
+            await api.request_program_enable(self.blink, "network", 123),
+            {"status": 200},
+        )
+
+    async def test_request_program_disable(self, mock_resp):
+        """Test request_program_disable."""
+        mock_resp.return_value = {"status": 200}
+        self.assertEqual(
+            await api.request_program_disable(self.blink, "network", 123),
+            {"status": 200},
+        )
diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py
index b35f56fd..ea56eed8 100644
--- a/tests/test_sync_module.py
+++ b/tests/test_sync_module.py
@@ -32,9 +32,7 @@ def setUp(self):
         self.blink: Blink = Blink(motion_interval=0, session=mock.AsyncMock())
         self.blink.last_refresh = 0
         self.blink.urls = BlinkURLHandler("test")
-        self.blink.sync["test"]: BlinkSyncModule = BlinkSyncModule(
-            self.blink, "test", "1234", []
-        )
+        self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", "1234", [])
         self.blink.sync["test"].network_info = {"network": {"armed": True}}
         self.camera: BlinkCamera = BlinkCamera(self.blink.sync)
         self.mock_start = [
@@ -142,9 +140,9 @@ async def test_get_camera_info_fail(self, mock_resp) -> None:
 
     async def test_get_network_info(self, mock_resp) -> None:
         """Test network retrieval."""
-        mock_resp.return_value = {"network": {"sync_module_error": False}}
+        mock_resp.side_effect = [{"network": {"sync_module_error": False}}, []]
         self.assertTrue(await self.blink.sync["test"].get_network_info())
-        mock_resp.return_value = {"network": {"sync_module_error": True}}
+        mock_resp.side_effect = [{"network": {"sync_module_error": True}}, []]
         self.assertFalse(await self.blink.sync["test"].get_network_info())
 
     async def test_get_network_info_failure(self, mock_resp) -> None:
@@ -158,6 +156,163 @@ async def test_get_network_info_failure(self, mock_resp) -> None:
         self.assertFalse(await self.blink.sync["test"].get_network_info())
         self.assertFalse(self.blink.sync["test"].available)
 
+    def test_scheduler_enabled(self, mock_resp) -> None:
+        """Test scheduler_enabled property."""
+        sync_module = self.blink.sync["test"]
+
+        # Primary path: active_program_id present in network summary
+        sync_module.network_info = {"network": {"active_program_id": 123}}
+        self.assertTrue(sync_module.scheduler_enabled)
+        sync_module.network_info = {"network": {"active_program_id": None}}
+        self.assertFalse(sync_module.scheduler_enabled)
+
+        # Primary path: autoarm_time_enable flag
+        sync_module.network_info = {"network": {"autoarm_time_enable": True}}
+        self.assertTrue(sync_module.scheduler_enabled)
+
+        # Fallback path: programs list cached by get_network_info
+        sync_module.network_info = {
+            "network": {},
+            "programs": [{"id": 122793, "status": "enabled"}],
+        }
+        self.assertTrue(sync_module.scheduler_enabled)
+
+        sync_module.network_info = {
+            "network": {},
+            "programs": [{"id": 122793, "status": "disabled"}],
+        }
+        self.assertFalse(sync_module.scheduler_enabled)
+
+        sync_module.network_info = {"network": {}, "programs": []}
+        self.assertFalse(sync_module.scheduler_enabled)
+
+        # No programs key: emits warning, returns False
+        sync_module.network_info = {"network": {}}
+        self.assertFalse(sync_module.scheduler_enabled)
+
+        # None network_info
+        sync_module.network_info = None
+        self.assertFalse(sync_module.scheduler_enabled)
+
+    async def test_async_set_scheduler_disable(self, mock_resp) -> None:
+        """Test async_set_scheduler disable updates the cache optimistically."""
+        sync_module = self.blink.sync["test"]
+        sync_module.network_info = {
+            "network": {"active_program_id": 123},
+            "programs": [{"id": 123, "status": "enabled"}],
+        }
+        mock_resp.return_value = {"status": 200}
+        self.assertEqual(await sync_module.async_set_scheduler(False), {"status": 200})
+        # Cache should be updated immediately without needing a refresh
+        self.assertFalse(sync_module.scheduler_enabled)
+        self.assertEqual(sync_module.network_info["programs"][0]["status"], "disabled")
+
+    async def test_async_set_scheduler_disable_already_disabled(
+        self, mock_resp
+    ) -> None:
+        """Test async_set_scheduler disable when already disabled."""
+        sync_module = self.blink.sync["test"]
+        sync_module.network_info = {"network": {"active_program_id": None}}
+        self.assertTrue(await sync_module.async_set_scheduler(False))
+
+    async def test_async_set_scheduler_enable(self, mock_resp) -> None:
+        """Test async_set_scheduler enable updates the cache optimistically."""
+        sync_module = self.blink.sync["test"]
+        # Pre-populate the cache so no extra API call is made
+        sync_module.network_info = {
+            "network": {"active_program_id": None},
+            "programs": [{"id": 456, "status": "disabled"}],
+        }
+        mock_resp.return_value = {"status": 200}
+        self.assertEqual(await sync_module.async_set_scheduler(True), {"status": 200})
+        # Cache should be updated immediately without needing a refresh
+        self.assertTrue(sync_module.scheduler_enabled)
+        self.assertEqual(sync_module.network_info["programs"][0]["status"], "enabled")
+
+    async def test_async_set_scheduler_enable_uncached(self, mock_resp) -> None:
+        """Test async_set_scheduler fetches programs when cache is absent."""
+        sync_module = self.blink.sync["test"]
+        # No "programs" key → falls back to live API call
+        sync_module.network_info = {"network": {"active_program_id": None}}
+        mock_resp.side_effect = [
+            [{"id": 456}],
+            {"status": 200},
+        ]
+        self.assertEqual(await sync_module.async_set_scheduler(True), {"status": 200})
+
+    async def test_async_set_scheduler_enable_uncached_none(self, mock_resp) -> None:
+        """Test fallback when request_programs returns None."""
+        sync_module = self.blink.sync["test"]
+        # No "programs" key → falls back to live API call
+        sync_module.network_info = {"network": {"active_program_id": None}}
+        mock_resp.return_value = None
+        self.assertFalse(await sync_module.async_set_scheduler(True))
+
+    async def test_async_set_scheduler_enable_already_enabled(self, mock_resp):
+        """Test async_set_scheduler enable when already enabled."""
+        sync_module = self.blink.sync["test"]
+        sync_module.network_info = {"network": {"active_program_id": 123}}
+        self.assertTrue(await sync_module.async_set_scheduler(True))
+
+    async def test_async_set_scheduler_enable_fail(self, mock_resp) -> None:
+        """Test async_set_scheduler enable failure when no programs exist."""
+        sync_module = self.blink.sync["test"]
+        # Empty cached programs list → nothing to enable
+        sync_module.network_info = {
+            "network": {"active_program_id": None},
+            "programs": [],
+        }
+        self.assertFalse(await sync_module.async_set_scheduler(True))
+
+    async def test_async_set_scheduler_disable_network_failure(self, mock_resp) -> None:
+        """Test async_set_scheduler disable API error without modifying cache."""
+        sync_module = self.blink.sync["test"]
+        sync_module.network_info = {
+            "network": {"active_program_id": 123},
+            "programs": [{"id": 123, "status": "enabled"}],
+        }
+        mock_resp.return_value = None  # Simulate network failure
+        self.assertIsNone(await sync_module.async_set_scheduler(False))
+        # Cache must not be modified since the API failed
+        self.assertTrue(sync_module.scheduler_enabled)
+        self.assertEqual(sync_module.network_info["programs"][0]["status"], "enabled")
+
+    async def test_async_set_scheduler_enable_network_failure(self, mock_resp) -> None:
+        """Test async_set_scheduler enable API error without modifying cache."""
+        sync_module = self.blink.sync["test"]
+        sync_module.network_info = {
+            "network": {"active_program_id": None},
+            "programs": [{"id": 456, "status": "disabled"}],
+        }
+        mock_resp.return_value = None  # Simulate network failure
+        self.assertIsNone(await sync_module.async_set_scheduler(True))
+        # Cache must not be modified since the API failed
+        self.assertFalse(sync_module.scheduler_enabled)
+        self.assertEqual(sync_module.network_info["programs"][0]["status"], "disabled")
+
+    def test_update_cached_program_status(self, mock_resp) -> None:
+        """Test that _update_cached_program_status mutates the cache correctly."""
+        sync_module = self.blink.sync["test"]
+        sync_module.network_info = {
+            "network": {},
+            "programs": [
+                {"id": 100, "status": "enabled"},
+                {"id": 200, "status": "disabled"},
+            ],
+        }
+        sync_module._update_cached_program_status(100, "disabled")
+        self.assertEqual(sync_module.network_info["programs"][0]["status"], "disabled")
+
+        sync_module._update_cached_program_status(200, "enabled")
+        self.assertEqual(sync_module.network_info["programs"][1]["status"], "enabled")
+
+        # Updating a non-existent program ID should not raise
+        sync_module._update_cached_program_status(999, "enabled")
+
+        # Should not raise when network_info is None
+        sync_module.network_info = None
+        sync_module._update_cached_program_status(100, "disabled")
+
     async def test_check_new_videos_startup(self, mock_resp) -> None:
         """Test that check_new_videos does not block startup."""
         sync_module = self.blink.sync["test"]
@@ -542,7 +697,7 @@ async def test_local_storage_media_item(self, mock_resp):
         item = LocalStorageMediaItem(
             "1234",
             "Backdoor",
-            datetime.datetime.utcnow().isoformat(),
+            (datetime.datetime.utcnow() - datetime.timedelta(seconds=60)).isoformat(),
             "432",
             " manifest_id",
             "url",