Skip to content

getting command_error ERROR for simple instrument query... #89

Description

@alexfournierahizoune

Can someone explain to me why this simple test does not seem to work?

pyvisa-sim version: 0.5.1

error:

    def test_get_identification():
        identifier = get_osp320_instrument().get_identification()
>       assert identifier == "OSP320, 12345, 00000001, 23.1.1/3"
E       AssertionError: assert 'ERROR' == 'OSP320, 1234...001, 23.1.1/3'
E         - OSP320, 12345, 00000001, 23.1.1/3
E         + null_response

tests/drivers/switch/rs/test_osp320.py:64: AssertionError

test_file.py

def get_osp320_instrument(
    config: OSP320Configuration = OSP320Configuration(
        read_termination="\n" # somehow need to add this even in TCPIP::INSTR otherwise a VisaIOError gets triggered...
    )
) -> OSP320Instrument:
    MockInstrument = OSP320Instrument
    # Use mocked resource manager
    MockInstrument.resource_manager = "<path>/test_file.yaml@sim"
    # MockInstrument.connection.read_termination = "\n"
    return MockInstrument(
        connection_string="TCPIP::localhost::inst0::INSTR",
        config=config,
    )

# ...

def test_get_identification():
    identifier = get_osp320_instrument().get_identification()
    assert identifier == "OSP320, 12345, 00000001, 23.1.1/3"

test_file.yaml

spec: "1.1"
devices:
  OSP320:
    eom:
      TCPIP INSTR:
        q: "\r\n"
        r: "\n"
      TCPIP SOCKET:
        q: "\r\n"
        r: "\n"
    dialogues:
      - q: "*IDN?"
        r: "OSP320, 12345, 00000001, 23.1.1/3"
      - q: "*OPC?"
        r: "1"
      - q: "*CLS"
    error:
      response:
        command_error: ERROR
      status_register:
        - q: "*ESR?"
          command_error: 32
          query_error: 4
    properties:
      status:
        default: 0
        getter:
          q: "*STB?"
          r: "{:d}"
        setter:
          q: "*STB {:d}"
        specs:
          type: int
      system_error:
        default: ""
        getter:
          q: "SYSTem:ERRor:NEXT?"
          r: "{:s}"
        setter:
          q: "SYSTem:ERRor:NEXT {:s}"

resources:
  TCPIP::localhost::inst0::INSTR:
    device: OSP320
  TCPIP::localhost::1001::SOCKET:
    device: OSP320

instrument.py

class OSP320Configuration(BaseModel):
    default_timeout: Optional[int] = None
    read_termination: Optional[str] = None


class OSP320Instrument(Instrument):
    name = "cka.instrument.rs.switch.osp320.v1"
    config: OSP320Configuration
    resource_manager: str = "@py"
    connection: pyvisa.resources.Resource

    @validate_call
    def __init__(
        self,
        connection_string: str,
        config: OSP320Configuration,
    ) -> None:
        self.config = config
        logger.info("Connecting to: `%s`", connection_string)
        rm = pyvisa.ResourceManager(self.resource_manager)
        self.connection = rm.open_resource(connection_string)

        if "SOCKET" in connection_string:
            self.connection.read_termination = "\n"
        elif self.config.read_termination is not None:
            self.connection.read_termination = self.config.read_termination

        self.connection.write("*CLS")
        self.connection.query("*OPC?")
        logger.info("Validated the connection!")
        self.write("CONFigure:COMPatible 0") # do not accept legacy commands (OSP1xx)

        if self.config.default_timeout is not None:
            self.connection.timeout = self.config.default_timeout
        super().__init__()

    def __exit__(self, *args):
        self.connection.close()

    def write(self, command: str):
        return self.connection.write(command)

    def try_write(self, command: str):
        data = self.connection.write(command)
        logger.info("OSP320 write: %s", command)
        if int(self.connection.query("*STB?")) != 0:
            try:
                error_messages = self.connection.query("SYSTEM:ERRor:NEXT?")
                logger.error("Error message: `%s`", error_messages)
                raise RuntimeError(f"Failed to write command: `{command}`. {error_messages}")
            finally:
                self.connection.write("*CLS")
        return data

    def query(self, command: str, timeout: Optional[int] = None):
        old_timeout = None
        logger.info("OSP320 query: %s", command)
        if timeout is not None:
            old_timeout = self.connection.timeout
            self.connection.timeout = timeout
        try:
            return self.connection.query(command)
        finally:
            if old_timeout is not None:
                self.connection.timeout = old_timeout

    @logged
    def get_identification(self):
        return self.query("*IDN?")
# ...

What am I missing?

Thanks

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions