Repository: tellor-io/telliot-feeds (language: python)

Issue:
Handle tips that require manual input, skip manual data sources after 10min
Query IDs like `[uspce-legacy|gas-price-oracle-example|snapshot-proposal-example|diva-protocol-example|string-query-example]` may require manual inputs from the reporter. 

Currently when these querys are tipped, telliot will pick these up and want to submit for them. This could be an issue for reporters who are not actively watching their reporters. If they are sleeping or away from their computer, telliot is stuck on a prompt for manual input and opportunities to automatically submit for other tips could be lost.

Should there be a flag in telliot to ignore tips/requests for QueryTypes that require manual inputs? 


Discussion / hints:
Thanks for this @spuddyminer . I think it should default to running continuously, so adding a flag to allow halts for manual input sounds good.
Fix- Wait for manual input for 10 minutes and if no answer then skip.

Fix patch:
diff --git a/src/telliot_feeds/feeds/daily_volatility_manual_feed.py b/src/telliot_feeds/feeds/daily_volatility_manual_feed.py
index 24d661c7..0c5b1b9c 100644
--- a/src/telliot_feeds/feeds/daily_volatility_manual_feed.py
+++ b/src/telliot_feeds/feeds/daily_volatility_manual_feed.py
@@ -1,6 +1,6 @@
 from telliot_feeds.datafeed import DataFeed
 from telliot_feeds.queries.daily_volatility import DailyVolatility
-from telliot_feeds.sources.manual_sources.daily_volatility_manual_source import DailyVolatilityManualSource
+from telliot_feeds.sources.manual.daily_volatility_manual_source import DailyVolatilityManualSource
 
 
 asset = ""
diff --git a/src/telliot_feeds/feeds/diva_manual_feed.py b/src/telliot_feeds/feeds/diva_manual_feed.py
index 6a3c9a92..8ec437ea 100644
--- a/src/telliot_feeds/feeds/diva_manual_feed.py
+++ b/src/telliot_feeds/feeds/diva_manual_feed.py
@@ -1,6 +1,6 @@
 from telliot_feeds.datafeed import DataFeed
 from telliot_feeds.queries.diva_protocol import DIVAProtocol
-from telliot_feeds.sources.manual_sources.diva_manual_source import DivaManualSource
+from telliot_feeds.sources.manual.diva_manual_source import DivaManualSource
 
 
 pool_id = None
diff --git a/src/telliot_feeds/feeds/numeric_api_response_manual_feed.py b/src/telliot_feeds/feeds/numeric_api_response_manual_feed.py
index b1ddf06c..dfb411d4 100644
--- a/src/telliot_feeds/feeds/numeric_api_response_manual_feed.py
+++ b/src/telliot_feeds/feeds/numeric_api_response_manual_feed.py
@@ -1,6 +1,6 @@
 from telliot_feeds.datafeed import DataFeed
 from telliot_feeds.queries.numeric_api_response_query import NumericApiResponse
-from telliot_feeds.sources.manual_sources.numeric_api_manual_response import NumericApiManualResponse
+from telliot_feeds.sources.manual.numeric_api_manual_response import NumericApiManualResponse
 
 
 url = None
diff --git a/src/telliot_feeds/feeds/spot_price_manual_feed.py b/src/telliot_feeds/feeds/spot_price_manual_feed.py
index ee7c3589..06b5e18c 100644
--- a/src/telliot_feeds/feeds/spot_price_manual_feed.py
+++ b/src/telliot_feeds/feeds/spot_price_manual_feed.py
@@ -1,6 +1,6 @@
 from telliot_feeds.datafeed import DataFeed
 from telliot_feeds.queries.price.spot_price import SpotPrice
-from telliot_feeds.sources.manual_sources.spot_price_input_source import SpotPriceManualSource
+from telliot_feeds.sources.manual.spot_price_input_source import SpotPriceManualSource
 
 # Using defaults to bypass an error handle for feeds w/out non-manual sources
 asset: str = "eth"
diff --git a/src/telliot_feeds/feeds/string_query_feed.py b/src/telliot_feeds/feeds/string_query_feed.py
index 4055c0e4..06e48b90 100644
--- a/src/telliot_feeds/feeds/string_query_feed.py
+++ b/src/telliot_feeds/feeds/string_query_feed.py
@@ -1,6 +1,6 @@
 from telliot_feeds.datafeed import DataFeed
 from telliot_feeds.queries.string_query import StringQuery
-from telliot_feeds.sources.manual_sources.string_query_manual_source import StringQueryManualSource
+from telliot_feeds.sources.manual.string_query_manual_source import StringQueryManualSource
 
 text = None
 
diff --git a/src/telliot_feeds/feeds/tellor_rng_manual_feed.py b/src/telliot_feeds/feeds/tellor_rng_manual_feed.py
index 3d9715d7..76dacec9 100644
--- a/src/telliot_feeds/feeds/tellor_rng_manual_feed.py
+++ b/src/telliot_feeds/feeds/tellor_rng_manual_feed.py
@@ -1,6 +1,6 @@
 from telliot_feeds.datafeed import DataFeed
 from telliot_feeds.queries.tellor_rng import TellorRNG
-from telliot_feeds.sources.manual_sources.tellor_rng_manual_source import TellorRNGManualInputSource
+from telliot_feeds.sources.manual.tellor_rng_manual_source import TellorRNGManualInputSource
 
 
 timestamp = None
diff --git a/src/telliot_feeds/feeds/twap_manual_feed.py b/src/telliot_feeds/feeds/twap_manual_feed.py
index a8b7e957..6553df03 100644
--- a/src/telliot_feeds/feeds/twap_manual_feed.py
+++ b/src/telliot_feeds/feeds/twap_manual_feed.py
@@ -2,7 +2,7 @@
 
 from telliot_feeds.datafeed import DataFeed
 from telliot_feeds.queries.price.twap import TWAP
-from telliot_feeds.sources.manual_sources.twap_manual_input_source import TWAPManualSource
+from telliot_feeds.sources.manual.twap_manual_input_source import TWAPManualSource
 
 
 # Using defaults to bypass an error handle for auto feeds and unsupported currencies
diff --git a/src/telliot_feeds/feeds/uspce_feed.py b/src/telliot_feeds/feeds/uspce_feed.py
index 16f3c636..84d5b698 100644
--- a/src/telliot_feeds/feeds/uspce_feed.py
+++ b/src/telliot_feeds/feeds/uspce_feed.py
@@ -1,7 +1,7 @@
 """Example datafeed used by USPCEReporter."""
 from telliot_feeds.datafeed import DataFeed
 from telliot_feeds.queries.ampleforth.uspce import AmpleforthUSPCE
-from telliot_feeds.sources.uspce import USPCESource
+from telliot_feeds.sources.manual.uspce import USPCESource
 
 
 uspce_feed = DataFeed(query=AmpleforthUSPCE(), source=USPCESource())
diff --git a/src/telliot_feeds/sources/manual_sources/__init__.py b/src/telliot_feeds/sources/manual/__init__.py
similarity index 100%
rename from src/telliot_feeds/sources/manual_sources/__init__.py
rename to src/telliot_feeds/sources/manual/__init__.py
diff --git a/src/telliot_feeds/sources/manual_sources/daily_volatility_manual_source.py b/src/telliot_feeds/sources/manual/daily_volatility_manual_source.py
similarity index 80%
rename from src/telliot_feeds/sources/manual_sources/daily_volatility_manual_source.py
rename to src/telliot_feeds/sources/manual/daily_volatility_manual_source.py
index 57e22100..67d4e0b9 100644
--- a/src/telliot_feeds/sources/manual_sources/daily_volatility_manual_source.py
+++ b/src/telliot_feeds/sources/manual/daily_volatility_manual_source.py
@@ -3,6 +3,8 @@
 from telliot_feeds.datasource import DataSource
 from telliot_feeds.dtypes.datapoint import datetime_now_utc
 from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
 from telliot_feeds.utils.log import get_logger
 
 
@@ -21,7 +23,7 @@ def parse_user_val(self) -> float:
         num = None
 
         while num is None:
-            usr_inpt = input()
+            usr_inpt = input_timeout()
 
             try:
                 inpt = float(usr_inpt)
@@ -35,7 +37,7 @@ def parse_user_val(self) -> float:
             print(f"\nVolatility index (with 18 decimals of precision) to be submitted on chain:  {inpt*10**18:.0f}")
             print("Press [ENTER] to confirm.")
 
-            _ = input()
+            _ = input_timeout()
 
             num = inpt
 
@@ -47,7 +49,11 @@ async def fetch_new_datapoint(self) -> OptionalDataPoint[float]:
         Returns:
             Current time-stamped value
         """
-        response = self.parse_user_val()
+        try:
+            response = self.parse_user_val()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user input")
+            return None, None
 
         datapoint = (response, datetime_now_utc())
         self.store_datapoint(datapoint)
diff --git a/src/telliot_feeds/sources/manual_sources/diva_manual_source.py b/src/telliot_feeds/sources/manual/diva_manual_source.py
similarity index 85%
rename from src/telliot_feeds/sources/manual_sources/diva_manual_source.py
rename to src/telliot_feeds/sources/manual/diva_manual_source.py
index 63024356..544fb854 100644
--- a/src/telliot_feeds/sources/manual_sources/diva_manual_source.py
+++ b/src/telliot_feeds/sources/manual/diva_manual_source.py
@@ -4,6 +4,8 @@
 from telliot_feeds.datasource import DataSource
 from telliot_feeds.dtypes.datapoint import datetime_now_utc
 from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
 from telliot_feeds.utils.log import get_logger
 
 
@@ -16,7 +18,7 @@ def get_price_from_user(param: str) -> float:
     print(f"Type price of {param} and press [ENTER]")
 
     while param_price is None:
-        inpt = input()
+        inpt = input_timeout()
         try:
             price = float(inpt)
         except ValueError:
@@ -52,7 +54,7 @@ def parse_user_val(self) -> list[float]:
         )
         print("Press [ENTER] to confirm.")
 
-        _ = input()
+        _ = input_timeout()
 
         return prices
 
@@ -62,7 +64,11 @@ async def fetch_new_datapoint(self) -> OptionalDataPoint[list[float]]:
         Returns:
             Current time-stamped value
         """
-        response = self.parse_user_val()
+        try:
+            response = self.parse_user_val()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user input")
+            return None, None
 
         datapoint = (response, datetime_now_utc())
         self.store_datapoint(datapoint)
diff --git a/src/telliot_feeds/sources/manual_sources/numeric_api_manual_response.py b/src/telliot_feeds/sources/manual/numeric_api_manual_response.py
similarity index 71%
rename from src/telliot_feeds/sources/manual_sources/numeric_api_manual_response.py
rename to src/telliot_feeds/sources/manual/numeric_api_manual_response.py
index 47fef80e..f2ddf426 100644
--- a/src/telliot_feeds/sources/manual_sources/numeric_api_manual_response.py
+++ b/src/telliot_feeds/sources/manual/numeric_api_manual_response.py
@@ -1,8 +1,10 @@
 from dataclasses import dataclass
 
 from telliot_feeds.datasource import DataSource
-from telliot_feeds.dtypes.datapoint import DataPoint
 from telliot_feeds.dtypes.datapoint import datetime_now_utc
+from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
 from telliot_feeds.utils.log import get_logger
 
 
@@ -21,10 +23,10 @@ def parse_user_val(self) -> float:
         val = None
 
         while val is None:
-            usr_inpt = input()
+            usr_inpt = input_timeout()
 
             try:
-                usr_inpt = float(usr_inpt)  # type: ignore
+                usr_inpt = float(usr_inpt)
             except ValueError:
                 print("Invalid input. Enter a numerical value (float).")
                 continue
@@ -35,19 +37,23 @@ def parse_user_val(self) -> float:
             )
             print("Press [ENTER] to confirm.")
 
-            _ = input()
+            _ = input_timeout()
 
             val = usr_inpt
 
         return val
 
-    async def fetch_new_datapoint(self) -> DataPoint[float]:
+    async def fetch_new_datapoint(self) -> OptionalDataPoint[float]:
         """Update current value with time-stamped value fetched from user input.
 
         Returns:
             Current time-stamped value
         """
-        response = self.parse_user_val()
+        try:
+            response = self.parse_user_val()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user input")
+            return None, None
 
         datapoint = (response, datetime_now_utc())
         self.store_datapoint(datapoint)
diff --git a/src/telliot_feeds/sources/manual_sources/spot_price_input_source.py b/src/telliot_feeds/sources/manual/spot_price_input_source.py
similarity index 64%
rename from src/telliot_feeds/sources/manual_sources/spot_price_input_source.py
rename to src/telliot_feeds/sources/manual/spot_price_input_source.py
index 13f7b108..ce550c97 100644
--- a/src/telliot_feeds/sources/manual_sources/spot_price_input_source.py
+++ b/src/telliot_feeds/sources/manual/spot_price_input_source.py
@@ -1,8 +1,11 @@
+import asyncio
 from dataclasses import dataclass
 
 from telliot_feeds.datasource import DataSource
-from telliot_feeds.dtypes.datapoint import DataPoint
 from telliot_feeds.dtypes.datapoint import datetime_now_utc
+from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
 from telliot_feeds.utils.log import get_logger
 
 
@@ -21,10 +24,10 @@ def parse_user_val(self) -> float:
         spot = None
 
         while spot is None:
-            usr_inpt = input()
+            usr_inpt = input_timeout()
 
             try:
-                usr_inpt = float(usr_inpt)  # type: ignore
+                usr_inpt = float(usr_inpt)
             except ValueError:
                 print("Invalid input. Enter decimal value (float).")
                 continue
@@ -32,19 +35,23 @@ def parse_user_val(self) -> float:
             print(f"\nSpot price (with 18 decimals of precision) to be submitted on chain: {usr_inpt*10**18:.0f}")
             print("Press [ENTER] to confirm.")
 
-            _ = input()
+            _ = input_timeout()
 
             spot = usr_inpt
 
         return spot
 
-    async def fetch_new_datapoint(self) -> DataPoint[float]:
+    async def fetch_new_datapoint(self) -> OptionalDataPoint[float]:
         """Update current value with time-stamped value fetched from user input.
 
         Returns:
             Current time-stamped value
         """
-        price = self.parse_user_val()
+        try:
+            price = self.parse_user_val()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user input")
+            return None, None
 
         datapoint = (price, datetime_now_utc())
         self.store_datapoint(datapoint)
@@ -52,3 +59,9 @@ async def fetch_new_datapoint(self) -> DataPoint[float]:
         logger.info(f"Spot price {datapoint[0]} retrieved at time {datapoint[1]}")
 
         return datapoint
+
+
+if __name__ == "__main__":
+    spot = SpotPriceManualSource()
+    v, t = asyncio.run(spot.fetch_new_datapoint())
+    print("datapoint:", v, t)
diff --git a/src/telliot_feeds/sources/manual_sources/string_query_manual_source.py b/src/telliot_feeds/sources/manual/string_query_manual_source.py
similarity index 52%
rename from src/telliot_feeds/sources/manual_sources/string_query_manual_source.py
rename to src/telliot_feeds/sources/manual/string_query_manual_source.py
index 57208e8d..bd9ad26c 100644
--- a/src/telliot_feeds/sources/manual_sources/string_query_manual_source.py
+++ b/src/telliot_feeds/sources/manual/string_query_manual_source.py
@@ -3,6 +3,12 @@
 from telliot_feeds.datasource import DataSource
 from telliot_feeds.dtypes.datapoint import datetime_now_utc
 from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
+from telliot_feeds.utils.log import get_logger
+
+
+logger = get_logger(__name__)
 
 
 class StringQueryManualSource(DataSource[Optional[str]]):
@@ -10,11 +16,19 @@ async def fetch_new_datapoint(self) -> OptionalDataPoint[str]:
 
         print("Type your string query response:\n")
 
-        usr_inpt = input()
+        try:
+            usr_inpt = input_timeout()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user input")
+            return None, None
 
         print(f"\nString query response to be submitted to oracle->: {usr_inpt}")
         print("Press [ENTER] to confirm.")
-        _ = input()
+        try:
+            _ = input_timeout()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user to confirm")
+            return None, None
 
         datapoint = (usr_inpt, datetime_now_utc())
         self.store_datapoint(datapoint)
diff --git a/src/telliot_feeds/sources/manual_sources/tellor_rng_manual_source.py b/src/telliot_feeds/sources/manual/tellor_rng_manual_source.py
similarity index 84%
rename from src/telliot_feeds/sources/manual_sources/tellor_rng_manual_source.py
rename to src/telliot_feeds/sources/manual/tellor_rng_manual_source.py
index dcc93653..bece6c46 100644
--- a/src/telliot_feeds/sources/manual_sources/tellor_rng_manual_source.py
+++ b/src/telliot_feeds/sources/manual/tellor_rng_manual_source.py
@@ -6,8 +6,11 @@
 from telliot_feeds.datasource import DataSource
 from telliot_feeds.dtypes.datapoint import datetime_now_utc
 from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
 from telliot_feeds.utils.log import get_logger
 
+
 logger = get_logger(__name__)
 
 
@@ -21,7 +24,7 @@ def parse_user_input(self) -> bytes:
         response = None
 
         while response is None:
-            user_input = input()
+            user_input = input_timeout()
             if len(user_input) < 2:
                 print(
                     "Invalid input! Not enough characters, "
@@ -41,7 +44,7 @@ def parse_user_input(self) -> bytes:
                 continue
             print(f"\nTellorRNG value to be submitted on chain: {user_input}")
             print("Press [ENTER] to continue")
-            _ = input()
+            _ = input_timeout()
             response = val
 
         return response
@@ -52,7 +55,11 @@ async def fetch_new_datapoint(self) -> OptionalDataPoint[bytes]:
         Returns:
             Current time-stamped value
         """
-        response = self.parse_user_input()
+        try:
+            response = self.parse_user_input()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user input")
+            return None, None
 
         datapoint = (response, datetime_now_utc())
         self.store_datapoint(datapoint)
diff --git a/src/telliot_feeds/sources/manual_sources/twap_manual_input_source.py b/src/telliot_feeds/sources/manual/twap_manual_input_source.py
similarity index 80%
rename from src/telliot_feeds/sources/manual_sources/twap_manual_input_source.py
rename to src/telliot_feeds/sources/manual/twap_manual_input_source.py
index 541d8cb0..be153e9b 100644
--- a/src/telliot_feeds/sources/manual_sources/twap_manual_input_source.py
+++ b/src/telliot_feeds/sources/manual/twap_manual_input_source.py
@@ -3,8 +3,11 @@
 from telliot_feeds.datasource import DataSource
 from telliot_feeds.dtypes.datapoint import datetime_now_utc
 from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
 from telliot_feeds.utils.log import get_logger
 
+
 logger = get_logger(__name__)
 
 
@@ -20,7 +23,7 @@ def parse_user_val(self) -> float:
         num = None
 
         while num is None:
-            usr_inpt = input()
+            usr_inpt = input_timeout()
 
             try:
                 inpt = float(usr_inpt)
@@ -34,7 +37,7 @@ def parse_user_val(self) -> float:
             print(f"\nTWAP value (with 18 decimals of precision) to be submitted on chain:  {inpt*10**18:.0f}")
             print("Press [ENTER] to confirm.")
 
-            _ = input()
+            _ = input_timeout()
 
             num = inpt
 
@@ -46,7 +49,11 @@ async def fetch_new_datapoint(self) -> OptionalDataPoint[float]:
         Returns:
             Current time-stamped value
         """
-        response = self.parse_user_val()
+        try:
+            response = self.parse_user_val()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user input")
+            return None, None
 
         datapoint = (response, datetime_now_utc())
         self.store_datapoint(datapoint)
diff --git a/src/telliot_feeds/sources/uspce.py b/src/telliot_feeds/sources/manual/uspce.py
similarity index 68%
rename from src/telliot_feeds/sources/uspce.py
rename to src/telliot_feeds/sources/manual/uspce.py
index a0709c98..504fca3c 100644
--- a/src/telliot_feeds/sources/uspce.py
+++ b/src/telliot_feeds/sources/manual/uspce.py
@@ -1,8 +1,10 @@
 from dataclasses import dataclass
 
 from telliot_feeds.datasource import DataSource
-from telliot_feeds.dtypes.datapoint import DataPoint
 from telliot_feeds.dtypes.datapoint import datetime_now_utc
+from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
 from telliot_feeds.utils.log import get_logger
 
 
@@ -16,7 +18,7 @@ class USPCESource(DataSource[float]):
     def parse_user_val(test_input: str) -> float:
         """Parse USPCE value from user input."""
         # This arg is to avoid a TypeError when the default
-        # input() method is overriden in test_source.py.
+        # input_timeout() method is overriden in test_source.py.
         # The error says this method expects no params,
         # but is passed one. TODO: fix
         _ = test_input
@@ -26,28 +28,32 @@ def parse_user_val(test_input: str) -> float:
         uspce = None
 
         while uspce is None:
-            inpt = input()
+            inpt = input_timeout()
 
             try:
-                inpt = float(inpt)  # type: ignore
+                inpt = float(inpt)
             except ValueError:
                 print("Invalid input. Enter decimal value (float).")
                 continue
 
             print(f"Submitting value: {inpt}\nPress [ENTER] to confirm.")
-            _ = input()
+            _ = input_timeout()
 
             uspce = inpt
 
         return uspce
 
-    async def fetch_new_datapoint(self) -> DataPoint[float]:
+    async def fetch_new_datapoint(self) -> OptionalDataPoint[float]:
         """Update current value with time-stamped value fetched from user input.
 
         Returns:
             Current time-stamped value
         """
-        uspce = self.parse_user_val()
+        try:
+            uspce = self.parse_user_val()
+        except TimeoutOccurred:
+            logger.info("Timeout occurred while waiting for user input")
+            return None, None
 
         datapoint = (uspce, datetime_now_utc())
         self.store_datapoint(datapoint)
diff --git a/src/telliot_feeds/utils/input_timeout.py b/src/telliot_feeds/utils/input_timeout.py
new file mode 100644
index 00000000..c2db8b94
--- /dev/null
+++ b/src/telliot_feeds/utils/input_timeout.py
@@ -0,0 +1,96 @@
+""""
+Original code taken from unmaintained package:
+https://github.com/johejo/inputimeout/blob/master/inputimeout/inputimeout.py
+"""
+import sys
+from typing import Any
+
+DEFAULT_TIMEOUT = 600.0  # 10 mins
+# DEFAULT_TIMEOUT = 10.0  # 10 secs
+INTERVAL = 0.05
+
+SP = " "
+CR = "\r"
+LF = "\n"
+CRLF = CR + LF
+
+
+class TimeoutOccurred(Exception):
+    pass
+
+
+def echo(string: str) -> None:
+    sys.stdout.write(string)
+    sys.stdout.flush()
+
+
+def posix_inputimeout(prompt: str = "", timeout: float = DEFAULT_TIMEOUT) -> Any:
+    echo(prompt)
+    sel = selectors.DefaultSelector()
+    sel.register(sys.stdin, selectors.EVENT_READ)
+    events = sel.select(timeout)
+
+    if events:
+        key, _ = events[0]
+        return key.fileobj.readline().rstrip(LF)  # type: ignore
+    else:
+        echo(LF)
+        termios.tcflush(sys.stdin, termios.TCIFLUSH)
+        raise TimeoutOccurred
+
+
+def win_inputimeout(prompt: str = "", timeout: float = DEFAULT_TIMEOUT) -> str:
+    echo(prompt)
+    begin = time.monotonic()
+    end = begin + timeout
+    line = ""
+
+    while time.monotonic() < end:
+        if msvcrt.kbhit():  # type: ignore
+            c = msvcrt.getwche()  # type: ignore
+            if c in (CR, LF):
+                echo(CRLF)
+                return line
+            if c == "\003":
+                raise KeyboardInterrupt
+            if c == "\b":
+                line = line[:-1]
+                cover = SP * len(prompt + line + SP)
+                echo("".join([CR, cover, CR, prompt, line]))
+            else:
+                line += c
+        time.sleep(INTERVAL)
+
+    echo(CRLF)
+    raise TimeoutOccurred
+
+
+try:
+    import msvcrt
+
+except ImportError:
+    import selectors
+    import termios
+
+    input_timeout_func = posix_inputimeout
+
+else:
+    import time
+
+    input_timeout_func = win_inputimeout
+
+
+class InputTimeout:
+    def __call__(self, prompt: str = "", timeout: float = DEFAULT_TIMEOUT) -> Any:
+        return input_timeout_func(prompt, timeout)
+
+
+input_timeout = InputTimeout()
+
+
+if __name__ == "__main__":
+    try:
+        user_input = input_timeout("Type something: ", 5.0)
+        print(f"You typed: {user_input}")
+    except TimeoutOccurred:
+        print("Timeout occurred")


Test patch:
diff --git a/tests/sources/manual/test_diva_manual_entry_source.py b/tests/sources/manual/test_diva_manual_entry_source.py
new file mode 100644
index 00000000..5d3e7c89
--- /dev/null
+++ b/tests/sources/manual/test_diva_manual_entry_source.py
@@ -0,0 +1,23 @@
+from datetime import datetime
+
+import pytest
+
+from telliot_feeds.sources.manual.diva_manual_source import DivaManualSource
+from telliot_feeds.utils.input_timeout import InputTimeout
+
+
+@pytest.mark.asyncio
+async def test_source(capsys, monkeypatch):
+    """Test retrieving DIVA Protocol query response from user input."""
+
+    def mock_input(timeout=0, *args, **kwargs):
+        return "1234.0\n"
+
+    monkeypatch.setattr(InputTimeout, "__call__", mock_input)
+    source = DivaManualSource()
+    val, dt = await source.fetch_new_datapoint()
+    captured = capsys.readouterr()
+
+    assert isinstance(dt, datetime)
+    assert val == [1234.0, 1234.0]
+    assert "[1234000000000000000000, " in captured.out
diff --git a/tests/sources/test_rng_manual_input_source.py b/tests/sources/manual/test_rng_manual_input_source.py
similarity index 75%
rename from tests/sources/test_rng_manual_input_source.py
rename to tests/sources/manual/test_rng_manual_input_source.py
index fd1f5ce7..c4b3dbf2 100644
--- a/tests/sources/test_rng_manual_input_source.py
+++ b/tests/sources/manual/test_rng_manual_input_source.py
@@ -2,12 +2,15 @@
 
 import pytest
 
-from telliot_feeds.sources.manual_sources.tellor_rng_manual_source import TellorRNGManualInputSource
+from telliot_feeds.sources.manual.tellor_rng_manual_source import TellorRNGManualInputSource
 
 
 @pytest.mark.asyncio
 async def test_rng_valid_input(capsys):
-    with mock.patch("builtins.input", side_effect=["0x2b563420722cbcfc84857129bef775e0dc5f1401", ""]):
+    with mock.patch(
+        "telliot_feeds.utils.input_timeout.InputTimeout.__call__",
+        side_effect=["0x2b563420722cbcfc84857129bef775e0dc5f1401", ""],
+    ):
         result, _ = await TellorRNGManualInputSource().fetch_new_datapoint()
         expected = "TellorRNG value to be submitted on chain: 2b563420722cbcfc84857129bef775e0dc5f1401\nPress [ENTER] to continue"  # noqa: E501
         captured_output = capsys.readouterr()
@@ -17,7 +20,10 @@ async def test_rng_valid_input(capsys):
 
 @pytest.mark.asyncio
 async def test_rng_invalid_input(capsys):
-    with mock.patch("builtins.input", side_effect=["exit", "0x2b563420722cbcfc84857129bef775e0dc5f1401", ""]):
+    with mock.patch(
+        "telliot_feeds.utils.input_timeout.InputTimeout.__call__",
+        side_effect=["exit", "0x2b563420722cbcfc84857129bef775e0dc5f1401", ""],
+    ):
         _, _ = await TellorRNGManualInputSource().fetch_new_datapoint()
         expected = "Invalid input! Enter hex string value (32 byte size)"
         captured_output = capsys.readouterr()
@@ -26,7 +32,10 @@ async def test_rng_invalid_input(capsys):
 
 @pytest.mark.asyncio
 async def test_rng_non_input(capsys):
-    with mock.patch("builtins.input", side_effect=["", "0x2b563420722cbcfc84857129bef775e0dc5f1401", ""]):
+    with mock.patch(
+        "telliot_feeds.utils.input_timeout.InputTimeout.__call__",
+        side_effect=["", "0x2b563420722cbcfc84857129bef775e0dc5f1401", ""],
+    ):
         _, _ = await TellorRNGManualInputSource().fetch_new_datapoint()
         expected = "Invalid input! Not enough characters, Enter a hex string (example: 0x2b563420722cbcfc84857129bef775e0dc5f1401"  # noqa: E501
         captured_output = capsys.readouterr()
@@ -36,7 +45,7 @@ async def test_rng_non_input(capsys):
 @pytest.mark.asyncio
 async def test_rng_non_bytes32_input(capsys):
     with mock.patch(
-        "builtins.input",
+        "telliot_feeds.utils.input_timeout.InputTimeout.__call__",
         side_effect=[
             "0x00000000000000000000000000000000000000000000000000000000000000400000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000954656c6c6f72524e470000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000004d2",  # noqa: E501
             "0x2b563420722cbcfc84857129bef775e0dc5f1401",
diff --git a/tests/sources/test_spot_price_manual_source.py b/tests/sources/manual/test_spot_price_manual_source.py
similarity index 53%
rename from tests/sources/test_spot_price_manual_source.py
rename to tests/sources/manual/test_spot_price_manual_source.py
index b2201f98..acb7e3a9 100644
--- a/tests/sources/test_spot_price_manual_source.py
+++ b/tests/sources/manual/test_spot_price_manual_source.py
@@ -2,24 +2,22 @@
 
 import pytest
 
-from telliot_feeds.sources.manual_sources.spot_price_input_source import SpotPriceManualSource
+from telliot_feeds.sources.manual.spot_price_input_source import SpotPriceManualSource
 from telliot_feeds.utils.log import get_logger
 
 logger = get_logger(__name__)
 
 
 @pytest.mark.asyncio
-async def test_manual_spot_price_good_input(
-    monkeypatch,
-):
-    monkeypatch.setattr("builtins.input", lambda: "1234")
-    result, _ = await SpotPriceManualSource().fetch_new_datapoint()
-    assert result == 1234
+async def test_manual_spot_price_good_input():
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["1234", ""]):
+        result, _ = await SpotPriceManualSource().fetch_new_datapoint()
+        assert result == 1234
 
 
 @pytest.mark.asyncio
 async def test_manual_spot_price_bad_input(capsys):
-    with mock.patch("builtins.input", side_effect=["float", "5678", ""]):
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["float", "5678", ""]):
         result, _ = await SpotPriceManualSource().fetch_new_datapoint()
         expected = "Invalid input. Enter decimal value (float)."
         captured_output = capsys.readouterr()
diff --git a/tests/sources/test_string_query_manual_source.py b/tests/sources/manual/test_string_query_manual_source.py
similarity index 71%
rename from tests/sources/test_string_query_manual_source.py
rename to tests/sources/manual/test_string_query_manual_source.py
index 26a66e37..ed700b6c 100644
--- a/tests/sources/test_string_query_manual_source.py
+++ b/tests/sources/manual/test_string_query_manual_source.py
@@ -2,12 +2,12 @@
 
 import pytest
 
-from telliot_feeds.sources.manual_sources.string_query_manual_source import StringQueryManualSource
+from telliot_feeds.sources.manual.string_query_manual_source import StringQueryManualSource
 
 
 @pytest.mark.asyncio
 async def test_string_query_input(capsys):
-    with mock.patch("builtins.input", side_effect=["hello", ""]):
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["hello", ""]):
         result, _ = await StringQueryManualSource().fetch_new_datapoint()
         expected = "String query response to be submitted to oracle->: hello"
         captured_output = capsys.readouterr()
@@ -17,7 +17,7 @@ async def test_string_query_input(capsys):
 
 @pytest.mark.asyncio
 async def test_string_query_empty_input(capsys):
-    with mock.patch("builtins.input", side_effect=["", ""]):
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["", ""]):
         result, _ = await StringQueryManualSource().fetch_new_datapoint()
         expected = "String query response to be submitted to oracle->: "
         captured_output = capsys.readouterr()
diff --git a/tests/sources/test_twap_manual_source.py b/tests/sources/manual/test_twap_manual_source.py
similarity index 59%
rename from tests/sources/test_twap_manual_source.py
rename to tests/sources/manual/test_twap_manual_source.py
index 59dbb6e3..e4fa08f6 100644
--- a/tests/sources/test_twap_manual_source.py
+++ b/tests/sources/manual/test_twap_manual_source.py
@@ -2,21 +2,19 @@
 
 import pytest
 
-from telliot_feeds.sources.manual_sources.twap_manual_input_source import TWAPManualSource
+from telliot_feeds.sources.manual.twap_manual_input_source import TWAPManualSource
 
 
 @pytest.mark.asyncio
-async def test_manual_twap_positive_input(
-    monkeypatch,
-):
-    monkeypatch.setattr("builtins.input", lambda: "1234")
-    result, _ = await TWAPManualSource().fetch_new_datapoint()
-    assert result == 1234
+async def test_manual_twap_positive_input():
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["1234", ""]):
+        result, _ = await TWAPManualSource().fetch_new_datapoint()
+        assert result == 1234
 
 
 @pytest.mark.asyncio
 async def test_manual_twap_negative_input(capsys):
