diff --git a/interfaces/epics_pva/README.md b/interfaces/epics_pva/README.md
new file mode 100644
index 0000000..7ae17ce
--- /dev/null
+++ b/interfaces/epics_pva/README.md
@@ -0,0 +1,69 @@
+# EPICS PVAccess Interface for Badger
+Interface for setting and reading EPICS PVAccess PV values utilising the [p4p](https://github.com/mdavidsaver/p4p) library. 
+
+## Prerequisites
+
+## Usage
+Some general notes on usage:
+
+### Read Only Mode
+In some cases we may want to be testing our software stack on a live system but not want to be changing values. For these occasions we utilise a `read_only` flag configured in the constructor. Instead of actually setting values in the control system, the `set_values()` call will log a message to indicate the values that would have been set. `get_values()` will still return valid observations.
+
+```python
+interface = Interface(read_only=True)
+```
+
+### Retry on Timeout
+In the case of a TimeoutError on a PV when doing a `put` call, we retry the call twice, waiting one second between. If this second attempt to set the PV fails, an exception is thrown. 
+
+### Validation Functions
+In the case of some PVs we may want to validate that the PV has in fact been set, for example by checking an associated readback PV. For these cases, we pass a `validation_function` to the `set_value()` call. It is expected that this function would be configured as a `functools.partial` function and multiple validation functions can be passed to the `set_values()` call as a dictionary with the key as the set PV name. An example is provided below:
+
+Validation function:
+```python
+def wait_for_readback(
+    set_pv: str, set_value: float, readback_pv: str, tolerance: float, count_down: int, offset: float, context: p4p.client.thread.Context, timeout: int
+):
+    time_limit = deepcopy(count_down)
+    while count_down > 0:
+        _value = context.get(readback_pv, timeout=timeout)
+        if np.isclose(_value.real, set_value + offset,atol=tolerance):
+            return _value.real
+        else:
+            time.sleep(0.1)
+            count_down -= 0.1
+    raise ValueError(
+        f"PV {set_pv} (current {readback_pv}: {_value.real}) cannot reach expected value ({set_value + offset}) in designated time {time_limit}!"
+    )
+```
+
+Expected usage:
+```python
+class Environment(badger.Environment):
+    ...
+    _configs: ClassVar[Dict[str, partial]] = {
+        "EXAMPLE:PV:SET": partial(
+            wait_for_readback,
+            readback_pv="EXAMPLE:PV:READ",
+            tolerance=0.05,
+            count_down=15,
+            offset=0,
+        ),
+    }
+    def set_variables(self, variable_inputs):
+        self.interface.set_values(
+            variable_inputs,
+            configs=self._configs,
+        )
+```
+
+
+### Parallel Execution
+When using these validation function, setting values on multiple PVs may cause long delays, espcially if the update rate of the READ PV is slow. Therefore we provide the option to run all of the `set` commands in parallel. This can be configured using the `parallel` flag in the constructor:
+```python
+interface = Interface(parallel=False)
+```
+
+
+## Testing
+Initial tests for the interface are available in the `tests.py` file and can be run locally using `pytest tests.py` or `pytest --cov=. tests.py` from within the `epics_pva` directory.
\ No newline at end of file
diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py
new file mode 100644
index 0000000..1cd32fe
--- /dev/null
+++ b/interfaces/epics_pva/__init__.py
@@ -0,0 +1,166 @@
+import logging
+import multiprocessing as mp
+import time
+from copy import deepcopy
+from functools import partial
+from typing import Any, Dict, List
+
+import numpy as np
+from badger import interface
+from joblib import Parallel, delayed
+from p4p.client.thread import Cancelled, Context, Disconnected, RemoteError, TimeoutError
+from pydantic import Field
+
+
+def timeit(func):
+    def wrapper_timeit(*args, **kwargs):
+        start_time = time.time()
+        func(*args, **kwargs)
+        end_time = time.time()
+        logging.debug(f"total set time: {end_time - start_time:.5f}s")
+
+    return wrapper_timeit
+
+
+class Interface(interface.Interface):
+    """Concrete interface for interacting with EPICS PVAccess PVs"""
+
+    name: str = "epics_pva"
+    context_str: str = "pva"
+    timeout: float = Field(default=3.0, description="Number of seconds to try connecting to a PV")
+    parallel: bool = Field(
+        default=False,
+        description="Flag indicating whether all variables should be set at once or in series.",
+    )
+    read_only: bool = Field(
+        default=False,
+        description="Flag to indicate whether the interface should allow values to be set or not.",
+    )
+
+    def get_default_params(self) -> dict:
+        return {
+            "context": self.context_str,
+            "read-only": self.read_only,
+        }
+
+    def get_value(self, channel: str):
+        context = Context(self.context_str)
+
+        result = context.get(channel, throw=False)
+        if isinstance(result, (Disconnected, TimeoutError, RemoteError, Cancelled)):
+            logging.warning("Could not retrieve value of %s due to %s", channel, type(result).__name__)
+            result = np.nan
+        else:
+            result = result.real
+        context.close()
+        return result
+
+    def get_values(self, channels: List[str]) -> Dict[str, float]:
+        if isinstance(channels, str):
+            channels = [channels]
+        context = Context(self.context_str)
+        book = {}
+
+        results = context.get(channels, throw=False)
+        failures = []
+
+        for i, result in enumerate(results):
+            if isinstance(result, (Disconnected, TimeoutError, RemoteError, Cancelled)):
+                failures.append(channels[i])
+                logging.warning(
+                    "Could not retrieve value of %s due to %s. Retrying individually.",
+                    channels[i],
+                    type(result).__name__,
+                )
+            else:
+                book[channels[i]] = result.real
+
+        # then if there are any failures, retry the failed ones in the same way
+        if len(failures) >= 1:
+            results = context.get(failures, throw=False)
+
+            for i, result in enumerate(results):
+                if isinstance(result, (Disconnected, TimeoutError, RemoteError, Cancelled)):
+                    logging.warning("Could not retrieve value of %s due to %s.", failures[i], type(result).__name__)
+                    book[failures[i]] = np.nan
+                else:
+                    book[failures[i]] = result.real
+        context.close()
+        return book
+
+    def _put(self, context, channel, value):
+        success = False
+        try:
+            result = context.put(channel, value, timeout=self.timeout, throw=False)
+        except TypeError:
+            value = value.item()
+            result = context.put(channel, value, timeout=self.timeout, throw=False)
+        if result is None:
+            logging.debug(
+                "Put value %d to %s",
+                value,
+                channel,
+            )
+            success = True
+        else:
+            logging.info("Could not put value %d to %s due to %s. Retrying...", value, channel, type(result).__name__)
+            # retry after a short sleep
+            time.sleep(1)
+            result = context.put(channel, value, timeout=self.timeout, throw=False)
+            if result is None:
+                logging.debug(
+                    "Put value %d to %s",
+                    value,
+                    channel,
+                )
+                success = True
+            else:
+                logging.warning(
+                    "Could not put value %d to %s (second attempt) due to %s after a 1 second delay.",
+                    value,
+                    channel,
+                    type(result).__name__,
+                )
+        return success
+
+    def set_value(self, channel: str, value, validation_function=None):
+        if not self.read_only:
+            # for parallel to work, context has to be made and closed within the function
+            context = Context(self.context_str)
+            # always put the value to the set PV
+
+            success = self._put(context, channel, value)
+            # if we weren't successful in setting the PV, there's no need to validate
+            # against the readback
+            if success and validation_function is not None:
+                try:
+                    validation_function(
+                        set_pv=channel,
+                        set_value=value,
+                        context=context,
+                        timeout=self.timeout,
+                    )
+                except (ValueError, TimeoutError) as e:
+                    logging.warning(e)
+            context.close()
+        else:
+            logging.info(
+                "Interface is set to read-only mode, cannot set value %s to %s",
+                value,
+                channel,
+            )
+
+    @timeit
+    def set_values(self, channel_inputs: Dict[str, Any], configs: Dict[str, partial]):
+        if not self.read_only:
+            if self.parallel:
+                Parallel(n_jobs=mp.cpu_count())(
+                    delayed(self.set_value)(channel, value, configs.get(channel))
+                    for channel, value in channel_inputs.items()
+                )
+
+            else:
+                for channel, value in channel_inputs.items():
+                    self.set_value(channel, value, configs.get(channel))
+        else:
+            logging.info("Interface is set to read-only mode, cannot set %s", channel_inputs)
diff --git a/interfaces/epics_pva/configs.yaml b/interfaces/epics_pva/configs.yaml
new file mode 100644
index 0000000..79e5fb7
--- /dev/null
+++ b/interfaces/epics_pva/configs.yaml
@@ -0,0 +1,8 @@
+---
+name: epics_pva
+version: "0.1"
+dependencies:
+  - numpy
+  - badger-opt
+  - p4p
+  - joblib
\ No newline at end of file
diff --git a/interfaces/epics_pva/tests.py b/interfaces/epics_pva/tests.py
new file mode 100644
index 0000000..c582752
--- /dev/null
+++ b/interfaces/epics_pva/tests.py
@@ -0,0 +1,257 @@
+import sys
+from pathlib import Path
+
+current_file = Path(__file__)
+root_dir = current_file.parents[1]
+sys.path.insert(0, str(root_dir))
+import logging
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+import pytest
+from p4p.client.thread import TimeoutError
+
+channels = ["test::channel:1", "test::channel:2"]
+values = [5.0, 6.0]
+
+from epics_pva import Interface
+
+
+class MockRaw:
+    def __init__(self, value) -> None:
+        self.value = value
+
+
+class p4pValue:
+    """Mock class designed to replicate the p4p Value class returned from `get` calls"""
+
+    def __init__(self, value) -> None:
+        self.raw = MockRaw(value)
+
+    @property
+    def real(self):
+        return self.raw.value
+
+
+@pytest.fixture
+def mock_context():
+    with patch("epics_pva.Context") as ctx:
+        yield ctx
+
+
+class TestEPICSInterface:
+    def test_epics_get_values(self, mock_context):
+        mock_context.return_value.get.return_value = [
+            p4pValue(values[0]),
+            p4pValue(values[1]),
+        ]
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+        result = interface.get_values(channels)
+        assert result == {
+            "test::channel:1": values[0],
+            "test::channel:2": values[1],
+        }
+        mock_context.return_value.close.assert_called_once_with()
+
+    def test_epics_get_values_timeout_retry(self, caplog, mock_context):
+        # we get a timeout error first, then we get the individual values
+        mock_context.return_value.get.side_effect = [
+            [
+                TimeoutError("original error"),
+                p4pValue(values[1]),
+            ],
+            [p4pValue(values[0])],
+        ]
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+        result = interface.get_values(channels)
+
+        # check the mock calls
+        assert len(mock_context.return_value.get.call_args_list) == 2
+
+        # first check the returned values
+        assert result["test::channel:1"] == values[0]
+        assert result["test::channel:2"] == values[1]
+
+        # then check the logs
+        assert len(caplog.records) == 1
+        assert len(mock_context.return_value.close.call_args_list) == 1
+
+    def test_epics_get_values_timeout_retry_fails(self, caplog, mock_context):
+        # we get a timeout error first, then we get the individual values
+        mock_context.return_value.get.side_effect = [
+            [
+                TimeoutError("original error"),
+                p4pValue(values[1]),
+            ],
+            [TimeoutError("retry")],
+        ]
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+        result = interface.get_values(channels)
+
+        # check the mock calls
+        assert len(mock_context.return_value.get.call_args_list) == 2
+
+        # first check the returned values
+        assert np.isnan(result["test::channel:1"])
+        assert result["test::channel:2"] == values[1]
+
+        # then check the logs
+        assert len(caplog.records) == 2
+        assert len(mock_context.return_value.close.call_args_list) == 1
+
+    @pytest.mark.parametrize(
+        "test_input",
+        [
+            ([value for value in values]),
+            ([np.array(value) for value in values]),
+        ],
+    )
+    def test_epics_set_values_no_validation(self, mock_context, test_input):
+        mock_context.return_value.put = MagicMock(return_value=None)
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+        interface.set_values(dict(zip(channels, test_input)), configs={})
+
+        assert mock_context.return_value.put.call_args_list[0][0] == (
+            channels[0],
+            values[0],
+        )
+        assert mock_context.return_value.put.call_args_list[1][0] == (
+            channels[1],
+            values[1],
+        )
+        assert len(mock_context.return_value.close.call_args_list) == len(channels)
+
+    def test_epics_set_values_no_validation_timeout(self, caplog, mock_context):
+        caplog.set_level(logging.INFO)
+        mock_context.return_value.put = MagicMock(return_value=TimeoutError("timeout"))
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+        interface.set_values(dict(zip(channels, [value for value in values])), configs={})
+
+        # here we check that the retry works
+        assert len(mock_context.return_value.put.call_args_list) == 4
+        assert mock_context.return_value.put.call_args_list[0][0] == (
+            channels[0],
+            values[0],
+        )
+        assert mock_context.return_value.put.call_args_list[1][0] == (
+            channels[0],
+            values[0],
+        )
+        assert len(mock_context.return_value.close.call_args_list) == len(channels)
+
+        assert len(caplog.records) == 4
+        assert str(caplog.records[0].getMessage()).endswith("Retrying...")
+        assert str(caplog.records[-1].getMessage()).endswith("after a 1 second delay.")
+
+    def test_epics_set_values_with_validation(self, mock_context):
+        mock_context.return_value.put = MagicMock(return_value=None)
+
+        configs = {
+            # in reality this would be a partial function instead of a MagicMock
+            "test::channel:1": MagicMock()
+        }
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+        interface.set_values(dict(zip(channels, values)), configs=configs)
+        # check that the correct put values were called
+        assert mock_context.return_value.put.call_args_list[0][0] == (
+            channels[0],
+            values[0],
+        )
+        assert mock_context.return_value.put.call_args_list[1][0] == (
+            channels[1],
+            values[1],
+        )
+        # check the validation function was called as expected
+        configs["test::channel:1"].assert_called_once_with(
+            set_pv="test::channel:1",
+            set_value=5.0,
+            context=mock_context.return_value,
+            timeout=1,
+        )
+        assert len(configs["test::channel:1"].call_args_list) == 1
+        # check that the context is closed correctly()
+        assert len(mock_context.return_value.close.call_args_list) == len(channels)
+
+    @pytest.mark.parametrize(
+        "error",
+        [(ValueError("validation function failed")), (TimeoutError("timeout error"))],
+    )
+    def test_epics_set_values_with_validation_errors(self, caplog, mock_context, error):
+        mock_context.return_value.put = MagicMock(return_value=None)
+
+        configs = {
+            # in reality this would be a partial function instead of a MagicMock
+            "test::channel:1": MagicMock(side_effect=error)
+        }
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+        interface.set_values(dict(zip(channels, values)), configs=configs)
+
+        assert len(caplog.records) == 1
+        assert caplog.records[0].getMessage() == str(error)
+
+    @patch("time.sleep")
+    def test_epics_set_values_put_timeout_retry_fails(self, mock_time, caplog, mock_context):
+        # here we test whether the put is called twice
+        caplog.set_level = logging.INFO
+        mock_context.return_value.put = MagicMock(return_value=TimeoutError("timeout error"))
+
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+
+        interface.set_values({channels[0]: values[0]}, configs={})
+
+        # check that the put was attempted twice
+        assert len(mock_context.return_value.put.call_args_list) == 2
+        # make sure we waited in between
+        mock_time.assert_called_once_with(1)
+
+        # finally we log a message if both attempts were unsuccessful
+        assert len(caplog.records) == 1
+        assert str(caplog.records[0].getMessage()).endswith("after a 1 second delay.")
+
+    @pytest.mark.skip(reason="Unsure how to test when side_effect will raise TimeoutError instead of return")
+    @patch("time.sleep")
+    def test_epics_set_values_put_timeout_retry_successful(self, mock_time, caplog, mock_context):
+        mock_context.return_value.put = MagicMock(side_effect=[TimeoutError("timeout error"), None])
+
+        interface = Interface(timeout=1, parallel=False, read_only=False)
+
+        with caplog.at_level(logging.DEBUG):
+            interface.set_values({channels[0]: values[0]}, configs={})
+
+        # check that the put was attempted twice
+        assert len(mock_context.return_value.put.call_args_list) == 2
+        # make sure we waited in between
+        mock_time.assert_called_once_with(1)
+
+        # finally we log a message if both attempts were unsuccessful and we should
+        # also get a message about how long it takes to set the values
+        assert len(caplog.records) == 1
+        assert caplog.records[0].getMessage() == "Retrying..."
+
+    def test_epics_set_values_read_only(self, caplog, mock_context):
+        caplog.set_level(logging.INFO)
+        mock_context.return_value.put = MagicMock()
+        interface = Interface(timeout=1, parallel=False, read_only=True)
+        set_vals = dict(zip(channels, values))
+        with caplog.at_level(logging.INFO):
+            interface.set_values(set_vals, configs={})
+
+        # check the log message is sent correctly and that context.put is not called
+        mock_context.return_value.put.assert_not_called()
+        assert len(caplog.records) == 1
+        assert caplog.records[0].getMessage() == f"Interface is set to read-only mode, cannot set {set_vals}"
+
+    def test_epics_set_value_read_only(self, caplog, mock_context):
+        caplog.set_level(logging.INFO)
+        mock_context.return_value.put = MagicMock()
+        interface = Interface(timeout=1, parallel=False, read_only=True)
+        with caplog.at_level(logging.INFO):
+            interface.set_value(channels[0], values[0])
+
+        # check the log message is sent correctly and that context.put is not called
+        mock_context.return_value.put.assert_not_called()
+        assert len(caplog.records) == 1
+        assert (
+            caplog.records[0].getMessage()
+            == f"Interface is set to read-only mode, cannot set value {values[0]} to {channels[0]}"
+        )
diff --git a/interfaces/mock/README.md b/interfaces/mock/README.md
new file mode 100644
index 0000000..bde9fc4
--- /dev/null
+++ b/interfaces/mock/README.md
@@ -0,0 +1,5 @@
+# Mock Interface for Badger
+
+## Prerequisites
+
+## Usage
diff --git a/interfaces/mock/__init__.py b/interfaces/mock/__init__.py
new file mode 100644
index 0000000..c145ade
--- /dev/null
+++ b/interfaces/mock/__init__.py
@@ -0,0 +1,26 @@
+from badger import interface
+
+from numpy import random
+
+
+class Interface(interface.Interface):
+    name = "mock"
+
+    @staticmethod
+    def get_default_params():
+        return None
+
+    def get_value(self, channel: str):
+        print("Called get_value for channel: {}.".format(channel))
+        return random.random()
+
+    def set_value(self, channel: str, value):
+        print("Called set_value for channel: {}, with value: {}".format(channel, value))
+
+    def get_values(self, channels: list):
+        return {channel: self.get_value(channel) for channel in channels}
+
+    def set_values(self, channel_inputs: dict, *args, **kwargs):
+        print(f"Called set_values with args:{args}, kwargs:{kwargs}")
+        for channel, val in channel_inputs.items():
+            self.set_value(channel, val)
diff --git a/interfaces/mock/configs.yaml b/interfaces/mock/configs.yaml
new file mode 100644
index 0000000..3242482
--- /dev/null
+++ b/interfaces/mock/configs.yaml
@@ -0,0 +1,6 @@
+---
+name: mock
+version: "0.1"
+dependencies:
+  - badger-opt
+  - numpy