dialects: (stream) remove abstract stream read and write operations

xdsl/dialects/memref_stream.py

@@ -46,6 +46,7 @@
     opt_prop_def,
     prop_def,
     region_def,
+    result_def,
     traits_def,
     var_operand_def,
 )
@@ -187,14 +188,44 @@ def offsets(self) -> tuple[tuple[int, ...], ...]:
 
 
 @irdl_op_definition
-class ReadOp(stream.ReadOperation):
+class ReadOp(IRDLOperation):
     name = "memref_stream.read"
 
+    T: ClassVar = VarConstraint("T", AnyAttr())
+
+    stream = operand_def(stream.ReadableStreamType.constr(T))
+    res = result_def(T)
+
+    assembly_format = "`from` $stream attr-dict `:` type($res)"
+
+    def __init__(self, stream_val: SSAValue, result_type: Attribute | None = None):
+        if result_type is None:
+            assert isinstance(stream_type := stream_val.type, stream.ReadableStreamType)
+            stream_type = cast(stream.ReadableStreamType[Attribute], stream_type)
+            result_type = stream_type.element_type
+        super().__init__(operands=[stream_val], result_types=[result_type])
+
+    def assembly_line(self) -> str | None:
+        return None
+
 
 @irdl_op_definition
-class WriteOp(stream.WriteOperation):
+class WriteOp(IRDLOperation):
     name = "memref_stream.write"
 
+    T: ClassVar = VarConstraint("T", AnyAttr())
+
+    value = operand_def(T)
+    stream = operand_def(stream.WritableStreamType.constr(T))
+
+    assembly_format = "$value `to` $stream attr-dict `:` type($value)"
+
+    def __init__(self, value: SSAValue, stream: SSAValue):
+        super().__init__(operands=[value, stream])
+
+    def assembly_line(self) -> str | None:
+        return None
+
 
 @irdl_op_definition
 class StreamingRegionOp(IRDLOperation):

xdsl/dialects/riscv_snitch.py

@@ -40,6 +40,7 @@
 )
 from xdsl.ir import Attribute, Block, Dialect, Operation, Region, SSAValue
 from xdsl.irdl import (
+    AnyAttr,
     VarConstraint,
     attr_def,
     base,
@@ -155,17 +156,41 @@ def assembly_line(self) -> str | None:
 
 
 @irdl_op_definition
-class ReadOp(stream.ReadOperation, RISCVAsmOperation):
+class ReadOp(RISCVAsmOperation):
     name = "riscv_snitch.read"
 
+    T: ClassVar = VarConstraint("T", AnyAttr())
+
+    stream = operand_def(stream.ReadableStreamType.constr(T))
+    res = result_def(T)
+
+    assembly_format = "`from` $stream attr-dict `:` type($res)"
+
+    def __init__(self, stream_val: SSAValue, result_type: Attribute | None = None):
+        if result_type is None:
+            assert isinstance(stream_type := stream_val.type, stream.ReadableStreamType)
+            stream_type = cast(stream.ReadableStreamType[Attribute], stream_type)
+            result_type = stream_type.element_type
+        super().__init__(operands=[stream_val], result_types=[result_type])
+
     def assembly_line(self) -> str | None:
         return None
 
 
 @irdl_op_definition
-class WriteOp(stream.WriteOperation, RISCVAsmOperation):
+class WriteOp(RISCVAsmOperation):
     name = "riscv_snitch.write"
 
+    T: ClassVar = VarConstraint("T", AnyAttr())
+
+    value = operand_def(T)
+    stream = operand_def(stream.WritableStreamType.constr(T))
+
+    assembly_format = "$value `to` $stream attr-dict `:` type($value)"
+
+    def __init__(self, value: SSAValue, stream: SSAValue):
+        super().__init__(operands=[value, stream])
+
     def assembly_line(self) -> str | None:
         return None
 

xdsl/dialects/stream.py

@@ -1,32 +1,21 @@
 from __future__ import annotations
 
-import abc
-from typing import ClassVar, Generic, TypeVar, cast
-
-from typing_extensions import Self
+from typing import Generic, TypeVar
 
 from xdsl.dialects.builtin import ContainerType
 from xdsl.ir import (
     Attribute,
     Dialect,
     ParametrizedAttribute,
-    SSAValue,
     TypeAttribute,
 )
 from xdsl.irdl import (
-    AnyAttr,
     BaseAttr,
     GenericAttrConstraint,
-    IRDLOperation,
     ParamAttrConstraint,
     ParameterDef,
-    VarConstraint,
     irdl_attr_definition,
-    operand_def,
-    result_def,
 )
-from xdsl.parser import Parser
-from xdsl.printer import Printer
 
 _StreamTypeElement = TypeVar("_StreamTypeElement", bound=Attribute, covariant=True)
 _StreamTypeElementConstrT = TypeVar("_StreamTypeElementConstrT", bound=Attribute)
@@ -91,74 +80,6 @@ def constr(
 )
 
 
-class ReadOperation(IRDLOperation, abc.ABC):
-    """
-    Abstract base class for operations that read from a stream.
-    """
-
-    T: ClassVar = VarConstraint("T", AnyAttr())
-
-    stream = operand_def(ReadableStreamType.constr(T))
-    res = result_def(T)
-
-    def __init__(self, stream: SSAValue, result_type: Attribute | None = None):
-        if result_type is None:
-            assert isinstance(stream_type := stream.type, ReadableStreamType)
-            stream_type = cast(ReadableStreamType[Attribute], stream_type)
-            result_type = stream_type.element_type
-        super().__init__(operands=[stream], result_types=[result_type])
-
-    @classmethod
-    def parse(cls, parser: Parser) -> Self:
-        parser.parse_characters("from")
-        unresolved = parser.parse_unresolved_operand()
-        parser.parse_punctuation(":")
-        result_type = parser.parse_attribute()
-        resolved = parser.resolve_operand(unresolved, ReadableStreamType(result_type))
-        return cls(resolved, result_type)
-
-    def print(self, printer: Printer):
-        printer.print_string(" from ")
-        printer.print(self.stream)
-        printer.print_string(" : ")
-        printer.print_attribute(self.res.type)
-
-
-class WriteOperation(IRDLOperation, abc.ABC):
-    """
-    Abstract base class for operations that write to a stream.
-    """
-
-    T: ClassVar = VarConstraint("T", AnyAttr())
-
-    value = operand_def(T)
-    stream = operand_def(WritableStreamType.constr(T))
-
-    def __init__(self, value: SSAValue, stream: SSAValue):
-        super().__init__(operands=[value, stream])
-
-    @classmethod
-    def parse(cls, parser: Parser) -> Self:
-        unresolved_value = parser.parse_unresolved_operand()
-        parser.parse_characters("to")
-        unresolved_stream = parser.parse_unresolved_operand()
-        parser.parse_punctuation(":")
-        result_type = parser.parse_attribute()
-        resolved_value = parser.resolve_operand(unresolved_value, result_type)
-        resolved_stream = parser.resolve_operand(
-            unresolved_stream, WritableStreamType(result_type)
-        )
-        return cls(resolved_value, resolved_stream)
-
-    def print(self, printer: Printer):
-        printer.print_string(" ")
-        printer.print_ssa_value(self.value)
-        printer.print_string(" to ")
-        printer.print_ssa_value(self.stream)
-        printer.print_string(" : ")
-        printer.print_attribute(self.value.type)
-
-
 Stream = Dialect(
     "stream",
     [],