-    with mock.patch("builtins.input", side_effect=["-1234", "1234", ""]):
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["-1234", "1234", ""]):
         result, _ = await TWAPManualSource().fetch_new_datapoint()
         expected = "Invalid input. Number must greater than 0."
         captured_output = capsys.readouterr()
@@ -27,7 +25,7 @@ async def test_manual_twap_negative_input(capsys):
 
 @pytest.mark.asyncio
 async def test_manual_twap_non_number_input(capsys):
-    with mock.patch("builtins.input", side_effect=["hello", "1234", ""]):
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["hello", "1234", ""]):
         result, _ = await TWAPManualSource().fetch_new_datapoint()
         expected = "Invalid input. Enter a decimal value (float)."
         captured_output = capsys.readouterr()
diff --git a/tests/sources/manual/test_uspce_source.py b/tests/sources/manual/test_uspce_source.py
new file mode 100644
index 00000000..82c6e1ce
--- /dev/null
+++ b/tests/sources/manual/test_uspce_source.py
@@ -0,0 +1,20 @@
+from datetime import datetime
+from unittest import mock
+
+import pytest
+
+from telliot_feeds.sources.manual.uspce import USPCESource
+
+
+@pytest.mark.asyncio
+async def test_uspce_source():
+    """Test retrieving USPCE data from user input."""
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["1234.0", ""]):
+
+        ampl_source = USPCESource()
+
+        value, timestamp = await ampl_source.fetch_new_datapoint()
+
+        assert isinstance(value, float)
+        assert isinstance(timestamp, datetime)
+        assert value > 0
diff --git a/tests/sources/test_diva_manual_entry_source.py b/tests/sources/test_diva_manual_entry_source.py
deleted file mode 100644
index 387fd738..00000000
--- a/tests/sources/test_diva_manual_entry_source.py
+++ /dev/null
@@ -1,19 +0,0 @@
-from datetime import datetime
-from unittest import mock
-
-import pytest
-
-from telliot_feeds.sources.manual_sources.diva_manual_source import DivaManualSource
-
-
-@pytest.mark.asyncio
-async def test_source(capsys):
-    """Test retrieving DIVA Protocol query response from user input."""
-    with mock.patch("builtins.input", side_effect=["-1234", "1234", "5678", ""]):
-        source = DivaManualSource()
-        val, dt = await source.fetch_new_datapoint()
-        captured = capsys.readouterr()
-
-        assert isinstance(dt, datetime)
-        assert val == [1234, 5678]
-        assert "[1234000000000000000000, 5678000000000000524288]" in captured.out
diff --git a/tests/sources/test_uspce_source.py b/tests/sources/test_uspce_source.py
deleted file mode 100644
index 8ca84d99..00000000
--- a/tests/sources/test_uspce_source.py
+++ /dev/null
@@ -1,21 +0,0 @@
-from datetime import datetime
-
-import pytest
-
-from telliot_feeds.sources import uspce
-from telliot_feeds.sources.uspce import USPCESource
-
-
-@pytest.mark.asyncio
-async def test_uspce_source():
-    """Test retrieving USPCE data from user input."""
-    # Override Python built-in input method
-    uspce.input = lambda: "1234.1234"
-
-    ampl_source = USPCESource()
-
-    value, timestamp = await ampl_source.fetch_new_datapoint()
-
-    assert isinstance(value, float)
-    assert isinstance(timestamp, datetime)
-    assert value > 0
diff --git a/tests/utils/test_input_timeout.py b/tests/utils/test_input_timeout.py
new file mode 100644
index 00000000..23f8ce7d
--- /dev/null
+++ b/tests/utils/test_input_timeout.py
@@ -0,0 +1,30 @@
+import pytest
+
+from telliot_feeds.utils.input_timeout import input_timeout
+from telliot_feeds.utils.input_timeout import TimeoutOccurred
+
+
+# Fixure to prevent pytest from capturing the input and producing a different error.
+# Using "with capsys.disabled():" didn't work. It doesn't do the same as adding the "-s" flag, which is
+# what we need. But this fixture works:
+@pytest.fixture(scope="class")
+def suspend_capture(pytestconfig):
+    class suspend_guard:
+        def __init__(self):
+            self.capmanager = pytestconfig.pluginmanager.getplugin("capturemanager")
+
+        def __enter__(self):
+            self.capmanager.suspend_global_capture(in_=True)
+            pass
+
+        def __exit__(self, _1, _2, _3):
+            self.capmanager.resume_global_capture()
+
+    yield suspend_guard()
+
+
+def test_input_timeout(suspend_capture) -> None:
+    """Test input_timeout() function."""
+    with suspend_capture:
+        with pytest.raises(TimeoutOccurred):
+            input_timeout(prompt="sup", timeout=0)