Admin

Admin for property package.

ParcelAdmin

Bases: ModelAdmin

Admin page for Parcel model.

run_patch_parcel_source

run_patch_parcel_source(request, queryset)

Admin action to patch source in parcels.

Source code in django_project/property/admin.py

@admin.action(
    description="Patch source in parcels"
)
def run_patch_parcel_source(self, request, queryset):
    """Admin action to patch source in parcels."""
    from frontend.tasks.parcel import (
        patch_parcel_sources
    )
    patch_parcel_sources.delay()

ParcelTypeAdmin

Bases: ModelAdmin

Admin page for ParcelType model.

PropertyAdmin

Bases: ModelAdmin

Admin page for Property model.

generate_spatial_filters_for_properties

generate_spatial_filters_for_properties(request, queryset)

Admin action to generate spatial filter data for selected properties.

Source code in django_project/property/admin.py

@admin.action(
    description="Generate spatial filters for selected properties"
)
def generate_spatial_filters_for_properties(self, request, queryset):
    """Admin action to generate spatial filter data for
        selected properties."""
    from property.tasks.generate_spatial_filter import (
        generate_spatial_filter_task
    )
    for property_obj in queryset:
        property_obj.spatialdatamodel_set.all().delete()
        generate_spatial_filter_task.delay(
            property_obj.id
        )

run_check_overlaps

run_check_overlaps(modeladmin, request, queryset)

Admin action to check for overlapping properties.

Source code in django_project/property/admin.py

@admin.action(
    description="Run check overlaps"
)
def run_check_overlaps(modeladmin, request, queryset):
    """Admin action to check for overlapping properties."""
    from property.tasks.check_overlaps import (
        property_check_overlaps_each_other
    )
    property_check_overlaps_each_other.delay()
    modeladmin.message_user(
        request,
        'Job check for overlapping properties will be run in background!',
        messages.SUCCESS
    )

run_patch_property_centroid

run_patch_property_centroid(request, queryset)

Admin action to patch property without centroid.

Source code in django_project/property/admin.py

@admin.action(
    description="Patch centroid field in properties"
)
def run_patch_property_centroid(self, request, queryset):
    """Admin action to patch property without centroid."""
    from property.tasks.generate_property_centroid import (
        generate_property_centroid
    )
    generate_property_centroid.delay()

run_patch_property_province

run_patch_property_province(request, queryset)

Admin action to patch province in properties.

Source code in django_project/property/admin.py

@admin.action(
    description="Patch province in properties"
)
def run_patch_property_province(self, request, queryset):
    """Admin action to patch province in properties."""
    from frontend.tasks.patch_province import (
        patch_province_in_properties
    )
    patch_province_in_properties.delay()

PropertyOverlapsAdmin

Bases: ModelAdmin

Admin page for PropertyOverlaps model.

resolve_overlaps

resolve_overlaps(modeladmin, request, queryset)

Admin action to manually resolve the overlap record.

Source code in django_project/property/admin.py

@admin.action(
    description="Resolve overlaps record"
)
def resolve_overlaps(modeladmin, request, queryset):
    """Admin action to manually resolve the overlap record."""
    for overlap in queryset:
        overlap.resolved = True
        overlap.resolved_at = timezone.now()
        overlap.save(update_fields=['resolved', 'resolved_at'])
    modeladmin.message_user(
        request,
        (
            f'{queryset.count()} records has successfully been '
            'marked as resolved!'
        ),
        messages.SUCCESS
    )

PropertyTypeAdmin

Bases: ModelAdmin

Admin page for PropertyType model.

ProvinceAdmin

Bases: ModelAdmin

Admin page for Province model.

Factories

ParcelFactory

Bases: DjangoModelFactory

Factory for Parcel.

ParcelTypeFactory

Bases: DjangoModelFactory

Factory for ParcelType.

PropertyFactory

Bases: DjangoModelFactory

Property factory.

PropertyTypeFactory

Bases: DjangoModelFactory

Factory for PropertyType model.

ProvinceFactory

Bases: DjangoModelFactory

Factory for Province.

Models

Parcel

Bases: Model

Parcel model.

ParcelType

Bases: Model

Parcel type model.

Property

Bases: Model

Property model.

PropertyOverlaps

Bases: Model

Model to store property that overlaps with each other.

Test Case

ParcelTestCase

Bases: TestCase

Parcel test case.

test_create_parcel

test_create_parcel()

Test create parcel.

Source code in django_project/property/tests/test_property_models.py

def test_create_parcel(self):
    """Test create parcel."""
    self.assertTrue(isinstance(self.parcel, Parcel))
    self.assertEqual(Parcel.objects.count(), 1)
    self.assertEqual(self.parcel.sg_number, 'SG_0')
    self.assertEqual(self.parcel.farm_number, 0)

test_delete_parcel

test_delete_parcel()

Test delete parcel.

Source code in django_project/property/tests/test_property_models.py

def test_delete_parcel(self):
    """Test delete parcel."""
    self.parcel.delete()
    self.assertEqual(Parcel.objects.count(), 0)

test_non_unique_parcel_sg_number_constraint

test_non_unique_parcel_sg_number_constraint()

Test unique parcel sg number constraint.

Source code in django_project/property/tests/test_property_models.py

def test_non_unique_parcel_sg_number_constraint(self):
    """Test unique parcel sg number constraint."""
    ParcelFactory(sg_number='SG_1')

test_update_parcel

test_update_parcel()

Test update parcel.

Source code in django_project/property/tests/test_property_models.py

def test_update_parcel(self):
    """Test update parcel."""
    self.parcel.sg_number = 'SG_1'
    self.parcel.save()
    self.assertEqual(Parcel.objects.get(id=self.parcel.id).sg_number, 'SG_1')

ParcelTypeTestCase

Bases: TestCase

Parcel type test case

test_create_parcel_type

test_create_parcel_type()

Test create parcel types

Source code in django_project/property/tests/test_property_models.py

def test_create_parcel_type(self):
    """Test create parcel types """
    self.assertTrue(isinstance(self.parcel_type, ParcelType))
    self.assertEqual(ParcelType.objects.count(), 1)

test_delete_parcel_type

test_delete_parcel_type()

Test delete parcel type.

Source code in django_project/property/tests/test_property_models.py

def test_delete_parcel_type(self):
    """Test delete parcel type."""
    self.parcel_type.delete()
    self.assertEqual(ParcelType.objects.count(), 0)

test_unique_parcel_type_name_constraint

test_unique_parcel_type_name_constraint()

Test unique parcel type name constraint.

Source code in django_project/property/tests/test_property_models.py

def test_unique_parcel_type_name_constraint(self):
    """Test unique parcel type name constraint."""
    with self.assertRaises(Exception) as raised:
        ParcelTypeFactory(name='ParcelType_1')
        self.assertEqual(IntegrityError, type(raised.exception))

test_update_parcel_type

test_update_parcel_type()

Test update parcel type.

Source code in django_project/property/tests/test_property_models.py

def test_update_parcel_type(self):
    """Test update parcel type."""
    self.parcel_type.name = 'ParcelType_1'
    self.parcel_type.save()
    self.assertEqual(ParcelType.objects.get(id=self.parcel_type.id).name, 'ParcelType_1')

PropertyTestCase

Bases: TestCase

Property test case.

test_create_property

test_create_property()

Test creating property

Source code in django_project/property/tests/test_property_models.py

def test_create_property(self):
    """Test creating property """
    self.assertTrue(isinstance(self.property, Property))
    self.assertEqual(Property.objects.count(), 1)
    self.assertEqual(self.property.name, Property.objects.get(id=self.property.id).name)

    self.assertEqual(
        self.property.short_code,
        'WCCALU0001'
    )

test_delete_property

test_delete_property()

Test delete property.

Source code in django_project/property/tests/test_property_models.py

def test_delete_property(self):
    """Test delete property."""
    self.property.delete()
    self.assertEqual(Property.objects.count(), 0)

test_update_property

test_update_property()

Test update property.

Source code in django_project/property/tests/test_property_models.py

def test_update_property(self):
    """Test update property."""
    self.property.name = 'Rex Mundi'
    self.property.save()
    self.assertEqual(Property.objects.get(id=self.property.id).name, 'Rex Mundi')

    self.assertEqual(
        self.property.short_code,
        'WCCARM0002'
    )

PropertyTypeTest

Bases: TestCase

Propert type test case

test_create_property_type

test_create_property_type()

Test creating a new property type

Source code in django_project/property/tests/test_property_models.py

def test_create_property_type(self):
    """Test creating a new property type"""
    self.assertTrue(
        isinstance(self.property_type, PropertyType)
    )
    self.assertEqual(PropertyType.objects.count(), 1)
    self.assertEqual(self.property_type.name, PropertyType.objects.get(id=self.property_type.id).name)

test_delete_property_type

test_delete_property_type()

Test deleting a property type

Source code in django_project/property/tests/test_property_models.py

def test_delete_property_type(self):
    """Test deleting a property type"""
    self.property_type.delete()
    self.assertEqual(PropertyType.objects.count(), 0)

test_property_type_unique_name_constraint

test_property_type_unique_name_constraint()

Test property type unique name constraint

Source code in django_project/property/tests/test_property_models.py

def test_property_type_unique_name_constraint(self):
    """Test property type unique name constraint"""
    with self.assertRaises(Exception) as raised:
        PropertyTypeFactory(name='PropertyType 2')
        self.assertEqual(IntegrityError, type(raised.exception))

test_update_property_type

test_update_property_type()

Test updating a property type

Source code in django_project/property/tests/test_property_models.py

def test_update_property_type(self):
    """Test updating a property type"""
    self.property_type.name = 'PropertyType 2'
    self.property_type.save()
    self.assertEqual(
        PropertyType.objects.get(id=self.property_type.id).name,
        'PropertyType 2',
    )

ProvinceTestCase

Bases: TestCase

Province test case

test_create_province

test_create_province()

Test create a province.

Source code in django_project/property/tests/test_property_models.py

def test_create_province(self):
    'Test create a province.'
    self.assertTrue(isinstance(self.province, Province))
    self.assertEqual(Province.objects.count(), 1)
    self.assertEqual(self.province.name, Province.objects.get(id=self.province.id).name)

test_delete_province

test_delete_province()

Test delete a province.

Source code in django_project/property/tests/test_property_models.py

def test_delete_province(self):
    'Test delete a province.'
    self.province.delete()
    self.assertEqual(Province.objects.count(), 0)

test_unique_province_name_constraint

test_unique_province_name_constraint()

Test unique province name constraint.

Source code in django_project/property/tests/test_property_models.py

def test_unique_province_name_constraint(self):
    'Test unique province name constraint.'
    with self.assertRaises(Exception) as raised:
        ProvinceFactory(name='Province 2')
        self.assertEqual(IntegrityError, type(raised.exception))

test_update_province

test_update_province()

Test update a province.

Source code in django_project/property/tests/test_property_models.py

def test_update_province(self):
    'Test update a province.'
    self.province.name = 'Province 2'
    self.province.save()
    self.assertEqual(
        Province.objects.get(id=self.province.id).name, 'Province 2'
    )

Tasks

Spatial Data

columns_and_srid

columns_and_srid(table_name)

Retrieve all column names along with their SRID (if applicable) for a given table in the database schema "layer".

Parameters:

Name	Type	Description	Default
table_name	AnyStr	Table name for which column information is required.	required

Returns:

Type	Description
Tuple[List[ColumnInfo], AnyStr]	Tuple[list[ColumnInfo], str]: A tuple containing two elements: 1. A list of ColumnInfo objects detailing each column. 2. A string representing the SRID of a geometry column, if present; returns an empty string if no geometry column is found.

Source code in django_project/property/spatial_data.py

def columns_and_srid(table_name: AnyStr) -> Tuple[List[ColumnInfo], AnyStr]:
    """
    Retrieve all column names along with their SRID (if applicable)
    for a given table in the database schema "layer".

    :param table_name: Table name for which column information is required.

    :return: Tuple[list[ColumnInfo], str]: A tuple containing two elements:
            1. A list of ColumnInfo objects detailing each column.
            2. A string representing the SRID of a geometry column, if present;
               returns an empty string if no geometry column is found.
    """
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                col.column_name,
                col.data_type,
                COALESCE(geom.srid::text, %s) AS srid
            FROM information_schema.columns col
            LEFT JOIN geometry_columns geom
                ON col.table_schema = geom.f_table_schema
                AND col.table_name = geom.f_table_name
                AND col.column_name = geom.f_geometry_column
            WHERE col.table_schema = %s
            AND col.table_name = %s;
        """, [NO_VALUE, TABLE_SCHEMA, table_name])

        rows = cursor.fetchall()

        columns = [ColumnInfo(
            row[0],
            row[1] if row[2] == NO_VALUE else GEOMETRY_KEY,
            row[2]
        )
            for row in rows
        ]
        srid = ''
        for column in columns:
            if column.srid != NO_VALUE:
                srid = column.srid

        if srid == "0":  # could not get srid information from table
            srids = get_distinct_srids(table_name)
            if len(srids) > 0:
                srid = srids[0]
            else:
                srid = ''

        return columns, srid

extract_spatial_data_from_property_and_layer

extract_spatial_data_from_property_and_layer(
    target_property, context_layer
)

Intersect a target property with a given context layer to extract spatial data.

Parameters:

Name	Type	Description	Default
target_property	Property	The property object that needs to be intersected.	required
context_layer	ContextLayer	The layer that provides contextual spatial data.	required

Returns:

Type	Description
Dict	The spatial data extracted from the intersection of the target property and context layer. Returns an empty dictionary if no spatial data is found.

Source code in django_project/property/spatial_data.py

def extract_spatial_data_from_property_and_layer(
        target_property: Property,
        context_layer: ContextLayer
) -> Dict:
    """
    Intersect a target property with a given context layer to
    extract spatial data.

    :param target_property: The property object that needs to be intersected.
    :type target_property: Property

    :param context_layer: The layer that provides contextual spatial data.
    :type context_layer: ContextLayer

    :return: The spatial data extracted from the intersection of the target
    property and context layer.
    Returns an empty dictionary if no spatial data is found.
    """
    if not target_property.geometry:
        return {}

    spatial_data = {}

    for layer in (
            context_layer.layer_set.filter(
                spatial_filter_field__isnull=False
            ).exclude(spatial_filter_field='')):
        layer_name = layer.name
        columns, srid = columns_and_srid(layer_name)

        if not srid:  # Table with geometry not found, continue
            continue

        column_names = [
            col.column_name for col in columns if col.srid == NO_VALUE
        ]

        query = f"""
            SELECT {'e.' + ',e.'.join(column_names)}
            FROM layer.{layer_name} e
                JOIN public.property p ON ST_Intersects(
                e.geom, ST_Transform(p.geometry, {int(srid)}))
            WHERE p.id = %s;
        """

        try:
            with connection.cursor() as cursor:
                cursor.execute(query, [target_property.id])
                rows = cursor.fetchall()

                spatial_data_values = []

            for row in rows:
                row_dict = dict(zip(column_names, row))
                spatial_data_values.append(row_dict)

                spatial_data[layer_name] = spatial_data_values
        except InternalError as e:
            print(e)

    return spatial_data

get_distinct_srids

get_distinct_srids(table_name)

Retrieve distinct SRIDs from the specified table's geometry column.

Parameters:

Name	Type	Description	Default
table_name	str	Name of the table to check.	required

Returns:

Type	Description
List	A list of distinct SRIDs.

Source code in django_project/property/spatial_data.py

def get_distinct_srids(table_name: str) -> List:
    """
    Retrieve distinct SRIDs from the specified table's geometry column.

    :param table_name: Name of the table to check.

    :return: A list of distinct SRIDs.
    """
    query = (
        f"SELECT DISTINCT ST_SRID(geom) AS srid "
        f"FROM {TABLE_SCHEMA}.{table_name};"
    )

    with connection.cursor() as cursor:
        cursor.execute(query)
        rows = cursor.fetchall()

    return [row[0] for row in rows if row[0] is not None]

save_spatial_values_from_property_layers

save_spatial_values_from_property_layers(target_property)

Extract spatial data from the given property for all context layers and save the extracted values to SpatialDataModel.

Parameters:

Name	Type	Description	Default
target_property	Property	The property object to be extracted	required

Source code in django_project/property/spatial_data.py

def save_spatial_values_from_property_layers(target_property: Property):
    """
    Extract spatial data from the given property for all context layers
    and save the extracted values to SpatialDataModel.

    :param target_property: The property object to be extracted
    :type target_property: Property
    """
    context_layers = ContextLayer.objects.filter(
        layer__spatial_filter_field__isnull=False
    ).distinct()

    layers = {layer.name: layer for layer in Layer.objects.all()}

    for context_layer in context_layers:
        spatial_data_by_layers = extract_spatial_data_from_property_and_layer(
            target_property,
            context_layer
        )
        if not spatial_data_by_layers:
            continue

        spatial_data_obj, _ = SpatialDataModel.objects.get_or_create(
            property=target_property,
            context_layer=context_layer
        )

        for layer_name, spatial_layer_data in spatial_data_by_layers.items():
            layer = layers.get(layer_name)
            if layer is None:
                continue

            for spatial_layer_value in spatial_layer_data:
                filter_field = layer.spatial_filter_field
                if filter_field not in spatial_layer_value:
                    continue
                SpatialDataValueModel.objects.update_or_create(
                    layer=layer,
                    spatial_data=spatial_data_obj,
                    context_layer_value=(
                        spatial_layer_value[filter_field]
                    )
                )