Skip to content

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

@alexfournierahizoune

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

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions